14995486503

Committed 13 May 2025 11:30AM UTC coverage: 99.195% (-0.8%) from 100.0%

Build # 14995486503

Build Type

Pull #119

github

Committed by

web-flow

Commit Message

Merge e47966ba1 into 716f96d7a

Pull Request Pull Request #119: Add non-blocking hasMessage() method to bypass stream_set_timeout() limitations

Coverage Stats

0 of 10 new or added lines in 2 files covered. (0.0%)

1232 of 1242 relevant lines covered (99.19%)

22.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.47

/src/Client.php

<?php

/**
 * Copyright (C) 2014-2025 Textalk and contributors.
 * This file is part of Websocket PHP and is free software under the ISC License.
 */

namespace WebSocket;

use InvalidArgumentException;
use Phrity\Net\{
    Context,
    StreamCollection,
    StreamFactory,
    Uri
};
use Psr\Http\Message\{
    RequestInterface,
    ResponseInterface,
    UriInterface,
};
use Psr\Log\{
    LoggerAwareInterface,
    LoggerInterface,
    NullLogger
};
use Stringable;
use Throwable;
use WebSocket\Exception\{
    BadUriException,
    ClientException,
    ConnectionLevelInterface,
    ExceptionInterface,
    HandshakeException,
    MessageLevelInterface,
    ReconnectException
};
use WebSocket\Http\{
    Request,
    Response
};
use WebSocket\Message\Message;
use WebSocket\Middleware\MiddlewareInterface;
use WebSocket\Trait\{
    ListenerTrait,
    SendMethodsTrait,
    StringableTrait
};

/**
 * WebSocket\Client class.
 * Entry class for WebSocket client.
 */
class Client implements LoggerAwareInterface, Stringable
{
    /** @use ListenerTrait<Client> */
    use ListenerTrait;
    use SendMethodsTrait;
    use StringableTrait;

    // Settings
    private LoggerInterface $logger;
    /** @var int<0, max> $timeout */
    private int $timeout = 60;
    /** @var int<1, max> $frameSize */
    private int $frameSize = 4096;
    private bool $persistent = false;
    private Context $context;
    /** @var array<string, mixed> $headers */
    private array $headers = [];

    // Internal resources
    private StreamFactory $streamFactory;
    private Uri $socketUri;
    private Connection|null $connection = null;
    /** @var array<MiddlewareInterface> $middlewares */
    private array $middlewares = [];
    private StreamCollection|null $streams = null;
    private bool $running = false;


    /* ---------- Magic methods ------------------------------------------------------------------------------------ */

    /**
     * @param UriInterface|string $uri A ws/wss-URI
     */
    public function __construct(UriInterface|string $uri)
    {
        $this->socketUri = $this->parseUri($uri);
        $this->logger = new NullLogger();
        $this->context = new Context();
        $this->setStreamFactory(new StreamFactory());
    }

    /**
     * Get string representation of instance.
     * @return string String representation
     */
    public function __toString(): string
    {
        return $this->stringable('%s', $this->connection ? $this->socketUri->__toString() : 'closed');
    }


    /* ---------- Configuration ------------------------------------------------------------------------------------ */

    /**
     * Set stream factory to use.
     * @param StreamFactory $streamFactory
     * @return self
     */
    public function setStreamFactory(StreamFactory $streamFactory): self
    {
        $this->streamFactory = $streamFactory;
        return $this;
    }

    /**
     * Set logger.
     * @param LoggerInterface $logger Logger implementation
     */
    public function setLogger(LoggerInterface $logger): void
    {
        $this->logger = $logger;
        if ($this->connection) {
            $this->connection->setLogger($this->logger);
        }
    }

    /**
     * Set timeout.
     * @param int<0, max> $timeout Timeout in seconds
     * @return self
     * @throws InvalidArgumentException If invalid timeout provided
     */
    public function setTimeout(int $timeout): self
    {
        if ($timeout < 0) {
            throw new InvalidArgumentException("Invalid timeout '{$timeout}' provided");
        }
        $this->timeout = $timeout;
        if ($this->connection) {
            $this->connection->setTimeout($timeout);
        }
        return $this;
    }

    /**
     * Get timeout.
     * @return int Timeout in seconds
     */
    public function getTimeout(): int
    {
        return $this->timeout;
    }

    /**
     * Set frame size.
     * @param int<1, max> $frameSize Max frame payload size in bytes
     * @return self
     * @throws InvalidArgumentException If invalid frameSize provided
     */
    public function setFrameSize(int $frameSize): self
    {
        if ($frameSize < 1) {
            throw new InvalidArgumentException("Invalid frameSize '{$frameSize}' provided");
        }
        $this->frameSize = $frameSize;
        if ($this->connection) {
            $this->connection->setFrameSize($frameSize);
        }
        return $this;
    }

