13760794555

Committed 10 Mar 2025 09:06AM UTC coverage: 99.53% (-0.5%) from 100.0%

Build # 13760794555

Build Type

push

github

Committed by

sirn-se

Commit Message

Use context class Client/Server/Connection

Coverage Stats

8 of 13 new or added lines in 3 files covered. (61.54%)

1058 of 1063 relevant lines covered (99.53%)

22.2 hits per line

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

99.13

/src/Server.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,
    SocketServer,
    StreamCollection,
    StreamException,
    StreamFactory,
    Uri
};
use Psr\Log\{
    LoggerAwareInterface,
    LoggerInterface,
    NullLogger
};
use Stringable;
use Throwable;
use WebSocket\Exception\{
    CloseException,
    ConnectionFailureException,
    ConnectionLevelInterface,
    Exception,
    HandshakeException,
    MessageLevelInterface,
    ServerException
};
use WebSocket\Http\{
    Response,
    ServerRequest
};
use WebSocket\Message\Message;
use WebSocket\Middleware\MiddlewareInterface;
use WebSocket\Trait\{
    ListenerTrait,
    SendMethodsTrait,
    StringableTrait
};

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

    private const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';

    // Settings
    private int $port;
    private string $scheme;
    private LoggerInterface $logger;
    private int $timeout = 60;
    private int $frameSize = 4096;
    private Context $context;

    // Internal resources
    private StreamFactory $streamFactory;
    private SocketServer|null $server = null;
    private StreamCollection|null $streams = null;
    private bool $running = false;
    /** @var array<Connection> $connections */
    private array $connections = [];
    /** @var array<MiddlewareInterface> $middlewares */
    private array $middlewares = [];
    private int|null $maxConnections = null;


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

    /**
     * @param int $port Socket port to listen to
     * @param bool $ssl If SSL should be used
     * @throws InvalidArgumentException If invalid port provided
     */
    public function __construct(int $port = 80, bool $ssl = false)
    {
        if ($port < 0 || $port > 65535) {
            throw new InvalidArgumentException("Invalid port '{$port}' provided");
        }
        $this->port = $port;
        $this->scheme = $ssl ? 'ssl' : 'tcp';
        $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->server ? "{$this->scheme}://0.0.0.0:{$this->port}" : '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;
        foreach ($this->connections as $connection) {
            $connection->setLogger($this->logger);
        }
    }

    /**
     * Set timeout.
     * @param int $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;
        foreach ($this->connections as $connection) {
            $connection->setTimeout($timeout);
        }
        return $this;
    }

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

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

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

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

    /**
     * Get connection scheme.
     * @return string scheme
     */
    public function getScheme(): string
    {
        return $this->scheme;
    }

    /**
     * Get connection scheme.
     * @return bool SSL mode
     */
    public function isSsl(): bool
    {
        return $this->scheme === 'ssl';
    }

    /**
     * Number of currently connected clients.
     * @return int Connection count
     */
    public function getConnectionCount(): int
    {
        return count($this->connections);
    }

    /**
     * Get currently connected clients.
     * @return array<Connection> Connections
     */
    public function getConnections(): array
    {
        return $this->connections;
    }

    /**
     * Get currently readable clients.
     * @return array<Connection> Connections
     */
    public function getReadableConnections(): array
    {
        return array_filter($this->connections, function (Connection $connection) {
            return $connection->isReadable();
        });
    }

    /**
     * Get currently writable clients.
     * @return array<Connection> Connections
     */
    public function getWritableConnections(): array
    {
        return array_filter($this->connections, function (Connection $connection) {
            return $connection->isWritable();
        });
    }

    /**
     * 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 a middleware.
     * @param MiddlewareInterface $middleware
     * @return self
     */
    public function addMiddleware(MiddlewareInterface $middleware): self
    {
        $this->middlewares[] = $middleware;
        foreach ($this->connections as $connection) {
            $connection->addMiddleware($middleware);
        }
        return $this;
    }

    /**
     * Set maximum number of connections allowed, null means unlimited.
     * @param int|null $maxConnections
     * @return self
     */
    public function setMaxConnections(int|null $maxConnections): self
    {
        if ($maxConnections !== null && $maxConnections < 1) {
            throw new InvalidArgumentException("Invalid maxConnections '{$maxConnections}' provided");
        }
        $this->maxConnections = $maxConnections;
        return $this;
    }


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

    /**
     * Send message (broadcast to all connected clients).
     * @template T of Message
     * @param T $message
     * @return T
     */
    public function send(Message $message): Message
    {
        foreach ($this->connections as $connection) {
            if ($connection->isWritable()) {
                $connection->send($message);
            }
        }
        return $message;
    }


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

    /**
     * Start server listener.
     * @throws Throwable On low level error
     */
    public function start(): void
    {
        // Create socket server
        if (empty($this->server)) {
            $this->createSocketServer();
        }

        // Check if running
        if ($this->running) {
            $this->logger->warning("[server] Server is already running");
            return;
        }
        $this->running = true;
        $this->logger->info("[server] Server is running");

        // Run handler
        while ($this->running) {
            try {
                // Clear closed connections
                $this->detachUnconnected();
                if (is_null($this->streams)) {
                    $this->stop();
                    return;
                }

                // Get streams with readable content
                $readables = $this->streams->waitRead($this->timeout);
                foreach ($readables as $key => $readable) {
                    try {
                        $connection = null;
                        // Accept new client connection
                        if ($readable instanceof SocketServer) {
                            $this->acceptSocket($readable);
                            continue;
                        }
                        // Read from connection
                        $connection = $this->connections[$key];
                        $message = $connection->pullMessage();
                        $this->dispatch($message->getOpcode(), [$this, $connection, $message]);
                    } catch (MessageLevelInterface $e) {
                        // Error, but keep connection open
                        $this->logger->error("[server] {$e->getMessage()}", ['exception' => $e]);
                        $this->dispatch('error', [$this, $connection, $e]);
                    } catch (ConnectionLevelInterface $e) {
                        // Error, disconnect connection
                        if ($connection) {
                            $this->streams->detach($key);
                            unset($this->connections[$key]);
                            $connection->disconnect();
                        }
                        $this->logger->error("[server] {$e->getMessage()}", ['exception' => $e]);
                        $this->dispatch('error', [$this, $connection, $e]);
                    } catch (CloseException $e) {
                        // Should close
                        if ($connection) {
                            $connection->close($e->getCloseStatus(), $e->getMessage());
                        }
                        $this->logger->error("[server] {$e->getMessage()}", ['exception' => $e]);
                        $this->dispatch('error', [$this, $connection, $e]);
                    }
                }
                foreach ($this->connections as $connection) {
                    $connection->tick();
                }
                $this->dispatch('tick', [$this]);
            } catch (Exception $e) {
                // Low-level error
                $this->logger->error("[server] {$e->getMessage()}", ['exception' => $e]);
                $this->dispatch('error', [$this, null, $e]);
            } catch (Throwable $e) {
                // Crash it
                $this->logger->error("[server] {$e->getMessage()}", ['exception' => $e]);
                $this->dispatch('error', [$this, null, $e]);
                $this->disconnect();
                throw $e;
            }
            gc_collect_cycles(); // Collect garbage
        }
    }

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

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


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

    /**
     * Orderly shutdown of server.
     * @param int $closeStatus Default is 1001 "Going away"
     */
    public function shutdown(int $closeStatus = 1001): void
    {
        $this->logger->info('[server] Shutting down');
        if ($this->getConnectionCount() == 0) {
            $this->disconnect();
            return;
        }
        // Store and reset settings, lock new connections, reset listeners
        $max = $this->maxConnections;
        $this->maxConnections = 0;
        $listeners = $this->listeners;
        $this->listeners = [];
        // Track disconnects
        $this->onDisconnect(function () use ($max, $listeners) {
            if ($this->getConnectionCount() > 0) {
                return;
            }
            $this->disconnect();
            // Restore settings
            $this->maxConnections = $max;
            $this->listeners = $listeners;
        });
        // Close all current connections, listen to acks
        $this->close($closeStatus);
        $this->start();
    }

    /**
     * Disconnect all connections and stop server.
     */
    public function disconnect(): void
    {
        $this->running = false;
        foreach ($this->connections as $connection) {
            $connection->disconnect();
            $this->dispatch('disconnect', [$this, $connection]);
        }
        $this->connections = [];
        if ($this->server) {
            $this->server->close();
        }
        $this->server = $this->streams = null;
        $this->logger->info('[server] Server disconnected');
    }


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

    // Create socket server
    protected function createSocketServer(): void
    {
        try {
            $uri = new Uri("{$this->scheme}://0.0.0.0:{$this->port}");
            $this->server = $this->streamFactory->createSocketServer($uri, $this->context);
            $this->streams = $this->streamFactory->createStreamCollection();
            $this->streams->attach($this->server, '@server');
            $this->logger->info("[server] Starting server on {$uri}.");
        } catch (Throwable $e) {
            $error = "Server failed to start: {$e->getMessage()}";
            throw new ServerException($error);
        }
    }

    // Accept connection on socket server
    protected function acceptSocket(SocketServer $socket): void
    {
        $connection = null;
        try {
            if (!is_null($this->maxConnections) && $this->getConnectionCount() >= $this->maxConnections) {
                $this->logger->warning("[server] Denied connection, reached max {$this->maxConnections}");
                return;
            }
            $stream = $socket->accept();
            $name = $stream->getRemoteName();
            $this->streams->attach($stream, $name);
            $connection = new Connection($stream, false, true, $this->isSsl());
            $connection->setLogger($this->logger);
            $connection
                ->setFrameSize($this->frameSize)
                ->setTimeout($this->timeout)
                ;
            foreach ($this->middlewares as $middleware) {
                $connection->addMiddleware($middleware);
            }
            $request = $this->performHandshake($connection);
            $this->connections[$name] = $connection;
            $this->logger->info("[server] Accepted connection from {$name}.");
            $this->dispatch('handshake', [
                $this,
                $connection,
                $connection->getHandshakeRequest(),
                $connection->getHandshakeResponse(),
            ]);
            $this->dispatch('connect', [$this, $connection, $request]);
        } catch (Exception | StreamException $e) {
            /** @var Connection|null $connection */
            if (isset($connection)) {
                $connection->disconnect();
            }
            throw new ConnectionFailureException("Server failed to accept: {$e->getMessage()}");
        }
    }

    // Detach connections no longer available
    protected function detachUnconnected(): void
    {
        foreach ($this->connections as $key => $connection) {
            if (!$connection->isConnected()) {
                $this->streams->detach($key);
                unset($this->connections[$key]);
                $this->logger->info("[server] Disconnected {$key}.");
                $this->dispatch('disconnect', [$this, $connection]);
            }
        }
    }

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

        // Read handshake request
        /** @var ServerRequest */
        $request = $connection->pullHttp();

        // Verify handshake request
        try {
            if ($request->getMethod() != 'GET') {
                throw new HandshakeException(
                    "Handshake request with invalid method: '{$request->getMethod()}'",
                    $response->withStatus(405)
                );
            }
            $connectionHeader = trim($request->getHeaderLine('Connection'));
            if (strtolower($connectionHeader) != 'upgrade') {
                throw new HandshakeException(
                    "Handshake request with invalid Connection header: '{$connectionHeader}'",
                    $response->withStatus(426)
                );
            }
            $upgradeHeader = trim($request->getHeaderLine('Upgrade'));
            if (strtolower($upgradeHeader) != 'websocket') {
                throw new HandshakeException(
                    "Handshake request with invalid Upgrade header: '{$upgradeHeader}'",
                    $response->withStatus(426)
                );
            }
            $versionHeader = trim($request->getHeaderLine('Sec-WebSocket-Version'));
            if ($versionHeader != '13') {
                throw new HandshakeException(
                    "Handshake request with invalid Sec-WebSocket-Version header: '{$versionHeader}'",
                    $response->withStatus(426)->withHeader('Sec-WebSocket-Version', '13')
                );
            }
            $keyHeader = trim($request->getHeaderLine('Sec-WebSocket-Key'));
            if (empty($keyHeader)) {
                throw new HandshakeException(
                    "Handshake request with invalid Sec-WebSocket-Key header: '{$keyHeader}'",
                    $response->withStatus(426)
                );
            }
            if (strlen(base64_decode($keyHeader)) != 16) {
                throw new HandshakeException(
                    "Handshake request with invalid Sec-WebSocket-Key header: '{$keyHeader}'",
                    $response->withStatus(426)
                );
            }

            $responseKey = base64_encode(pack('H*', sha1($keyHeader . self::GUID)));
            $response = $response
                ->withHeader('Upgrade', 'websocket')
                ->withHeader('Connection', 'Upgrade')
                ->withHeader('Sec-WebSocket-Accept', $responseKey);
        } catch (HandshakeException $e) {
            $this->logger->warning("[server] {$e->getMessage()}", ['exception' => $e]);
            $response = $e->getResponse();
            $exception = $e;
        }

        // Respond to handshake
        /** @var Response */
        $response = $connection->pushHttp($response);
        if ($response->getStatusCode() != 101) {
            $exception = new HandshakeException("Invalid status code {$response->getStatusCode()}", $response);
        }

        if ($exception) {
            throw $exception;
        }

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

        return $request;
    }
}

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

sirn-se / websocket-php / 13760794555

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