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

0.0

/src/Server.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\{
    StreamFactory,
    Uri
};
use Psr\Log\{
    LoggerAwareInterface,
    LoggerAwareTrait,
    NullLogger
};
use RuntimeException;
use Throwable;
use WebSocket\Http\{
    Request,
    Response
};
use WebSocket\Message\{
    Message,
    Binary,
    Close,
    Ping,
    Pong,
    Text
};
use WebSocket\Middleware\{
    PingResponder
};

/**
 * WebSocket\Server class.
 * Entry class for WebSocket server.
 */
class Server implements LoggerAwareInterface
{
    use LoggerAwareTrait; // Provides setLogger(LoggerInterface $logger)
    use OpcodeTrait;

    // Default options
    protected static $default_options = [
        'fragment_size' => 4096,
        'logger'        => null,
        'masked'        => false,
        'port'          => 8000,
        'schema'        => 'tcp',
        'timeout'       => null,
    ];

    private $streamFactory;
    private $port;
    private $listening;
    private $handshakeRequest;
    private $connection;
    private $options = [];


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

    /**
     * @param array $options
     *   Associative array containing:
     *   - filter:        Array of opcodes to handle. Default: ['text', 'binary'].
     *   - fragment_size: Set framgemnt size.  Default: 4096
     *   - logger:        PSR-3 compatible logger.  Default NullLogger.
     *   - port:          Chose port for listening.  Default 8000.
     *   - schema:        Set socket schema (tcp or ssl).
     *   - timeout:       Set the socket timeout in seconds.
     */
    public function __construct(array $options = [])
    {
        $this->options = array_merge(self::$default_options, $options);
        $this->port = $this->options['port'];
        $this->setLogger($this->options['logger'] ?: new NullLogger());
        $this->setStreamFactory(new StreamFactory());
    }

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


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

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

    /**
     * Set timeout.
     * @param int $timeout Timeout in seconds.
     */
    public function setTimeout(int $timeout): void
    {
        $this->options['timeout'] = $timeout;
        if (!$this->isConnected()) {
            return;
        }
        $this->connection->setTimeout($timeout);
    }

    /**
     * Set fragmentation size.
     * @param int $fragment_size Fragment size in bytes.
     * @return self.
     */
    public function setFragmentSize(int $fragment_size): self
    {
        $this->options['fragment_size'] = $fragment_size;
        if (!$this->connection) {
            return $this;
        }
        $this->connection->setOptions(['fragment_size' => $fragment_size]);
        return $this;
    }

    /**
     * Get fragmentation size.
     * @return int $fragment_size Fragment size in bytes.
     */
    public function getFragmentSize(): int
    {
        return $this->options['fragment_size'];
    }


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

    /**
     * Send text message.
     * @param string $message Content as string.
     * @param bool $masked If message should be masked
     */
    public function text(string $message, ?bool $masked = null): void
    {
        $this->send(new Text($message), $masked);
    }

    /**
     * Send binary message.
     * @param string $message Content as binary string.
     * @param bool $masked If message should be masked
     */
    public function binary(string $message, ?bool $masked = null): void
    {
        $this->send(new Binary($message), $masked);
    }

    /**
     * Send ping.
     * @param string $message Optional text as string.
     * @param bool $masked If message should be masked
     */
    public function ping(string $message = '', ?bool $masked = null): void
    {
        $this->send(new Ping($message), $masked);
    }

    /**
     * Send unsolicited pong.
     * @param string $message Optional text as string.
     * @param bool $masked If message should be masked
     */
    public function pong(string $message = '', ?bool $masked = null): void
    {
        $this->send(new Pong($message), $masked);
    }

    /**
     * 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'): void
    {
        if (!$this->isConnected()) {
            return;
        }
        $this->connection->close($status, $message);
    }

    /**
     * Send message.
     * @param Message $message Message to send.
     * @param bool $masked If message should be masked
     */
    public function send(Message $message, ?bool $masked = null): void
    {
        if (!$this->isConnected()) {
            $this->connect();
        }
        $this->connection->pushMessage($message, $masked);
    }

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


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

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

    // Connect when read/write operation is performed.
    public function connect(): void
    {
        try {
            if (isset($this->options['timeout'])) {
                $socket = $this->listening->accept($this->options['timeout']);
            } else {
                $socket = $this->listening->accept();
            }
            if (!$socket) {
                throw new RuntimeException('No socket');
            }
        } catch (RuntimeException $e) {
            $error = "Server failed to connect. {$e->getMessage()}";
            $this->logger->error($error);
            throw new ConnectionException($error, ConnectionException::SERVER_ACCEPT_ERR, [], $e);
        }

        $this->connection = new Connection($socket, $this->options);
        $this->connection->setLogger($this->logger);
        $this->connection->addMiddleware(new PingResponder());

        $this->logger->info("Client has connected to port {port}", [
            'port' => $this->port,
        ]);
        $this->performHandshake($this->connection);
    }

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