    /**
     * Get frame size.
     * @return int Frame size in bytes
     */
    public function getFrameSize(): int
    {
        return $this->frameSize;
    }

    /**
     * Set connection persistence.
     * @param bool $persistent True for persistent connection.
     * @return self
     */
    public function setPersistent(bool $persistent): self
    {
        $this->persistent = $persistent;
        return $this;
    }

    /**
     * Set stream context.
     * @param Context|array<string, mixed> $context Context or options as array
     * @see https://www.php.net/manual/en/context.php
     * @return self
     */
    public function setContext(Context|array $context): self
    {
        if ($context instanceof Context) {
            $this->context = $context;
        } else {
            $this->context->setOptions($context);
        }
        return $this;
    }

    /**
     * Get current stream context.
     * @return Context
     */
    public function getContext(): Context
    {
        return $this->context;
    }

    /**
     * Add header for handshake.
     * @param string $name Header name
     * @param string $content Header content
     * @return self
     */
    public function addHeader(string $name, string $content): self
    {
        $this->headers[$name] = $content;
        return $this;
    }

    /**
     * Add a middleware.
     * @param MiddlewareInterface $middleware
     * @return self
     */
    public function addMiddleware(MiddlewareInterface $middleware): self
    {
        $this->middlewares[] = $middleware;
        if ($this->connection) {
            $this->connection->addMiddleware($middleware);
        }
        return $this;
    }


    /* ---------- Messaging operations ----------------------------------------------------------------------------- */

    /**
     * Send message.
     * @template T of Message
     * @param T $message
     * @return T
     */
    public function send(Message $message): Message
    {
        if (!$this->isConnected()) {
            $this->connect();
        }
        return $this->connection->pushMessage($message);
    }

    /**
     * Receive message.
     * Note that this operation will block reading.
     * @return Message
     */
    public function receive(): Message
    {
        if (!$this->isConnected()) {
            $this->connect();
        }
        return $this->connection->pullMessage();
    }


    /* ---------- Listener operations ------------------------------------------------------------------------------ */

