5610959835

Build Type

push

github

Committed by Sören Jensen

Commit Message

Middleware support

Run Details

14 of 14 new or added lines in 4 files covered. (100.0%)

283 of 676 relevant lines covered (41.86%)

1.66 hits per line

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

98.59

/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 $logger;
    private $frameSize = 4096;


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

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

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

    public function __toString(): string
    {
        return sprintf(
            "%s(%s)",
            get_class($this),
            $this->getName() ?: 'closed'
        );
    }


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

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

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

    /**
     * Set fragmentation size.
     * @param int $frameSize Frame size in bytes.
     * @return self.
     */
    public function setFrameSize(int $frameSize): self
    {
        $this->frameSize = $frameSize;
        return $this;
    }

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


    /* ---------- 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();
    }


    /* ---------- 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): Message
    {
        try {
            return $this->middlewareHandler->processOutgoing($this, $message, function (Message $message) {
                $this->messageHandler->push($message, $this->frameSize);
            });
        } 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 $logger;
53	private $frameSize = 4096;
54
55
56	/* ---------- Magic methods ------------------------------------------------------------------------------------ */
57
58	public function __construct(SocketStream $stream, bool $pushMasked, bool $pullMaskedRequired)
59	{
60	$this->stream = $stream;	12✔
61	$this->httpHandler = new HttpHandler($this->stream);	12✔
62	$this->messageHandler = new MessageHandler(new FrameHandler($this->stream, $pushMasked, $pullMaskedRequired));	12✔
63	$this->middlewareHandler = new MiddlewareHandler();	12✔
64	$this->setLogger(new NullLogger());	12✔
65	}
66
67	public function __destruct()
68	{
69	if ($this->isConnected()) {	12✔
70	$this->stream->close();	×
71	}
72	}
73
74	public function __toString(): string
75	{
76	return sprintf(	1✔
77	"%s(%s)",	1✔
78	get_class($this),	1✔
79	$this->getName() ?: 'closed'	1✔
80	);	1✔
81	}
82
83
84	/* ---------- Configuration ------------------------------------------------------------------------------------ */
85
86	/**
87	* Set logger.
88	* @param Psr\Log\LoggerInterface $logger Logger implementation
89	* @return self.
90	*/
91	public function setLogger(LoggerInterface $logger): self
92	{
93	$this->logger = $logger;	12✔
94	$this->httpHandler->setLogger($logger);	12✔
95	$this->messageHandler->setLogger($logger);	12✔
96	$this->middlewareHandler->setLogger($logger);	12✔
97	$this->logger->debug("[connection] Setting logger: " . get_class($logger));	12✔
98	return $this;	12✔
99	}
100
101	/**
102	* Set time out on connection.
103	* @param int $seconds Timeout part in seconds
104	* @return self.
105	*/
106	public function setTimeout(int $seconds): self
107	{
108	$this->stream->setTimeout($seconds, 0);	1✔
109	$this->logger->debug("[connection] Setting timeout: {$seconds} seconds");	1✔
110	return $this;	1✔
111	}
112
113	/**
114	* Set fragmentation size.
115	* @param int $frameSize Frame size in bytes.
116	* @return self.
117	*/
118	public function setFrameSize(int $frameSize): self
119	{
120	$this->frameSize = $frameSize;	1✔
121	return $this;	1✔
122	}
123
124	/**
125	* Add a middleware.
126	* @param MiddlewareInterface $middleware
127	* @return self.
128	*/
129	public function addMiddleware(MiddlewareInterface $middleware): self
130	{
131	$this->middlewareHandler->add($middleware);	1✔
132	$this->logger->debug("[connection] Addded middleware: {$middleware}");	1✔
133	return $this;	1✔
134	}
135
136
137	/* ---------- Connection management ---------------------------------------------------------------------------- */
138
139	/**
140	* If connected to stream.
141	* @return bool
142	*/
143	public function isConnected(): bool
144	{
145	return $this->stream->isConnected();	12✔
146	}
147
148	/**
149	* If connecttion is readable.
150	* @return bool
151	*/
152	public function isReadable(): bool
153	{
154	return $this->stream->isReadable();	1✔
155	}
156
157	/**
158	* If connecttion is writable.
159	* @return bool
160	*/
161	public function isWritable(): bool
162	{
163	return $this->stream->isWritable();	1✔
164	}
165
166	/**
167	* Close connection stream.
168	* @return bool
169	*/
170	public function disconnect(): bool
171	{
172	$this->logger->info('[connection] Closing connection');	12✔
173	$this->stream->close();	12✔
174	return true;	12✔
175	}
176
177	/**
178	* Close connection stream eading.
179	*/
180	public function closeRead(): void
181	{
182	$this->logger->info('[connection] Closing further reading');	1✔
183	$this->stream->closeRead();	1✔
184	}
185
186	/**
187	* Close connection stream writing.
188	*/
189	public function closeWrite(): void
190	{
191	$this->logger->info('[connection] Closing further writing');	1✔
192	$this->stream->closeWrite();	1✔
193	}
194
195
196	/* ---------- Connection state --------------------------------------------------------------------------------- */
197
198	/**
199	* Get name of local socket, or null if not connected.
200	* @return string\|null
201	*/
202	public function getName(): ?string
203	{
204	return $this->stream->getLocalName();	1✔
205	}
206
207	/**
208	* Get name of remote socket, or null if not connected.
209	* @return string\|null
210	*/
211	public function getRemoteName(): ?string
212	{
213	return $this->stream->getRemoteName();	1✔
214	}
215
216
217	/* ---------- WebSocket Message methods ------------------------------------------------------------------------ */
218
219	// Push a message to stream
220	public function pushMessage(Message $message): Message
221	{
222	try {
223	return $this->middlewareHandler->processOutgoing($this, $message, function (Message $message) {	9✔
224	$this->messageHandler->push($message, $this->frameSize);	9✔
225	});	9✔
226	} catch (Throwable $e) {	8✔
227	$this->throwException($e);	8✔
228	}
229	}
230
231	// Pull a message from stream
232	public function pullMessage(): Message
233	{
234	try {
235	return $this->middlewareHandler->processIncoming($this, function () {	2✔
236	return $this->messageHandler->pull();	2✔
237	});	2✔
238	} catch (Throwable $e) {	1✔
239	$this->throwException($e);	1✔
240	}
241	}
242
243
244	/* ---------- HTTP Message methods ----------------------------------------------------------------------------- */
245
246	public function pushHttp(HttpMessage $message): void
247	{
248	$this->httpHandler->push($message);	1✔
249	}
250
251	public function pullHttp(): HttpMessage
252	{
253	return $this->httpHandler->pull();	1✔
254	}
255
256
257	/* ---------- Internal helper methods -------------------------------------------------------------------------- */
258
259	protected function throwException(Throwable $e): void
260	{
261	// Internal exceptions are handled and re-thrown
262	if ($e instanceof Exception) {	9✔
263	$this->logger->error("[connection] {$e->getMessage()} ({$e->getCode()})");	6✔
264	$this->disconnect();	6✔
265	throw $e;	6✔
266	}
267	// External exceptions are converted to internal
268	$exception = get_class($e);	3✔
269	if ($this->isConnected()) {	3✔
270	$meta = $this->stream->getMetadata();	2✔
271	if (!empty($meta['timed_out'])) {	2✔
272	$message = "Connection timeout: {$e->getMessage()}";	1✔
273	$this->logger->error("[connection] {$e->getMessage()} ({$e->getCode()}) original: {$exception}");	1✔
274	$this->disconnect();	1✔
275	throw new TimeoutException($message, Exception::TIMED_OUT, $meta);	1✔
276	}
277	if (!empty($meta['eof'])) {	1✔
278	$message = "Connection closed: {$e->getMessage()}";	1✔
279	$this->logger->error("[connection] {$e->getMessage()} ({$e->getCode()}) original: {$exception}");	1✔
280	$this->disconnect();	1✔
281	throw new ConnectionException($message, Exception::EOF, $meta);	1✔
282	}
283	}
284	$this->disconnect();	1✔
285	$message = "Connection error: {$e->getMessage()}";	1✔
286	$this->logger->error("[connection] {$message} original: {$exception}");	1✔
287	throw new ConnectionException($message, 0);	1✔
288	}
289	}

sirn-se / websocket-php / 5610959835

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