5608975860

Build Type

push

github

Committed by Sören Jensen

Commit Message

Middleware support

Run Details

90 of 90 new or added lines in 8 files covered. (100.0%)

245 of 671 relevant lines covered (36.51%)

1.27 hits per line

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

47.83

/src/Connection.php

<?php

/**
 * Copyright (C) 2014-2023 Textalk and contributors.
 *
 * This file is part of Websocket PHP and is free software under the ISC License.
 * License text: https://raw.githubusercontent.com/sirn-se/websocket-php/master/COPYING.md
 */

namespace WebSocket;

use Phrity\Net\SocketStream;
use Psr\Log\{
    LoggerAwareInterface,
    LoggerInterface,
    NullLogger
};
use RuntimeException;
use Throwable;
use WebSocket\Exception;
use WebSocket\Frame\FrameHandler;
use WebSocket\Http\{
    HttpHandler,
    Message as HttpMessage
};
use WebSocket\Message\{
    Message,
    Binary,
    Close,
    Ping,
    Pong,
    Text,
    MessageHandler
};
use WebSocket\Middleware\{
    MiddlewareHandler,
    MiddlewareInterface
};

/**
 * WebSocket\Connection class.
 * A client/server connection.
 */
class Connection implements LoggerAwareInterface
{
    use OpcodeTrait;

    private $stream;
    private $httpHandler;
    private $messageHandler;
    private $middlewareHandler;
    private $options = ['masked' => false, 'fragment_size' => 4096];
    private $logger;


    /* ---------- Construct & Destruct ----------------------------------------------------------------------------- */

    public function __construct(SocketStream $stream, array $options = [])
    {
        $this->stream = $stream;
        $this->httpHandler = new HttpHandler($this->stream);
        $this->messageHandler = new MessageHandler(new FrameHandler($this->stream));
        $this->middlewareHandler = new MiddlewareHandler();
        $this->setLogger(new NullLogger());
        $this->setOptions($options);
    }

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

    public function __toString(): string
    {
        return get_class($this);
    }


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

    /**
     * Set logger.
     * @param Psr\Log\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);
    }

    /**
     * Set connection options.
     * @param array $options Options
     */
    public function setOptions(array $options = []): void
    {
        $this->options = array_merge($this->options, $options);
        if (!empty($options['logger'])) {
            $this->setLogger($options['logger']);
        }
        if (!empty($options['timeout'])) {
            $this->setTimeout($options['timeout']);
        }
    }

    /**
     * Set time out on connection.
     * @param int $seconds Timeout part in seconds
     * @param int $microseconds Timeout part in microseconds
     */
    public function setTimeout(int $seconds, int $microseconds = 0): void
    {
        $this->stream->setTimeout($seconds, $microseconds);
        $this->logger->debug("[connection] Setting timeout {$seconds}.{$microseconds} seconds");
    }

    /**
     * Add a middleware.
     * @param MiddlewareInterface $middleware
     */
    public function addMiddleware(MiddlewareInterface $middleware): void
    {
        $this->middlewareHandler->add($middleware);
    }


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

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

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

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

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

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

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

    /**
     * Tell the socket to close.
     * @param integer $status  http://tools.ietf.org/html/rfc6455#section-7.4
     * @param string  $message A closing message, max 125 bytes.
     */
    public function close(int $status = 1000, string $message = 'ttfn', ?bool $masked = null): void
    {
        $this->pushMessage(new Close($status, $message), $masked);
        $this->logger->debug("[connection] Closing with status: {$status}.");
    }


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

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

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


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

    // Push a message to stream
    public function pushMessage(Message $message, ?bool $masked = null): void
    {
        try {
            $masked = is_null($masked) ? $this->options['masked'] : $masked;
            $this->middlewareHandler->processOutgoing($this, $message, function (Message $message) use ($masked) {
                $this->messageHandler->push($message, $masked, $this->options['fragment_size']);
            });
        } catch (Throwable $e) {
            $this->throwException($e);
        }
    }

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


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

    public function pushHttp(HttpMessage $message): void
    {
        $this->httpHandler->push($message);
    }

    public function pullHttp(): HttpMessage
    {
        return $this->httpHandler->pull();
    }


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