    /**
     * Start client listener.
     * @throws Throwable On low level error
     */
    public function start(): void
    {
        // Check if running
        if ($this->running) {
            $this->logger->warning("[client] Client is already running");
            return;
        }
        $this->running = true;
        $this->logger->info("[client] Client is running");

        if (!$this->isConnected()) {
            $this->connect();
        }

        // Run handler
        while ($this->running) {
            try {
                // Get streams with readable content
                $readables = $this->streams->waitRead($this->timeout);
                foreach ($readables as $key => $readable) {
                    try {
                        // Read from connection
                        $message = $this->connection->pullMessage();
                        $this->dispatch($message->getOpcode(), [$this, $this->connection, $message]);
                    } catch (MessageLevelInterface $e) {
                        // Error, but keep connection open
                        $this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);
                        $this->dispatch('error', [$this, $this->connection, $e]);
                    } catch (ConnectionLevelInterface $e) {
                        // Error, disconnect connection
                        $this->disconnect();
                        $this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);
                        $this->dispatch('error', [$this, $this->connection, $e]);
                    }
                }
                if (!$this->connection->isConnected()) {
                    $this->running = false;
                }
                $this->connection->tick();
                $this->dispatch('tick', [$this]);
            } catch (ExceptionInterface $e) {
                $this->disconnect();
                $this->running = false;

                // Low-level error
                $this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);
                $this->dispatch('error', [$this, null, $e]);
            } catch (Throwable $e) {
                $this->disconnect();
                $this->running = false;

                // Crash it
                $this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);
                throw $e;
            }
            gc_collect_cycles(); // Collect garbage
        }
    }

    /**
     * Stop client listener (resumable).
     */
    public function stop(): void
    {
        $this->running = false;
        $this->logger->info("[client] Client is stopped");
    }

    /**
     * If client is running (accepting messages).
     * @return bool
     */
    public function isRunning(): bool
    {
        return $this->running;
    }


    /* ---------- Connection management ---------------------------------------------------------------------------- */

    /**
     * If Client has active connection.
     * @return bool True if active connection.
     */
    public function isConnected(): bool
    {
        return $this->connection && $this->connection->isConnected();
    }

    /**
     * If Client is readable.
     * @return bool
     */
    public function isReadable(): bool
    {
        return $this->connection && $this->connection->isReadable();
    }

    /**
     * If Client is writable.
     * @return bool
     */
    public function isWritable(): bool
    {
        return $this->connection && $this->connection->isWritable();
    }


    /**
     * Connect to server and perform upgrade.
     * @throws ClientException On failed connection
     */
    public function connect(): void
    {
        $this->disconnect();
        $this->streams = $this->streamFactory->createStreamCollection();

        $host_uri = (new Uri())
            ->withScheme(match ($this->socketUri->getScheme()) {
                'ws', 'http' => 'tcp',
                'wss', 'https' => 'ssl',
                default => throw new ClientException("Invalid socket scheme: {$this->socketUri->getScheme()}")
            })
            ->withHost($this->socketUri->getHost(Uri::IDN_ENCODE))
            ->withPort($this->socketUri->getPort(Uri::REQUIRE_PORT));

        $stream = null;

        try {
            $client = $this->streamFactory->createSocketClient($host_uri, $this->context);
            $client->setPersistent($this->persistent);
            $client->setTimeout($this->timeout);
            $stream = $client->connect();
        } catch (Throwable $e) {
            $error = "Could not open socket to \"{$host_uri}\": {$e->getMessage()}";
            $this->logger->error("[client] {$error}", ['exception' => $e]);
            throw new ClientException($error);
        }
        $name = $stream->getRemoteName();
        $this->streams->attach($stream, $name);
        $this->connection = new Connection($stream, true, false, $host_uri->getScheme() === 'ssl');
        $this->connection->setFrameSize($this->frameSize);
        $this->connection->setTimeout($this->timeout);
        $this->connection->setLogger($this->logger);
        foreach ($this->middlewares as $middleware) {
            $this->connection->addMiddleware($middleware);
        }

        if (!$this->isConnected()) {
            $error = "Invalid stream on \"{$host_uri}\".";
            $this->logger->error("[client] {$error}");
            throw new ClientException($error);
        }
        try {
            if (!$this->persistent || $stream->tell() == 0) {
                $response = $this->performHandshake($this->socketUri);
            }
        } catch (ReconnectException $e) {
            $this->logger->info("[client] {$e->getMessage()}", ['exception' => $e]);
            if ($uri = $e->getUri()) {
                $this->socketUri = $uri;
            }
            $this->connect();
            return;
        }
        $this->logger->info("[client] Client connected to {$this->socketUri}");
        $this->dispatch('handshake', [
            $this,
            $this->connection,
            $this->connection->getHandshakeRequest(),
            $this->connection->getHandshakeResponse(),
        ]);
        $this->dispatch('connect', [$this, $this->connection, $this->connection->getHandshakeResponse()]);
    }

    /**
     * Disconnect from server.
     */
    public function disconnect(): void
    {
        if ($this->isConnected()) {
            $this->connection->disconnect();
            $this->logger->info('[client] Client disconnected');
            $this->dispatch('disconnect', [$this, $this->connection]);
        }
    }


    /* ---------- Connection wrapper methods ----------------------------------------------------------------------- */

    /**
     * Get name of local socket, or null if not connected.
     * @return string|null
     */
    public function getName(): string|null
    {
        return $this->isConnected() ? $this->connection->getName() : null;
    }

    /**
     * Get name of remote socket, or null if not connected.
     * @return string|null
     */
    public function getRemoteName(): string|null
    {
        return $this->isConnected() ? $this->connection->getRemoteName() : null;
    }

    /**
     * Get meta value on connection.
     * @param string $key Meta key
     * @return mixed Meta value
     */
    public function getMeta(string $key): mixed
    {
        return $this->isConnected() ? $this->connection->getMeta($key) : null;
    }

    /**
     * Get Response for handshake procedure.
     * @return ResponseInterface|null Handshake.
     */
    public function getHandshakeResponse(): ResponseInterface|null
    {
        return $this->connection ? $this->connection->getHandshakeResponse() : null;
    }

    /**
     * Check if there are messages available to read without blocking.
     * @return bool True if there are messages available to read, false otherwise.
     */
    public function hasMessages(): bool
    {
        return $this->connection ? $this->connection->hasMessages() : false;
    }


    /* ---------- Internal helper methods -------------------------------------------------------------------------- */

    /**
     * Perform upgrade handshake on new connections.
     * @throws HandshakeException On failed handshake
     * @throws ReconnectException On reconnect/redirect requirement
     */
    protected function performHandshake(Uri $uri): ResponseInterface
    {
        // Generate the WebSocket key.
        $key = $this->generateKey();

        $request = new Request('GET', $uri);

        $request = $request
            ->withHeader('User-Agent', 'websocket-client-php')
            ->withHeader('Connection', 'Upgrade')
            ->withHeader('Upgrade', 'websocket')
            ->withHeader('Sec-WebSocket-Key', $key)
            ->withHeader('Sec-WebSocket-Version', '13');

        // Handle basic authentication.
        if ($userinfo = $uri->getUserInfo(Uri::URI_DECODE)) {
            $request = $request->withHeader('Authorization', 'Basic ' . base64_encode($userinfo));
        }

        // Add and override with headers.
        foreach ($this->headers as $name => $content) {
            $request = $request->withHeader($name, $content);
        }

        try {
            /** @var RequestInterface */
            $request = $this->connection->pushHttp($request);
            /** @var ResponseInterface */
            $response = $this->connection->pullHttp();

            if ($response->getStatusCode() != 101) {
                throw new HandshakeException("Invalid status code {$response->getStatusCode()}.", $response);
            }

            if (empty($response->getHeaderLine('Sec-WebSocket-Accept'))) {
                throw new HandshakeException(
                    "Connection to '{$uri}' failed: Server sent invalid upgrade response.",
                    $response
                );
            }

            $response_key = trim($response->getHeaderLine('Sec-WebSocket-Accept'));
            $expected_key = base64_encode(
                pack('H*', sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'))
            );

            if ($response_key !== $expected_key) {
                throw new HandshakeException("Server sent bad upgrade response.", $response);
            }
        } catch (HandshakeException $e) {
            $this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);
            throw $e;
        }

        $this->logger->debug("[client] Handshake on {$uri->getPath()}");
        $this->connection->setHandshakeRequest($request);
        $this->connection->setHandshakeResponse($response);

        return $response;
    }

    /**
     * Generate a random string for WebSocket key.
     * @return string Random string
     */
    protected function generateKey(): string
    {
        $key = '';
        for ($i = 0; $i < 16; $i++) {
            $key .= chr(rand(33, 126));
        }
        return base64_encode($key);
    }

    /**
     * Ensure URI instance to use in client.
     * @param UriInterface|string $uri A ws/wss-URI
     * @return Uri
     * @throws BadUriException On invalid URI
     */
    protected function parseUri(UriInterface|string $uri): Uri
    {
        try {
            if ($uri instanceof Uri) {
                $uri_instance = $uri;
            } elseif ($uri instanceof UriInterface) {
                $uri_instance = new Uri("{$uri}");
            } else {
                $uri_instance = new Uri($uri);
            }
        } catch (InvalidArgumentException $e) {
            throw new BadUriException("Invalid URI '{$uri}' provided.");
        }


        if (!in_array($uri_instance->getScheme(), ['ws', 'wss'])) {
            throw new BadUriException("Invalid URI scheme, must be 'ws' or 'wss'.");
        }
        if (!$uri_instance->getHost()) {
            throw new BadUriException("Invalid URI host.");
        }
        return $uri_instance;
    }
}