    /**
     * Accept a single incoming request.
     * Note that this operation will block accepting additional requests.
     * @return bool True if listening.
     */
    public function accept(): bool
    {
        $this->disconnect();
        $exception = null;

        do {
            try {
                $uri = new Uri("{$this->options['schema']}://0.0.0.0:{$this->port}");
                $this->listening = $this->streamFactory->createSocketServer($uri);
            } catch (RuntimeException $e) {
                $this->logger->error("Could not connect on port {$this->port}: {$e->getMessage()}");
                $exception = $e;
            }
        } while (is_null($this->listening) && $this->port++ < 10000);

        if (!$this->listening) {
            $error = "Could not open listening socket: {$exception->getMessage()}";
            $this->logger->error($error);
            throw new ConnectionException($error, ConnectionException::SERVER_SOCKET_ERR);
        }

        $this->logger->info("Server listening to port {$uri}");

        return (bool)$this->listening;
    }


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

    /**
     * Get name of local socket from single connection.
     * @return string|null Name of local socket.
     */
    public function getName(): ?string
    {
        return $this->isConnected() ? $this->connection->getName() : null;
    }

    /**
     * Get name of remote socket from single connection.
     * @return string|null Name of remote socket.
     */
    public function getRemoteName(): ?string
    {
        return $this->isConnected() ? $this->connection->getRemoteName() : null;
    }

    /**
     * Get current port.
     * @return int port.
     */
    public function getPort(): int
    {
        return $this->port;
    }

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


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