    protected function throwException(Throwable $e): void
    {
        // Internal exceptions are handled and re-thrown
        if ($e instanceof Exception) {
            $this->logger->error("[connection] {$e->getMessage()} ({$e->getCode()})");
            $this->disconnect();
            throw $e;
        }
        // External exceptions are converted to internal
        $exception = get_class($e);
        if ($this->isConnected()) {
            $meta = $this->stream->getMetadata();
            if (!empty($meta['timed_out'])) {
                $message = "Connection timeout: {$e->getMessage()}";
                $this->logger->error("[connection] {$e->getMessage()} ({$e->getCode()}) original: {$exception}");
                $this->disconnect();
                throw new TimeoutException($message, Exception::TIMED_OUT, $meta);
            }
            if (!empty($meta['eof'])) {
                $message = "Connection closed: {$e->getMessage()}";
                $this->logger->error("[connection] {$e->getMessage()} ({$e->getCode()}) original: {$exception}");
                $this->disconnect();
                throw new ConnectionException($message, Exception::EOF, $meta);
            }
        }
        $this->disconnect();
        $message = "Connection error: {$e->getMessage()}";
        $this->logger->error("[connection] {$message}  original: {$exception}");
        throw new ConnectionException($message, 0);
    }
}

1	<?php
2
3	/**
4	* Copyright (C) 2014-2023 Textalk and contributors.
5	*
6	* This file is part of Websocket PHP and is free software under the ISC License.
7	* License text: https://raw.githubusercontent.com/sirn-se/websocket-php/master/COPYING.md
8	*/
9
10	namespace WebSocket;
11
12	use Phrity\Net\SocketStream;
13	use Psr\Log\{
14	LoggerAwareInterface,
15	LoggerInterface,
16	NullLogger
17	};
18	use RuntimeException;
19	use Throwable;
20	use WebSocket\Exception;
21	use WebSocket\Frame\FrameHandler;
22	use WebSocket\Http\{
23	HttpHandler,
24	Message as HttpMessage
25	};
26	use WebSocket\Message\{
27	Message,
28	Binary,
29	Close,
30	Ping,
31	Pong,
32	Text,
33	MessageHandler
34	};
35	use WebSocket\Middleware\{
36	MiddlewareHandler,
37	MiddlewareInterface
38	};
39
40	/**
41	* WebSocket\Connection class.
42	* A client/server connection.
43	*/
44	class Connection implements LoggerAwareInterface
45	{
46	use OpcodeTrait;
47
48	private $stream;
49	private $httpHandler;
50	private $messageHandler;
51	private $middlewareHandler;
52	private $options = ['masked' => false, 'fragment_size' => 4096];
53	private $logger;
54
55
56	/* ---------- Construct & Destruct ----------------------------------------------------------------------------- */
57
58	public function __construct(SocketStream $stream, array $options = [])
59	{
60	$this->stream = $stream;	5✔
61	$this->httpHandler = new HttpHandler($this->stream);	5✔
62	$this->messageHandler = new MessageHandler(new FrameHandler($this->stream));	5✔
63	$this->middlewareHandler = new MiddlewareHandler();	5✔
64	$this->setLogger(new NullLogger());	5✔
65	$this->setOptions($options);	5✔
66	}
67
68	public function __destruct()
69	{
70	if ($this->isConnected()) {	5✔
71	$this->stream->close();	2✔
72	}
73	}
74
75	public function __toString(): string
76	{
77	return get_class($this);	×
78	}
79
80
81	/* ---------- Configuration ------------------------------------------------------------------------------------ */
82
83	/**
84	* Set logger.
85	* @param Psr\Log\LoggerInterface $logger Logger implementation
86	*/
87	public function setLogger(LoggerInterface $logger): void
88	{
89	$this->logger = $logger;	5✔
90	$this->httpHandler->setLogger($logger);	5✔
91	$this->messageHandler->setLogger($logger);	5✔
92	$this->middlewareHandler->setLogger($logger);	5✔
93	}
94
95	/**
96	* Set connection options.
97	* @param array $options Options
98	*/
99	public function setOptions(array $options = []): void
100	{
101	$this->options = array_merge($this->options, $options);	5✔
102	if (!empty($options['logger'])) {	5✔
103	$this->setLogger($options['logger']);	×
104	}
105	if (!empty($options['timeout'])) {	5✔
106	$this->setTimeout($options['timeout']);	×
107	}
108	}
109
110	/**
111	* Set time out on connection.
112	* @param int $seconds Timeout part in seconds
113	* @param int $microseconds Timeout part in microseconds
114	*/
115	public function setTimeout(int $seconds, int $microseconds = 0): void
116	{
117	$this->stream->setTimeout($seconds, $microseconds);	×
118	$this->logger->debug("[connection] Setting timeout {$seconds}.{$microseconds} seconds");	×
119	}
120
121	/**
122	* Add a middleware.
123	* @param MiddlewareInterface $middleware
124	*/
125	public function addMiddleware(MiddlewareInterface $middleware): void
126	{
127	$this->middlewareHandler->add($middleware);	5✔
128	}
129
130
131	/* ---------- Connection management ---------------------------------------------------------------------------- */
132
133	/**
134	* If connected to stream.
135	* @return bool
136	*/
137	public function isConnected(): bool
138	{
139	return $this->stream->isConnected();	5✔
140	}
141
142	/**
143	* If connecttion is readable.
144	* @return bool
145	*/
146	public function isReadable(): bool
147	{
148	return $this->stream->isReadable();	2✔
149	}
150
151	/**
152	* If connecttion is writable.
153	* @return bool
154	*/
155	public function isWritable(): bool
156	{
157	return $this->stream->isWritable();	2✔
158	}
159
160	/**
161	* Close connection stream.
162	* @return bool
163	*/
164	public function disconnect(): bool
165	{
166	$this->logger->info('[connection] Closing connection');	3✔
167	$this->stream->close();	3✔
168	return true;	3✔
169	}
170
171	/**
172	* Close connection stream eading.
173	*/
174	public function closeRead(): void
175	{
176	$this->logger->info('[connection] Closing further reading');	1✔
177	$this->stream->closeRead();	1✔
178	}
179
180	/**
181	* Close connection stream writing.
182	*/
183	public function closeWrite(): void
184	{
185	$this->logger->info('[connection] Closing further writing');	1✔
186	$this->stream->closeWrite();	1✔
187	}
188
189	/**
190	* Tell the socket to close.
191	* @param integer $status http://tools.ietf.org/html/rfc6455#section-7.4
192	* @param string $message A closing message, max 125 bytes.
193	*/
194	public function close(int $status = 1000, string $message = 'ttfn', ?bool $masked = null): void
195	{
196	$this->pushMessage(new Close($status, $message), $masked);	×
197	$this->logger->debug("[connection] Closing with status: {$status}.");	×
198	}
199
200
201	/* ---------- Connection state --------------------------------------------------------------------------------- */
202
203	/**
204	* Get name of local socket, or null if not connected.
205	* @return string\|null
206	*/
207	public function getName(): ?string
208	{
209	return $this->stream->getLocalName();	×
210	}
211
212	/**
213	* Get name of remote socket, or null if not connected.
214	* @return string\|null
215	*/
216	public function getRemoteName(): ?string
217	{
218	return $this->stream->getRemoteName();	×
219	}
220
221
222	/* ---------- WebSocket Message methods ------------------------------------------------------------------------ */
223
224	// Push a message to stream
225	public function pushMessage(Message $message, ?bool $masked = null): void
226	{
227	try {
228	$masked = is_null($masked) ? $this->options['masked'] : $masked;	4✔
229	$this->middlewareHandler->processOutgoing($this, $message, function (Message $message) use ($masked) {	4✔
230	$this->messageHandler->push($message, $masked, $this->options['fragment_size']);	4✔
231	});	4✔
232	} catch (Throwable $e) {	×
233	$this->throwException($e);	×
234	}
235	}
236
237	// Pull a message from stream
238	public function pullMessage(): Message
239	{
240	try {
241	return $this->middlewareHandler->processIncoming($this, function () {	4✔
242	return $this->messageHandler->pull();	4✔
243	});	4✔
244	} catch (Throwable $e) {	×
245	$this->throwException($e);	×
246	}
247	}
248
249
250	/* ---------- HTTP Message methods ----------------------------------------------------------------------------- */
251
252	public function pushHttp(HttpMessage $message): void
253	{
254	$this->httpHandler->push($message);	×
255	}
256
257	public function pullHttp(): HttpMessage
258	{
259	return $this->httpHandler->pull();	×
260	}
261
262
263	/* ---------- Internal helper methods -------------------------------------------------------------------------- */
264
265	protected function throwException(Throwable $e): void
266	{
267	// Internal exceptions are handled and re-thrown
268	if ($e instanceof Exception) {	×
269	$this->logger->error("[connection] {$e->getMessage()} ({$e->getCode()})");	×
270	$this->disconnect();	×
271	throw $e;	×
272	}
273	// External exceptions are converted to internal
274	$exception = get_class($e);	×
275	if ($this->isConnected()) {	×
276	$meta = $this->stream->getMetadata();	×
277	if (!empty($meta['timed_out'])) {	×
278	$message = "Connection timeout: {$e->getMessage()}";	×
279	$this->logger->error("[connection] {$e->getMessage()} ({$e->getCode()}) original: {$exception}");	×
280	$this->disconnect();	×
281	throw new TimeoutException($message, Exception::TIMED_OUT, $meta);	×
282	}
283	if (!empty($meta['eof'])) {	×
284	$message = "Connection closed: {$e->getMessage()}";	×
285	$this->logger->error("[connection] {$e->getMessage()} ({$e->getCode()}) original: {$exception}");	×
286	$this->disconnect();	×
287	throw new ConnectionException($message, Exception::EOF, $meta);	×
288	}
289	}
290	$this->disconnect();	×
291	$message = "Connection error: {$e->getMessage()}";	×
292	$this->logger->error("[connection] {$message} original: {$exception}");	×
293	throw new ConnectionException($message, 0);	×
294	}
295	}

sirn-se / websocket-php / 5608975860

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