1	<?php
2
3	/**
4	* Copyright (C) 2014-2025 Textalk and contributors.
5	* This file is part of Websocket PHP and is free software under the ISC License.
6	*/
7
8	namespace WebSocket;
9
10	use InvalidArgumentException;
11	use Phrity\Net\{
12	Context,
13	StreamCollection,
14	StreamFactory,
15	Uri
16	};
17	use Psr\Http\Message\{
18	RequestInterface,
19	ResponseInterface,
20	UriInterface,
21	};
22	use Psr\Log\{
23	LoggerAwareInterface,
24	LoggerInterface,
25	NullLogger
26	};
27	use Stringable;
28	use Throwable;
29	use WebSocket\Exception\{
30	BadUriException,
31	ClientException,
32	ConnectionLevelInterface,
33	ExceptionInterface,
34	HandshakeException,
35	MessageLevelInterface,
36	ReconnectException
37	};
38	use WebSocket\Http\{
39	Request,
40	Response
41	};
42	use WebSocket\Message\Message;
43	use WebSocket\Middleware\MiddlewareInterface;
44	use WebSocket\Trait\{
45	ListenerTrait,
46	SendMethodsTrait,
47	StringableTrait
48	};
49
50	/**
51	* WebSocket\Client class.
52	* Entry class for WebSocket client.
53	*/
54	class Client implements LoggerAwareInterface, Stringable
55	{
56	/** @use ListenerTrait<Client> */
57	use ListenerTrait;
58	use SendMethodsTrait;
59	use StringableTrait;
60
61	// Settings
62	private LoggerInterface $logger;
63	/** @var int<0, max> $timeout */
64	private int $timeout = 60;
65	/** @var int<1, max> $frameSize */
66	private int $frameSize = 4096;
67	private bool $persistent = false;
68	private Context $context;
69	/** @var array<string, mixed> $headers */
70	private array $headers = [];
71
72	// Internal resources
73	private StreamFactory $streamFactory;
74	private Uri $socketUri;
75	private Connection\|null $connection = null;
76	/** @var array<MiddlewareInterface> $middlewares */
77	private array $middlewares = [];
78	private StreamCollection\|null $streams = null;
79	private bool $running = false;
80
81
82	/* ---------- Magic methods ------------------------------------------------------------------------------------ */
83
84	/**
85	* @param UriInterface\|string $uri A ws/wss-URI
86	*/
87	public function __construct(UriInterface\|string $uri)
88	{
89	$this->socketUri = $this->parseUri($uri);	51✔
90	$this->logger = new NullLogger();	48✔
91	$this->context = new Context();	48✔
92	$this->setStreamFactory(new StreamFactory());	48✔
93	}
94
95	/**
96	* Get string representation of instance.
97	* @return string String representation
98	*/
99	public function __toString(): string
100	{
101	return $this->stringable('%s', $this->connection ? $this->socketUri->__toString() : 'closed');	1✔
102	}
103
104
105	/* ---------- Configuration ------------------------------------------------------------------------------------ */
106
107	/**
108	* Set stream factory to use.
109	* @param StreamFactory $streamFactory
110	* @return self
111	*/
112	public function setStreamFactory(StreamFactory $streamFactory): self
113	{
114	$this->streamFactory = $streamFactory;	48✔
115	return $this;	48✔
116	}
117
118	/**
119	* Set logger.
120	* @param LoggerInterface $logger Logger implementation
121	*/
122	public function setLogger(LoggerInterface $logger): void
123	{
124	$this->logger = $logger;	2✔
125	if ($this->connection) {	2✔
126	$this->connection->setLogger($this->logger);	1✔
127	}
128	}
129
130	/**
131	* Set timeout.
132	* @param int<0, max> $timeout Timeout in seconds
133	* @return self
134	* @throws InvalidArgumentException If invalid timeout provided
135	*/
136	public function setTimeout(int $timeout): self
137	{
138	if ($timeout < 0) {	4✔
139	throw new InvalidArgumentException("Invalid timeout '{$timeout}' provided");	1✔
140	}
141	$this->timeout = $timeout;	3✔
142	if ($this->connection) {	3✔
143	$this->connection->setTimeout($timeout);	1✔
144	}
145	return $this;	3✔
146	}
147
148	/**
149	* Get timeout.
150	* @return int Timeout in seconds
151	*/
152	public function getTimeout(): int
153	{
154	return $this->timeout;	1✔
155	}
156
157	/**
158	* Set frame size.
159	* @param int<1, max> $frameSize Max frame payload size in bytes
160	* @return self
161	* @throws InvalidArgumentException If invalid frameSize provided
162	*/
163	public function setFrameSize(int $frameSize): self
164	{
165	if ($frameSize < 1) {	6✔
166	throw new InvalidArgumentException("Invalid frameSize '{$frameSize}' provided");	1✔
167	}
168	$this->frameSize = $frameSize;	5✔
169	if ($this->connection) {	5✔
170	$this->connection->setFrameSize($frameSize);	2✔
171	}
172	return $this;	5✔
173	}
174
175	/**
176	* Get frame size.
177	* @return int Frame size in bytes
178	*/
179	public function getFrameSize(): int
180	{
181	return $this->frameSize;	7✔
182	}
183
184	/**
185	* Set connection persistence.
186	* @param bool $persistent True for persistent connection.
187	* @return self
188	*/
189	public function setPersistent(bool $persistent): self
190	{
191	$this->persistent = $persistent;	1✔
192	return $this;	1✔
193	}
194
195	/**
196	* Set stream context.
197	* @param Context\|array<string, mixed> $context Context or options as array
198	* @see https://www.php.net/manual/en/context.php
199	* @return self
200	*/
201	public function setContext(Context\|array $context): self
202	{
203	if ($context instanceof Context) {	2✔
204	$this->context = $context;	1✔
205	} else {
206	$this->context->setOptions($context);	1✔
207	}
208	return $this;	2✔
209	}
210
211	/**
212	* Get current stream context.
213	* @return Context
214	*/
215	public function getContext(): Context
216	{
217	return $this->context;	1✔
218	}
219
220	/**
221	* Add header for handshake.
222	* @param string $name Header name
223	* @param string $content Header content
224	* @return self
225	*/
226	public function addHeader(string $name, string $content): self
227	{
228	$this->headers[$name] = $content;	1✔
229	return $this;	1✔
230	}
231
232	/**
233	* Add a middleware.
234	* @param MiddlewareInterface $middleware
235	* @return self
236	*/
237	public function addMiddleware(MiddlewareInterface $middleware): self
238	{
239	$this->middlewares[] = $middleware;	6✔
240	if ($this->connection) {	6✔
241	$this->connection->addMiddleware($middleware);	1✔
242	}
243	return $this;	6✔
244	}
245
246
247	/* ---------- Messaging operations ----------------------------------------------------------------------------- */
248
249	/**
250	* Send message.
251	* @template T of Message
252	* @param T $message
253	* @return T
254	*/
255	public function send(Message $message): Message
256	{
257	if (!$this->isConnected()) {	9✔
258	$this->connect();	1✔
259	}
260	return $this->connection->pushMessage($message);	9✔
261	}
262
263	/**
264	* Receive message.
265	* Note that this operation will block reading.
266	* @return Message
267	*/
268	public function receive(): Message
269	{
270	if (!$this->isConnected()) {	11✔
271	$this->connect();	1✔
272	}
273	return $this->connection->pullMessage();	11✔
274	}
275
276
277	/* ---------- Listener operations ------------------------------------------------------------------------------ */
278
279	/**
280	* Start client listener.
281	* @throws Throwable On low level error
282	*/
283	public function start(): void
284	{
285	// Check if running
286	if ($this->running) {	6✔
287	$this->logger->warning("[client] Client is already running");	1✔
288	return;	1✔
289	}
290	$this->running = true;	6✔
291	$this->logger->info("[client] Client is running");	6✔
292
293	if (!$this->isConnected()) {	6✔
294	$this->connect();	6✔
295	}
296
297	// Run handler
298	while ($this->running) {	6✔
299	try {
300	// Get streams with readable content
301	$readables = $this->streams->waitRead($this->timeout);	5✔
302	foreach ($readables as $key => $readable) {	4✔
303	try {
304	// Read from connection
305	$message = $this->connection->pullMessage();	4✔
306	$this->dispatch($message->getOpcode(), [$this, $this->connection, $message]);	2✔
307	} catch (MessageLevelInterface $e) {	3✔
308	// Error, but keep connection open
309	$this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);	1✔
310	$this->dispatch('error', [$this, $this->connection, $e]);	1✔
311	} catch (ConnectionLevelInterface $e) {	2✔
312	// Error, disconnect connection
313	$this->disconnect();	1✔
314	$this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);	1✔
315	$this->dispatch('error', [$this, $this->connection, $e]);	1✔
316	}
317	}
318	if (!$this->connection->isConnected()) {	3✔
319	$this->running = false;	1✔
320	}
321	$this->connection->tick();	3✔
322	$this->dispatch('tick', [$this]);	3✔
323	} catch (ExceptionInterface $e) {	3✔
324	$this->disconnect();	1✔
325	$this->running = false;	1✔
326
327	// Low-level error
328	$this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);	1✔
329	$this->dispatch('error', [$this, null, $e]);	1✔
330	} catch (Throwable $e) {	2✔
331	$this->disconnect();	2✔
332	$this->running = false;	2✔
333
334	// Crash it
335	$this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);	2✔
336	throw $e;	2✔
337	}
338	gc_collect_cycles(); // Collect garbage	3✔
339	}
340	}
341
342	/**
343	* Stop client listener (resumable).
344	*/
345	public function stop(): void
346	{
347	$this->running = false;	2✔
348	$this->logger->info("[client] Client is stopped");	2✔
349	}
350
351	/**
352	* If client is running (accepting messages).
353	* @return bool
354	*/
355	public function isRunning(): bool
356	{
357	return $this->running;	1✔
358	}
359
360
361	/* ---------- Connection management ---------------------------------------------------------------------------- */
362
363	/**
364	* If Client has active connection.
365	* @return bool True if active connection.
366	*/
367	public function isConnected(): bool
368	{
369	return $this->connection && $this->connection->isConnected();	45✔
370	}
371
372	/**
373	* If Client is readable.
374	* @return bool
375	*/
376	public function isReadable(): bool
377	{
378	return $this->connection && $this->connection->isReadable();	1✔
379	}
380
381	/**
382	* If Client is writable.
383	* @return bool
384	*/
385	public function isWritable(): bool
386	{
387	return $this->connection && $this->connection->isWritable();	1✔
388	}
389
390
391	/**
392	* Connect to server and perform upgrade.
393	* @throws ClientException On failed connection
394	*/
395	public function connect(): void
396	{
397	$this->disconnect();	44✔
398	$this->streams = $this->streamFactory->createStreamCollection();	44✔
399
400	$host_uri = (new Uri())	44✔
401	->withScheme(match ($this->socketUri->getScheme()) {	44✔
402	'ws', 'http' => 'tcp',	40✔
403	'wss', 'https' => 'ssl',	4✔
404	default => throw new ClientException("Invalid socket scheme: {$this->socketUri->getScheme()}")	44✔
405	})	44✔
406	->withHost($this->socketUri->getHost(Uri::IDN_ENCODE))	44✔
407	->withPort($this->socketUri->getPort(Uri::REQUIRE_PORT));	44✔
408
409	$stream = null;	44✔
410
411	try {
412	$client = $this->streamFactory->createSocketClient($host_uri, $this->context);	44✔
413	$client->setPersistent($this->persistent);	44✔
414	$client->setTimeout($this->timeout);	44✔
415	$stream = $client->connect();	44✔
416	} catch (Throwable $e) {	1✔
417	$error = "Could not open socket to \"{$host_uri}\": {$e->getMessage()}";	1✔
418	$this->logger->error("[client] {$error}", ['exception' => $e]);	1✔
419	throw new ClientException($error);	1✔
420	}
421	$name = $stream->getRemoteName();	43✔
422	$this->streams->attach($stream, $name);	43✔
423	$this->connection = new Connection($stream, true, false, $host_uri->getScheme() === 'ssl');	43✔
424	$this->connection->setFrameSize($this->frameSize);	43✔
425	$this->connection->setTimeout($this->timeout);	43✔
426	$this->connection->setLogger($this->logger);	43✔
427	foreach ($this->middlewares as $middleware) {	43✔
428	$this->connection->addMiddleware($middleware);	5✔
429	}
430
431	if (!$this->isConnected()) {	43✔
432	$error = "Invalid stream on \"{$host_uri}\".";	1✔
433	$this->logger->error("[client] {$error}");	1✔
434	throw new ClientException($error);	1✔
435	}
436	try {
437	if (!$this->persistent \|\| $stream->tell() == 0) {	42✔
438	$response = $this->performHandshake($this->socketUri);	42✔
439	}
440	} catch (ReconnectException $e) {	5✔
441	$this->logger->info("[client] {$e->getMessage()}", ['exception' => $e]);	1✔
442	if ($uri = $e->getUri()) {	1✔
443	$this->socketUri = $uri;	1✔
444	}
445	$this->connect();	1✔
446	return;	1✔
447	}
448	$this->logger->info("[client] Client connected to {$this->socketUri}");	38✔
449	$this->dispatch('handshake', [	38✔
450	$this,	38✔
451	$this->connection,	38✔
452	$this->connection->getHandshakeRequest(),	38✔
453	$this->connection->getHandshakeResponse(),	38✔
454	]);	38✔
455	$this->dispatch('connect', [$this, $this->connection, $this->connection->getHandshakeResponse()]);	38✔
456	}
457
458	/**
459	* Disconnect from server.
460	*/
461	public function disconnect(): void
462	{
463	if ($this->isConnected()) {	44✔
464	$this->connection->disconnect();	6✔
465	$this->logger->info('[client] Client disconnected');	6✔
466	$this->dispatch('disconnect', [$this, $this->connection]);	6✔
467	}
468	}
469
470
471	/* ---------- Connection wrapper methods ----------------------------------------------------------------------- */
472
473	/**
474	* Get name of local socket, or null if not connected.
475	* @return string\|null
476	*/
477	public function getName(): string\|null
478	{
479	return $this->isConnected() ? $this->connection->getName() : null;	1✔
480	}
481
482	/**
483	* Get name of remote socket, or null if not connected.
484	* @return string\|null
485	*/
486	public function getRemoteName(): string\|null
487	{
488	return $this->isConnected() ? $this->connection->getRemoteName() : null;	1✔
489	}
490
491	/**
492	* Get meta value on connection.
493	* @param string $key Meta key
494	* @return mixed Meta value
495	*/
496	public function getMeta(string $key): mixed
497	{
498	return $this->isConnected() ? $this->connection->getMeta($key) : null;	1✔
499	}
500
501	/**
502	* Get Response for handshake procedure.
503	* @return ResponseInterface\|null Handshake.
504	*/
505	public function getHandshakeResponse(): ResponseInterface\|null
506	{
507	return $this->connection ? $this->connection->getHandshakeResponse() : null;	3✔
508	}
509
510	/**
511	* Check if there are messages available to read without blocking.
512	* @return bool True if there are messages available to read, false otherwise.
513	*/
514	public function hasMessages(): bool
515	{
NEW 516	return $this->connection ? $this->connection->hasMessages() : false;	×
517	}
518
519
520	/* ---------- Internal helper methods -------------------------------------------------------------------------- */
521
522	/**
523	* Perform upgrade handshake on new connections.
524	* @throws HandshakeException On failed handshake
525	* @throws ReconnectException On reconnect/redirect requirement
526	*/
527	protected function performHandshake(Uri $uri): ResponseInterface
528	{
529	// Generate the WebSocket key.
530	$key = $this->generateKey();	42✔
531
532	$request = new Request('GET', $uri);	42✔
533
534	$request = $request	42✔
535	->withHeader('User-Agent', 'websocket-client-php')	42✔
536	->withHeader('Connection', 'Upgrade')	42✔
537	->withHeader('Upgrade', 'websocket')	42✔
538	->withHeader('Sec-WebSocket-Key', $key)	42✔
539	->withHeader('Sec-WebSocket-Version', '13');	42✔
540
541	// Handle basic authentication.
542	if ($userinfo = $uri->getUserInfo(Uri::URI_DECODE)) {	42✔
543	$request = $request->withHeader('Authorization', 'Basic ' . base64_encode($userinfo));	3✔
544	}
545
546	// Add and override with headers.
547	foreach ($this->headers as $name => $content) {	42✔
548	$request = $request->withHeader($name, $content);	1✔
549	}
550
551	try {
552	/** @var RequestInterface */
553	$request = $this->connection->pushHttp($request);	42✔
554	/** @var ResponseInterface */
555	$response = $this->connection->pullHttp();	42✔
556
557	if ($response->getStatusCode() != 101) {	41✔
558	throw new HandshakeException("Invalid status code {$response->getStatusCode()}.", $response);	1✔
559	}
560
561	if (empty($response->getHeaderLine('Sec-WebSocket-Accept'))) {	40✔
562	throw new HandshakeException(	1✔
563	"Connection to '{$uri}' failed: Server sent invalid upgrade response.",	1✔
564	$response	1✔
565	);	1✔
566	}
567
568	$response_key = trim($response->getHeaderLine('Sec-WebSocket-Accept'));	39✔
569	$expected_key = base64_encode(	39✔
570	pack('H*', sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'))	39✔
571	);	39✔
572
573	if ($response_key !== $expected_key) {	39✔
574	throw new HandshakeException("Server sent bad upgrade response.", $response);	39✔
575	}
576	} catch (HandshakeException $e) {	5✔
577	$this->logger->error("[client] {$e->getMessage()}", ['exception' => $e]);	3✔
578	throw $e;	3✔
579	}
580
581	$this->logger->debug("[client] Handshake on {$uri->getPath()}");	38✔
582	$this->connection->setHandshakeRequest($request);	38✔
583	$this->connection->setHandshakeResponse($response);	38✔
584
585	return $response;	38✔
586	}
587
588	/**
589	* Generate a random string for WebSocket key.
590	* @return string Random string
591	*/
592	protected function generateKey(): string
593	{
594	$key = '';	42✔
595	for ($i = 0; $i < 16; $i++) {	42✔
596	$key .= chr(rand(33, 126));	42✔
597	}
598	return base64_encode($key);	42✔
599	}
600
601	/**
602	* Ensure URI instance to use in client.
603	* @param UriInterface\|string $uri A ws/wss-URI
604	* @return Uri
605	* @throws BadUriException On invalid URI
606	*/
607	protected function parseUri(UriInterface\|string $uri): Uri
608	{
609	try {
610	if ($uri instanceof Uri) {	51✔
611	$uri_instance = $uri;	3✔
612	} elseif ($uri instanceof UriInterface) {	48✔
613	$uri_instance = new Uri("{$uri}");	1✔
614	} else {
615	$uri_instance = new Uri($uri);	51✔
616	}
617	} catch (InvalidArgumentException $e) {	1✔
618	throw new BadUriException("Invalid URI '{$uri}' provided.");	1✔
619	}
620
621
622	if (!in_array($uri_instance->getScheme(), ['ws', 'wss'])) {	50✔
623	throw new BadUriException("Invalid URI scheme, must be 'ws' or 'wss'.");	1✔
624	}
625	if (!$uri_instance->getHost()) {	49✔
626	throw new BadUriException("Invalid URI host.");	1✔
627	}
628	return $uri_instance;	48✔
629	}
630	}

sirn-se / websocket-php / 14995486503

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous