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

90.43

/src/Connection.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,
    SocketStream,
};
use Psr\Http\Message\{
    MessageInterface,
    RequestInterface,
    ResponseInterface,
};
use Psr\Log\{
    LoggerAwareInterface,
    LoggerInterface,
    NullLogger
};
use Stringable;
use Throwable;
use WebSocket\Frame\FrameHandler;
use WebSocket\Http\HttpHandler;
use WebSocket\Exception\{
    ConnectionClosedException,
    ConnectionFailureException,
    ConnectionTimeoutException,
    Exception,
    ReconnectException,
};
use WebSocket\Message\{
    Message,
    MessageHandler
};
use WebSocket\Middleware\{
    MiddlewareHandler,
    MiddlewareInterface
};
use WebSocket\Trait\{
    SendMethodsTrait,
    StringableTrait
};

/**
 * WebSocket\Connection class.
 * A client/server connection, wrapping socket stream.
 */
class Connection implements LoggerAwareInterface, Stringable
{
    use SendMethodsTrait;
    use StringableTrait;

    private SocketStream $stream;
    private HttpHandler $httpHandler;
    private MessageHandler $messageHandler;
    private MiddlewareHandler $middlewareHandler;
    private LoggerInterface $logger;
    /** @var int<1, max> $frameSize */
    private int $frameSize = 4096;
    /** @var int<0, max> $timeout */
    private int $timeout = 60;
    private string $localName;
    private string $remoteName;
    private RequestInterface|null $handshakeRequest = null;
    private ResponseInterface|null $handshakeResponse = null;
    /** @var array<string, mixed> $meta */
    private array $meta = [];
    private bool $closed = false;


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

    public function __construct(SocketStream $stream, bool $pushMasked, bool $pullMaskedRequired, bool $ssl = false)
    {
        $this->stream = $stream;
        $this->httpHandler = new HttpHandler($this->stream, $ssl);
        $this->messageHandler = new MessageHandler(new FrameHandler($this->stream, $pushMasked, $pullMaskedRequired));
        $this->middlewareHandler = new MiddlewareHandler($this->messageHandler, $this->httpHandler);
        $this->setLogger(new NullLogger());
        $this->localName = $this->stream->getLocalName();
        $this->remoteName = $this->stream->getRemoteName();
    }

    public function __destruct()
    {
        if (!$this->closed && $this->isConnected()) {
            $this->stream->close();
        }
    }

    public function __toString(): string
    {
        return $this->stringable('%s:%s', $this->localName, $this->remoteName);
    }


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

    /**
     * Set logger.
     * @param LoggerInterface $logger Logger implementation
     */
    public function setLogger(LoggerInterface $logger): void
    {
        $this->logger = $logger;
        $this->httpHandler->setLogger($logger);
        $this->messageHandler->setLogger($logger);
        $this->middlewareHandler->setLogger($logger);
        $this->logger->debug("[connection] Setting logger: " . get_class($logger));
    }

    /**
     * Set time out on connection.
     * @param int<0, max> $timeout Timeout part in seconds
     * @return self
     */
    public function setTimeout(int $timeout): self
    {
        if ($timeout < 0) {
            throw new InvalidArgumentException("Invalid timeout '{$timeout}' provided");
        }
        $this->timeout = $timeout;
        $this->stream->setTimeout($timeout, 0);
        $this->logger->debug("[connection] Setting timeout: {$timeout} seconds");
        return $this;
    }

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

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

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

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

    /**
     * Add a middleware.
     * @param MiddlewareInterface $middleware
     * @return self
     */
    public function addMiddleware(MiddlewareInterface $middleware): self
    {
        $this->middlewareHandler->add($middleware);
        $this->logger->debug("[connection] Added middleware: {$middleware}");
        return $this;
    }


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

    /**
     * If connected to stream.
     * @return bool
     */
    public function isConnected(): bool
    {
        return $this->stream->isConnected();
    }

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

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

    /**
     * Close connection stream.
     * @return self
     */
    public function disconnect(): self
    {
        $this->logger->info('[connection] Closing connection');
        $this->stream->close();
        $this->closed = true;
        return $this;
    }

    /**
     * Close connection stream reading.
     * @return self
     */
    public function closeRead(): self
    {
        $this->logger->info('[connection] Closing further reading');
        $this->stream->closeRead();
        return $this;
    }

    /**
     * Close connection stream writing.
     * @return self
     */
    public function closeWrite(): self
    {
        $this->logger->info('[connection] Closing further writing');
        $this->stream->closeWrite();
        return $this;
    }

    /**
     * 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
    {
        if (!$this->isConnected()) {
            return false;
        }

        $resource = $this->stream->getResource();
        if (!$resource) {
            return false;
        }

        $read = [$resource];
        $write = null;
        $except = null;
        
        return @stream_select($read, $write, $except, 0, 0) > 0;
    }


    /* ---------- Connection state --------------------------------------------------------------------------------- */

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

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

    /**
     * Set meta value on connection.
     * @param string $key Meta key
     * @param mixed $value Meta value
     */
    public function setMeta(string $key, mixed $value): void
    {
        $this->meta[$key] = $value;
    }

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

    /**
     * Tick operation on connection.
     */
    public function tick(): void
    {
        $this->middlewareHandler->processTick($this);
    }


    /* ---------- WebSocket Message methods ------------------------------------------------------------------------ */

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

    /**
     * Push a message to stream.
     * @template T of Message
     * @param T $message
     * @return T
     */
    public function pushMessage(Message $message): Message
    {
        try {
            return $this->middlewareHandler->processOutgoing($this, $message);
        } catch (Throwable $e) {
            $this->throwException($e);
        }
    }

    // Pull a message from stream
    public function pullMessage(): Message
    {
        try {
            return $this->middlewareHandler->processIncoming($this);
        } catch (Throwable $e) {
            $this->throwException($e);
        }
    }


    /* ---------- HTTP Message methods ----------------------------------------------------------------------------- */

    public function pushHttp(MessageInterface $message): MessageInterface
    {
        try {
            return $this->middlewareHandler->processHttpOutgoing($this, $message);
        } catch (Throwable $e) {
            $this->throwException($e);
        }
    }

    public function pullHttp(): MessageInterface
    {
        try {
            return $this->middlewareHandler->processHttpIncoming($this);
        } catch (Throwable $e) {
            $this->throwException($e);
        }
    }

    public function setHandshakeRequest(RequestInterface $request): self
    {
        $this->handshakeRequest = $request;
        return $this;
    }

    public function getHandshakeRequest(): RequestInterface|null
    {
        return $this->handshakeRequest;
    }

    public function setHandshakeResponse(ResponseInterface $response): self
    {
        $this->handshakeResponse = $response;
        return $this;
    }

    public function getHandshakeResponse(): ResponseInterface|null
    {
        return $this->handshakeResponse;
    }


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