    // Perform upgrade handshake on new connections.
    protected function performHandshake(Connection $connection): void
    {
        $response = new Response(101);

        try {
            $request = $connection->pullHttp();
        } catch (RuntimeException $e) {
            $error = 'Client handshake error';
            $this->logger->error($error);
            throw new ConnectionException($error, ConnectionException::SERVER_HANDSHAKE_ERR);
        }

        $key = trim((string)$request->getHeaderLine('Sec-WebSocket-Key'));
        if (empty($key)) {
            $error = sprintf(
                "Client had no Key in upgrade request: %s",
                json_encode($request->getHeaders())
            );
            $this->logger->error($error);
            throw new ConnectionException($error, ConnectionException::SERVER_HANDSHAKE_ERR);
        }

        /// @todo Validate key length and base 64...
        $response_key = base64_encode(pack('H*', sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11')));

        $response = $response
            ->withHeader('Upgrade', 'websocket')
            ->withHeader('Connection', 'Upgrade')
            ->withHeader('Sec-WebSocket-Accept', $response_key);
        $connection->pushHttp($response);

        $this->logger->debug("Handshake on {$request->getUri()->getPath()}");

        $this->handshakeRequest = $request;
    }
}

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\{
13	StreamFactory,
14	Uri
15	};
16	use Psr\Log\{
17	LoggerAwareInterface,
18	LoggerAwareTrait,
19	NullLogger
20	};
21	use RuntimeException;
22	use Throwable;
23	use WebSocket\Http\{
24	Request,
25	Response
26	};
27	use WebSocket\Message\{
28	Message,
29	Binary,
30	Close,
31	Ping,
32	Pong,
33	Text
34	};
35	use WebSocket\Middleware\{
36	PingResponder
37	};
38
39	/**
40	* WebSocket\Server class.
41	* Entry class for WebSocket server.
42	*/
43	class Server implements LoggerAwareInterface
44	{
45	use LoggerAwareTrait; // Provides setLogger(LoggerInterface $logger)
46	use OpcodeTrait;
47
48	// Default options
49	protected static $default_options = [
50	'fragment_size' => 4096,
51	'logger' => null,
52	'masked' => false,
53	'port' => 8000,
54	'schema' => 'tcp',
55	'timeout' => null,
56	];
57
58	private $streamFactory;
59	private $port;
60	private $listening;
61	private $handshakeRequest;
62	private $connection;
63	private $options = [];
64
65
66	/* ---------- Magic methods ------------------------------------------------------------------------------------ */
67
68	/**
69	* @param array $options
70	* Associative array containing:
71	* - filter: Array of opcodes to handle. Default: ['text', 'binary'].
72	* - fragment_size: Set framgemnt size. Default: 4096
73	* - logger: PSR-3 compatible logger. Default NullLogger.
74	* - port: Chose port for listening. Default 8000.
75	* - schema: Set socket schema (tcp or ssl).
76	* - timeout: Set the socket timeout in seconds.
77	*/
78	public function __construct(array $options = [])
79	{
80	$this->options = array_merge(self::$default_options, $options);	×
81	$this->port = $this->options['port'];	×
82	$this->setLogger($this->options['logger'] ?: new NullLogger());	×
83	$this->setStreamFactory(new StreamFactory());	×
84	}
85
86	/**
87	* Get string representation of instance.
88	* @return string String representation.
89	*/
90	public function __toString(): string
91	{
92	return sprintf(	×
93	"%s(%s)",	×
94	get_class($this),	×
95	$this->getName() ?: 'closed'	×
96	);	×
97	}
98
99
100	/* ---------- Configuration ------------------------------------------------------------------------------------ */
101
102	/**
103	* Set stream factory to use.
104	* @param StreamFactory $streamFactory.
105	*/
106	public function setStreamFactory(StreamFactory $streamFactory)
107	{
108	$this->streamFactory = $streamFactory;	×
109	}
110
111	/**
112	* Set timeout.
113	* @param int $timeout Timeout in seconds.
114	*/
115	public function setTimeout(int $timeout): void
116	{
117	$this->options['timeout'] = $timeout;	×
118	if (!$this->isConnected()) {	×
119	return;	×
120	}
121	$this->connection->setTimeout($timeout);	×
122	}
123
124	/**
125	* Set fragmentation size.
126	* @param int $fragment_size Fragment size in bytes.
127	* @return self.
128	*/
129	public function setFragmentSize(int $fragment_size): self
130	{
131	$this->options['fragment_size'] = $fragment_size;	×
132	if (!$this->connection) {	×
133	return $this;	×
134	}
135	$this->connection->setOptions(['fragment_size' => $fragment_size]);	×
136	return $this;	×
137	}
138
139	/**
140	* Get fragmentation size.
141	* @return int $fragment_size Fragment size in bytes.
142	*/
143	public function getFragmentSize(): int
144	{
145	return $this->options['fragment_size'];	×
146	}
147
148
149	/* ---------- Messaging operations ----------------------------------------------------------------------------- */
150
151	/**
152	* Send text message.
153	* @param string $message Content as string.
154	* @param bool $masked If message should be masked
155	*/
156	public function text(string $message, ?bool $masked = null): void
157	{
158	$this->send(new Text($message), $masked);	×
159	}
160
161	/**
162	* Send binary message.
163	* @param string $message Content as binary string.
164	* @param bool $masked If message should be masked
165	*/
166	public function binary(string $message, ?bool $masked = null): void
167	{
168	$this->send(new Binary($message), $masked);	×
169	}
170
171	/**
172	* Send ping.
173	* @param string $message Optional text as string.
174	* @param bool $masked If message should be masked
175	*/
176	public function ping(string $message = '', ?bool $masked = null): void
177	{
178	$this->send(new Ping($message), $masked);	×
179	}
180
181	/**
182	* Send unsolicited pong.
183	* @param string $message Optional text as string.
184	* @param bool $masked If message should be masked
185	*/
186	public function pong(string $message = '', ?bool $masked = null): void
187	{
188	$this->send(new Pong($message), $masked);	×
189	}
190
191	/**
192	* Tell the socket to close.
193	* @param integer $status http://tools.ietf.org/html/rfc6455#section-7.4
194	* @param string $message A closing message, max 125 bytes.
195	*/
196	public function close(int $status = 1000, string $message = 'ttfn'): void
197	{
198	if (!$this->isConnected()) {	×
199	return;	×
200	}
201	$this->connection->close($status, $message);	×
202	}
203
204	/**
205	* Send message.
206	* @param Message $message Message to send.
207	* @param bool $masked If message should be masked
208	*/
209	public function send(Message $message, ?bool $masked = null): void
210	{
211	if (!$this->isConnected()) {	×
212	$this->connect();	×
213	}
214	$this->connection->pushMessage($message, $masked);	×
215	}
216
217	/**
218	* Receive message.
219	* Note that this operation will block reading.
220	* @return Message\|null
221	*/
222	public function receive(): ?Message
223	{
224	if (!$this->isConnected()) {	×
225	$this->connect();	×
226	}
227	return $this->connection->pullMessage();	×
228	}
229
230
231	/* ---------- Connection management ---------------------------------------------------------------------------- */
232
233	/**
234	* If Server has active connections.
235	* @return bool True if active connection.
236	*/
237	public function isConnected(): bool
238	{
239	return $this->connection && $this->connection->isConnected();	×
240	}
241
242	// Connect when read/write operation is performed.
243	public function connect(): void
244	{
245	try {
246	if (isset($this->options['timeout'])) {	×
247	$socket = $this->listening->accept($this->options['timeout']);	×
248	} else {
249	$socket = $this->listening->accept();	×
250	}
251	if (!$socket) {	×
252	throw new RuntimeException('No socket');	×
253	}
254	} catch (RuntimeException $e) {	×
255	$error = "Server failed to connect. {$e->getMessage()}";	×
256	$this->logger->error($error);	×
257	throw new ConnectionException($error, ConnectionException::SERVER_ACCEPT_ERR, [], $e);	×
258	}
259
260	$this->connection = new Connection($socket, $this->options);	×
261	$this->connection->setLogger($this->logger);	×
262	$this->connection->addMiddleware(new PingResponder());	×
263
264	$this->logger->info("Client has connected to port {port}", [	×
265	'port' => $this->port,	×
266	]);	×
267	$this->performHandshake($this->connection);	×
268	}
269
270	/**
271	* Disconnect client.
272	*/
273	public function disconnect(): void
274	{
275	if ($this->isConnected()) {	×
276	$this->connection->disconnect();	×
277	$this->logger->info('[server] Server disconnected');	×
278	}
279	}
280
281	/**
282	* Accept a single incoming request.
283	* Note that this operation will block accepting additional requests.
284	* @return bool True if listening.
285	*/
286	public function accept(): bool
287	{
288	$this->disconnect();	×
289	$exception = null;	×
290
291	do {
292	try {
293	$uri = new Uri("{$this->options['schema']}://0.0.0.0:{$this->port}");	×
294	$this->listening = $this->streamFactory->createSocketServer($uri);	×
295	} catch (RuntimeException $e) {	×
296	$this->logger->error("Could not connect on port {$this->port}: {$e->getMessage()}");	×
297	$exception = $e;	×
298	}
299	} while (is_null($this->listening) && $this->port++ < 10000);	×
300
301	if (!$this->listening) {	×
302	$error = "Could not open listening socket: {$exception->getMessage()}";	×
303	$this->logger->error($error);	×
304	throw new ConnectionException($error, ConnectionException::SERVER_SOCKET_ERR);	×
305	}
306
307	$this->logger->info("Server listening to port {$uri}");	×
308
309	return (bool)$this->listening;	×
310	}
311
312
313	/* ---------- Connection state --------------------------------------------------------------------------------- */
314
315	/**
316	* Get name of local socket from single connection.
317	* @return string\|null Name of local socket.
318	*/
319	public function getName(): ?string
320	{
321	return $this->isConnected() ? $this->connection->getName() : null;	×
322	}
323
324	/**
325	* Get name of remote socket from single connection.
326	* @return string\|null Name of remote socket.
327	*/
328	public function getRemoteName(): ?string
329	{
330	return $this->isConnected() ? $this->connection->getRemoteName() : null;	×
331	}
332
333	/**
334	* Get current port.
335	* @return int port.
336	*/
337	public function getPort(): int
338	{
339	return $this->port;	×
340	}
341
342	/**
343	* Get Request for handshake procedure.
344	* @return Request\|null Handshake.
345	*/
346	public function getHandshakeRequest(): ?Request
347	{
348	return $this->connection ? $this->handshakeRequest : null;	×
349	}
350
351
352	/* ---------- Internal helper methods -------------------------------------------------------------------------- */
353
354	// Perform upgrade handshake on new connections.
355	protected function performHandshake(Connection $connection): void
356	{
357	$response = new Response(101);	×
358
359	try {
360	$request = $connection->pullHttp();	×
361	} catch (RuntimeException $e) {	×
362	$error = 'Client handshake error';	×
363	$this->logger->error($error);	×
364	throw new ConnectionException($error, ConnectionException::SERVER_HANDSHAKE_ERR);	×
365	}
366
367	$key = trim((string)$request->getHeaderLine('Sec-WebSocket-Key'));	×
368	if (empty($key)) {	×
369	$error = sprintf(	×
370	"Client had no Key in upgrade request: %s",	×
371	json_encode($request->getHeaders())	×
372	);	×
373	$this->logger->error($error);	×
374	throw new ConnectionException($error, ConnectionException::SERVER_HANDSHAKE_ERR);	×
375	}
376
377	/// @todo Validate key length and base 64...
378	$response_key = base64_encode(pack('H*', sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11')));	×
379
380	$response = $response	×
381	->withHeader('Upgrade', 'websocket')	×
382	->withHeader('Connection', 'Upgrade')	×
383	->withHeader('Sec-WebSocket-Accept', $response_key);	×
384	$connection->pushHttp($response);	×
385
386	$this->logger->debug("Handshake on {$request->getUri()->getPath()}");	×
387
388	$this->handshakeRequest = $request;	×
389	}
390	}

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