    protected function throwException(Throwable $e): never
    {
        // Internal exceptions are handled and re-thrown
        if ($e instanceof ReconnectException) {
            $this->logger->info("[connection] {$e->getMessage()}", ['exception' => $e]);
            throw $e;
        }
        if ($e instanceof Exception) {
            $this->logger->error("[connection] {$e->getMessage()}", ['exception' => $e]);
            throw $e;
        }
        // External exceptions are converted to internal
        if ($this->isConnected()) {
            $meta = $this->stream->getMetadata();
            $json = json_encode($meta);
            if (!empty($meta['timed_out'])) {
                $this->logger->error("[connection] {$e->getMessage()}", ['exception' => $e, 'meta' => $meta]);
                throw new ConnectionTimeoutException();
            }
            if (!empty($meta['eof'])) {
                $this->logger->error("[connection] {$e->getMessage()}", ['exception' => $e, 'meta' => $meta]);
                throw new ConnectionClosedException();
            }
        }
        $this->logger->error("[connection] {$e->getMessage()}", ['exception' => $e]);
        throw new ConnectionFailureException();
    }
}

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	SocketStream,
14	};
15	use Psr\Http\Message\{
16	MessageInterface,
17	RequestInterface,
18	ResponseInterface,
19	};
20	use Psr\Log\{
21	LoggerAwareInterface,
22	LoggerInterface,
23	NullLogger
24	};
25	use Stringable;
26	use Throwable;
27	use WebSocket\Frame\FrameHandler;
28	use WebSocket\Http\HttpHandler;
29	use WebSocket\Exception\{
30	ConnectionClosedException,
31	ConnectionFailureException,
32	ConnectionTimeoutException,
33	Exception,
34	ReconnectException,
35	};
36	use WebSocket\Message\{
37	Message,
38	MessageHandler
39	};
40	use WebSocket\Middleware\{
41	MiddlewareHandler,
42	MiddlewareInterface
43	};
44	use WebSocket\Trait\{
45	SendMethodsTrait,
46	StringableTrait
47	};
48
49	/**
50	* WebSocket\Connection class.
51	* A client/server connection, wrapping socket stream.
52	*/
53	class Connection implements LoggerAwareInterface, Stringable
54	{
55	use SendMethodsTrait;
56	use StringableTrait;
57
58	private SocketStream $stream;
59	private HttpHandler $httpHandler;
60	private MessageHandler $messageHandler;
61	private MiddlewareHandler $middlewareHandler;
62	private LoggerInterface $logger;
63	/** @var int<1, max> $frameSize */
64	private int $frameSize = 4096;
65	/** @var int<0, max> $timeout */
66	private int $timeout = 60;
67	private string $localName;
68	private string $remoteName;
69	private RequestInterface\|null $handshakeRequest = null;
70	private ResponseInterface\|null $handshakeResponse = null;
71	/** @var array<string, mixed> $meta */
72	private array $meta = [];
73	private bool $closed = false;
74
75
76	/* ---------- Magic methods ------------------------------------------------------------------------------------ */
77
78	public function __construct(SocketStream $stream, bool $pushMasked, bool $pullMaskedRequired, bool $ssl = false)
79	{
80	$this->stream = $stream;	114✔
81	$this->httpHandler = new HttpHandler($this->stream, $ssl);	114✔
82	$this->messageHandler = new MessageHandler(new FrameHandler($this->stream, $pushMasked, $pullMaskedRequired));	114✔
83	$this->middlewareHandler = new MiddlewareHandler($this->messageHandler, $this->httpHandler);	114✔
84	$this->setLogger(new NullLogger());	114✔
85	$this->localName = $this->stream->getLocalName();	114✔
86	$this->remoteName = $this->stream->getRemoteName();	114✔
87	}
88
89	public function __destruct()
90	{
91	if (!$this->closed && $this->isConnected()) {	114✔
92	$this->stream->close();	83✔
93	}
94	}
95
96	public function __toString(): string
97	{
98	return $this->stringable('%s:%s', $this->localName, $this->remoteName);	1✔
99	}
100
101
102	/* ---------- Configuration ------------------------------------------------------------------------------------ */
103
104	/**
105	* Set logger.
106	* @param LoggerInterface $logger Logger implementation
107	*/
108	public function setLogger(LoggerInterface $logger): void
109	{
110	$this->logger = $logger;	114✔
111	$this->httpHandler->setLogger($logger);	114✔
112	$this->messageHandler->setLogger($logger);	114✔
113	$this->middlewareHandler->setLogger($logger);	114✔
114	$this->logger->debug("[connection] Setting logger: " . get_class($logger));	114✔
115	}
116
117	/**
118	* Set time out on connection.
119	* @param int<0, max> $timeout Timeout part in seconds
120	* @return self
121	*/
122	public function setTimeout(int $timeout): self
123	{
124	if ($timeout < 0) {	70✔
125	throw new InvalidArgumentException("Invalid timeout '{$timeout}' provided");	1✔
126	}
127	$this->timeout = $timeout;	69✔
128	$this->stream->setTimeout($timeout, 0);	69✔
129	$this->logger->debug("[connection] Setting timeout: {$timeout} seconds");	69✔
130	return $this;	69✔
131	}
132
133	/**
134	* Get timeout.
135	* @return int Timeout in seconds.
136	*/
137	public function getTimeout(): int
138	{
139	return $this->timeout;	1✔
140	}
141
142	/**
143	* Set frame size.
144	* @param int<1, max> $frameSize Frame size in bytes.
145	* @return self
146	*/
147	public function setFrameSize(int $frameSize): self
148	{
149	if ($frameSize < 1) {	70✔
150	throw new InvalidArgumentException("Invalid frameSize '{$frameSize}' provided");	1✔
151	}
152	$this->frameSize = $frameSize;	69✔
153	return $this;	69✔
154	}
155
156	/**
157	* Get frame size.
158	* @return int<1, max> Frame size in bytes
159	*/
160	public function getFrameSize(): int
161	{
162	return max(1, $this->frameSize);	39✔
163	}
164
165	/**
166	* Get current stream context.
167	* @return Context
168	*/
169	public function getContext(): Context
170	{
171	return $this->stream->getContext();	3✔
172	}
173
174	/**
175	* Add a middleware.
176	* @param MiddlewareInterface $middleware
177	* @return self
178	*/
179	public function addMiddleware(MiddlewareInterface $middleware): self
180	{
181	$this->middlewareHandler->add($middleware);	37✔
182	$this->logger->debug("[connection] Added middleware: {$middleware}");	37✔
183	return $this;	37✔
184	}
185
186
187	/* ---------- Connection management ---------------------------------------------------------------------------- */
188
189	/**
190	* If connected to stream.
191	* @return bool
192	*/
193	public function isConnected(): bool
194	{
195	return $this->stream->isConnected();	101✔
196	}
197
198	/**
199	* If connection is readable.
200	* @return bool
201	*/
202	public function isReadable(): bool
203	{
204	return $this->stream->isReadable();	12✔
205	}
206
207	/**
208	* If connection is writable.
209	* @return bool
210	*/
211	public function isWritable(): bool
212	{
213	return $this->stream->isWritable();	15✔
214	}
215
216	/**
217	* Close connection stream.
218	* @return self
219	*/
220	public function disconnect(): self
221	{
222	$this->logger->info('[connection] Closing connection');	32✔
223	$this->stream->close();	32✔
224	$this->closed = true;	32✔
225	return $this;	32✔
226	}
227
228	/**
229	* Close connection stream reading.
230	* @return self
231	*/
232	public function closeRead(): self
233	{
234	$this->logger->info('[connection] Closing further reading');	4✔
235	$this->stream->closeRead();	4✔
236	return $this;	4✔
237	}
238
239	/**
240	* Close connection stream writing.
241	* @return self
242	*/
243	public function closeWrite(): self
244	{
245	$this->logger->info('[connection] Closing further writing');	5✔
246	$this->stream->closeWrite();	5✔
247	return $this;	5✔
248	}
249
250	/**
251	* Check if there are messages available to read without blocking.
252	* @return bool True if there are messages available to read, false otherwise.
253	*/
254	public function hasMessages(): bool
255	{
NEW 256	if (!$this->isConnected()) {	×
NEW 257	return false;	×
258	}
259
NEW 260	$resource = $this->stream->getResource();	×
NEW 261	if (!$resource) {	×
NEW 262	return false;	×
263	}
264
NEW 265	$read = [$resource];	×
NEW 266	$write = null;	×
NEW 267	$except = null;	×
268
NEW 269	return @stream_select($read, $write, $except, 0, 0) > 0;	×
270	}
271
272
273	/* ---------- Connection state --------------------------------------------------------------------------------- */
274
275	/**
276	* Get name of local socket, or null if not connected.
277	* @return string\|null
278	*/
279	public function getName(): string\|null
280	{
281	return $this->localName;	3✔
282	}
283
284	/**
285	* Get name of remote socket, or null if not connected.
286	* @return string\|null
287	*/
288	public function getRemoteName(): string\|null
289	{
290	return $this->remoteName;	2✔
291	}
292
293	/**
294	* Set meta value on connection.
295	* @param string $key Meta key
296	* @param mixed $value Meta value
297	*/
298	public function setMeta(string $key, mixed $value): void
299	{
300	$this->meta[$key] = $value;	15✔
301	}
302
303	/**
304	* Get meta value on connection.
305	* @param string $key Meta key
306	* @return mixed Meta value
307	*/
308	public function getMeta(string $key): mixed
309	{
310	return $this->meta[$key] ?? null;	15✔
311	}
312
313	/**
314	* Tick operation on connection.
315	*/
316	public function tick(): void
317	{
318	$this->middlewareHandler->processTick($this);	23✔
319	}
320
321
322	/* ---------- WebSocket Message methods ------------------------------------------------------------------------ */
323
324	/**
325	* Send message.
326	* @template T of Message
327	* @param T $message
328	* @return T
329	*/
330	public function send(Message $message): Message
331	{
332	return $this->pushMessage($message);	30✔
333	}
334
335	/**
336	* Push a message to stream.
337	* @template T of Message
338	* @param T $message
339	* @return T
340	*/
341	public function pushMessage(Message $message): Message
342	{
343	try {
344	return $this->middlewareHandler->processOutgoing($this, $message);	38✔
345	} catch (Throwable $e) {	11✔
346	$this->throwException($e);	11✔
347	}
348	}
349
350	// Pull a message from stream
351	public function pullMessage(): Message
352	{
353	try {
354	return $this->middlewareHandler->processIncoming($this);	34✔
355	} catch (Throwable $e) {	12✔
356	$this->throwException($e);	12✔
357	}
358	}
359
360
361	/* ---------- HTTP Message methods ----------------------------------------------------------------------------- */
362
363	public function pushHttp(MessageInterface $message): MessageInterface
364	{
365	try {
366	return $this->middlewareHandler->processHttpOutgoing($this, $message);	82✔
367	} catch (Throwable $e) {	2✔
368	$this->throwException($e);	2✔
369	}
370	}
371
372	public function pullHttp(): MessageInterface
373	{
374	try {
375	return $this->middlewareHandler->processHttpIncoming($this);	86✔
376	} catch (Throwable $e) {	7✔
377	$this->throwException($e);	7✔
378	}
379	}
380
381	public function setHandshakeRequest(RequestInterface $request): self
382	{
383	$this->handshakeRequest = $request;	56✔
384	return $this;	56✔
385	}
386
387	public function getHandshakeRequest(): RequestInterface\|null
388	{
389	return $this->handshakeRequest;	56✔
390	}
391
392	public function setHandshakeResponse(ResponseInterface $response): self
393	{
394	$this->handshakeResponse = $response;	56✔
395	return $this;	56✔
396	}
397
398	public function getHandshakeResponse(): ResponseInterface\|null
399	{
400	return $this->handshakeResponse;	56✔
401	}
402
403
404	/* ---------- Internal helper methods -------------------------------------------------------------------------- */
405
406	protected function throwException(Throwable $e): never
407	{
408	// Internal exceptions are handled and re-thrown
409	if ($e instanceof ReconnectException) {	32✔
410	$this->logger->info("[connection] {$e->getMessage()}", ['exception' => $e]);	2✔
411	throw $e;	2✔
412	}
413	if ($e instanceof Exception) {	30✔
414	$this->logger->error("[connection] {$e->getMessage()}", ['exception' => $e]);	20✔
415	throw $e;	20✔
416	}
417	// External exceptions are converted to internal
418	if ($this->isConnected()) {	10✔
419	$meta = $this->stream->getMetadata();	9✔
420	$json = json_encode($meta);	9✔
421	if (!empty($meta['timed_out'])) {	9✔
422	$this->logger->error("[connection] {$e->getMessage()}", ['exception' => $e, 'meta' => $meta]);	3✔
423	throw new ConnectionTimeoutException();	3✔
424	}
425	if (!empty($meta['eof'])) {	6✔
426	$this->logger->error("[connection] {$e->getMessage()}", ['exception' => $e, 'meta' => $meta]);	2✔
427	throw new ConnectionClosedException();	2✔
428	}
429	}
430	$this->logger->error("[connection] {$e->getMessage()}", ['exception' => $e]);	5✔
431	throw new ConnectionFailureException();	5✔
432	}
433	}

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