7557012541

Committed 17 Jan 2024 02:15PM UTC coverage: 99.466% (-0.5%) from 100.0%

Build # 7557012541

Build Type

Pull #34

github

Committed by

web-flow

Commit Message

Merge 75fa38531 into b59cf5de3

Pull Request Pull Request #34: Add supported protocols to server and client

Run Details

11 of 16 new or added lines in 2 files covered. (68.75%)

4 existing lines in 1 file now uncovered.

931 of 936 relevant lines covered (99.47%)

42.16 hits per line

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

99.44

/src/Client.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 InvalidArgumentException;
use Phrity\Net\{
    StreamFactory,
    Uri
};
use Psr\Http\Message\UriInterface;
use Psr\Log\{
    LoggerAwareInterface,
    LoggerInterface,
    NullLogger
};
use Stringable;
use Throwable;
use WebSocket\Exception\{
    BadUriException,
    ClientException,
    ConnectionLevelInterface,
    Exception,
    HandshakeException,
    MessageLevelInterface
};
use WebSocket\Http\{
    Request,
    Response
};
use WebSocket\Message\Message;
use WebSocket\Middleware\MiddlewareInterface;
use WebSocket\Trait\{
    ListenerTrait,
    SendMethodsTrait,
    StringableTrait
};

/**
 * WebSocket\Client class.
 * Entry class for WebSocket client.
 */
class Client implements LoggerAwareInterface, Stringable
{
    use ListenerTrait;
    use SendMethodsTrait;
    use StringableTrait;

    // Settings
    private $logger;
    private $timeout = 60;
    private $frameSize = 4096;
    private $persistent = false;
    private $context = [];
    private $headers = [];

    // Internal resources
    private $streamFactory;
    private $socketUri;
    private $connection;
    private $middlewares = [];
    private $streams;
    private $running = false;
    private $subProtocol;


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

    /**
     * @param Psr\Http\Message\UriInterface|string $uri A ws/wss-URI
     */
    public function __construct(UriInterface|string $uri)
    {
        $this->socketUri = $this->parseUri($uri);
        $this->logger = new NullLogger();
        $this->setStreamFactory(new StreamFactory());
    }

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


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

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

    /**
     * Set logger.
     * @param Psr\Log\LoggerInterface $logger Logger implementation
     * @return self.
     */
    public function setLogger(LoggerInterface $logger): self
    {
        $this->logger = $logger;
        if ($this->connection) {
            $this->connection->setLogger($this->logger);
        }
        return $this;
    }

    /**
     * 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;
        if ($this->connection) {
            $this->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 < 1) {
            throw new InvalidArgumentException("Invalid frameSize '{$frameSize}' provided");
        }
        $this->frameSize = $frameSize;
        if ($this->connection) {
            $this->connection->setFrameSize($frameSize);
        }
        return $this;
    }

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

    /**
     * Set connection persistence.
     * @param bool $persistent True for persistent connection.
     * @return self.
     */
    public function setPersistent(bool $persistent): self
    {
        $this->persistent = $persistent;
        return $this;
    }

    /**
     * Set connection context.
     * @param array $context Context as array, see https://www.php.net/manual/en/context.php
     * @return self
     */
    public function setContext(array $context): self
    {
        $this->context = $context;
        return $this;
    }

    /**
     * Add header for handshake.
     * @param string $name Header name
     * @param string $content Header content
     * @return self
     */
    public function addHeader(string $name, string $content): self
    {
        $this->headers[$name] = $content;
        return $this;
    }

    /**
     * Add a middleware.
     * @param WebSocket\Middleware\MiddlewareInterface $middleware
     * @return self
     */
    public function addMiddleware(MiddlewareInterface $middleware): self
    {
        $this->middlewares[] = $middleware;
        if ($this->connection) {
            $this->connection->addMiddleware($middleware);
        }
        return $this;
    }

    /**
     * @return string
     */
    public function getSubProtocol(): string
    {
        return $this->subProtocol;
    }

    /**
     * @param string $subProtocol
     * @return Client
     */
    public function setSubProtocol(string $subProtocol): self
    {
        $this->subProtocol = $subProtocol;

        return $this;
    }

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

    /**
     * Send message.
     * @param Message $message Message to send.
     * @return Message Sent message
     */
    public function send(Message $message): Message
    {
        if (!$this->isConnected()) {
            $this->connect();
        }
        return $this->connection->pushMessage($message);
    }

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


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

    /**
     * Start client listener.
     * @throws Throwable On low level error
     */
    public function start(): void
    {
        // Check if running
        if ($this->running) {
            $this->logger->warning("[client] Client is already running");
            return;
        }
        $this->running = true;
        $this->logger->info("[client] Client is running");

        if (!$this->isConnected()) {
            $this->connect();
        }

        // Run handler
        while ($this->running) {
            try {
                // Get streams with readable content
                $readables = $this->streams->waitRead($this->timeout);
                foreach ($readables as $key => $readable) {
                    try {
                        // Read from connection
                        if ($message = $this->connection->pullMessage()) {
                            $this->dispatch($message->getOpcode(), [$this, $this->connection, $message]);
                        }
                    } catch (MessageLevelInterface $e) {
                        // Error, but keep connection open
                        $this->logger->error("[client] {$e->getMessage()}");
                        $this->dispatch('error', [$this, $this->connection, $e]);
                    } catch (ConnectionLevelInterface $e) {
                        // Error, disconnect connection
                        $this->disconnect();
                        $this->logger->error("[client] {$e->getMessage()}");
                        $this->dispatch('error', [$this, $this->connection, $e]);
                    }
                }
                if (!$this->connection->isConnected()) {
                    $this->running = false;
                }
                $this->connection->tick();
                $this->dispatch('tick', [$this]);
            } catch (Exception $e) {
                $this->disconnect();
                $this->running = false;

                // Low-level error
                $this->logger->error("[client] {$e->getMessage()}");
                $this->dispatch('error', [$this, null, $e]);
            } catch (Throwable $e) {
                $this->disconnect();
                $this->running = false;

                // Crash it
                $this->logger->error("[client] {$e->getMessage()}");
                $this->dispatch('error', [$this, null, $e]);
                throw $e;
            }
            gc_collect_cycles(); // Collect garbage
        }
    }

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

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


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

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

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

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


    /**
     * Connect to server and perform upgrade.
     * @throws ClientException On failed connection
     */
    public function connect(): void
    {
        $this->disconnect();
        $this->streams = $this->streamFactory->createStreamCollection();

        $host_uri = (new Uri())
            ->withScheme($this->socketUri->getScheme() == 'wss' ? 'ssl' : 'tcp')
            ->withHost($this->socketUri->getHost(Uri::IDNA))
            ->withPort($this->socketUri->getPort(Uri::REQUIRE_PORT));

        $stream = null;

        try {
            $client = $this->streamFactory->createSocketClient($host_uri);
            $client->setPersistent($this->persistent);
            $client->setTimeout($this->timeout);
            $client->setContext($this->context);
            $stream = $client->connect();
        } catch (Throwable $e) {
            $error = "Could not open socket to \"{$host_uri}\": {$e->getMessage()}";
            $this->logger->error("[client] {$error}", []);
            throw new ClientException($error);
        }
        $name = $stream->getRemoteName();
        $this->streams->attach($stream, $name);
        $this->connection = new Connection($stream, true, false);
        $this->connection->setFrameSize($this->frameSize);
        $this->connection->setTimeout($this->timeout);
        $this->connection->setLogger($this->logger);
        foreach ($this->middlewares as $middleware) {
            $this->connection->addMiddleware($middleware);
        }

        if (!$this->isConnected()) {
            $error = "Invalid stream on \"{$host_uri}\".";
            $this->logger->error("[client] {$error}");
            throw new ClientException($error);
        }

        if (!$this->persistent || $stream->tell() == 0) {
            $response = $this->performHandshake($host_uri);
        }

        $this->logger->info("[client] Client connected to {$this->socketUri}");
        $this->dispatch('connect', [$this, $this->connection, $response]);
    }

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


    /* ---------- Connection wrapper methods ----------------------------------------------------------------------- */

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

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

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


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

    /**
     * Perform upgrade handshake on new connections.
     * @throws HandshakeException On failed handshake
     */
    protected function performHandshake(Uri $host_uri): Response
    {
        $http_uri = (new Uri())
            ->withPath($this->socketUri->getPath(), Uri::ABSOLUTE_PATH)
            ->withQuery($this->socketUri->getQuery());

        // Generate the WebSocket key.
        $key = $this->generateKey();

        $request = new Request('GET', $http_uri);

        $request = $request
            ->withHeader('Host', $host_uri->getAuthority())
            ->withHeader('User-Agent', 'websocket-client-php')
            ->withHeader('Connection', 'Upgrade')
            ->withHeader('Upgrade', 'websocket')
            ->withHeader('Sec-WebSocket-Key', $key)
            ->withHeader('Sec-WebSocket-Version', '13');

        if ($this->subProtocol) {
            $request = $request->withHeader('Sec-WebSocket-Protocol', $this->subProtocol);
        }

        // Handle basic authentication.
        if ($userinfo = $this->socketUri->getUserInfo()) {
            $request = $request->withHeader('authorization', 'Basic ' . base64_encode($userinfo));
        }

        // Add and override with headers.
        foreach ($this->headers as $name => $content) {
            $request = $request->withHeader($name, $content);
        }

        $this->connection->pushHttp($request);
        $response = $this->connection->pullHttp();

        try {
            if ($response->getStatusCode() != 101) {
                throw new HandshakeException("Invalid status code {$response->getStatusCode()}.", $response);
            }

            if (empty($response->getHeaderLine('Sec-WebSocket-Accept'))) {
                throw new HandshakeException(
                    "Connection to '{$this->socketUri}' failed: Server sent invalid upgrade response.",
                    $response
                );
            }

            $response_key = trim($response->getHeaderLine('Sec-WebSocket-Accept'));
            $expected_key = base64_encode(
                pack('H*', sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'))
            );

            if ($response_key !== $expected_key) {
                throw new HandshakeException("Server sent bad upgrade response.", $response);
            }
        } catch (HandshakeException $e) {
            $this->logger->error("[client] {$e->getMessage()}");
            throw $e;
        }

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

        return $response;
    }

    /**
     * Generate a random string for WebSocket key.
     * @return string Random string
     */
    protected function generateKey(): string
    {
        $key = '';
        for ($i = 0; $i < 16; $i++) {
            $key .= chr(rand(33, 126));
        }
        return base64_encode($key);
    }

    /**
     * Ensure URI insatnce to use in client.
     * @param UriInterface|string $uri A ws/wss-URI
     * @return Uri
     * @throws BadUriException On invalid URI
     */
    protected function parseUri(UriInterface|string $uri): Uri
    {
        if ($uri instanceof Uri) {
            $uri_instance = $uri;
        } elseif ($uri instanceof UriInterface) {
            $uri_instance = new Uri("{$uri}");
        } elseif (is_string($uri)) {
            try {
                $uri_instance = new Uri($uri);
            } catch (InvalidArgumentException $e) {
                throw new BadUriException("Invalid URI '{$uri}' provided.");
            }
        }
        if (!in_array($uri_instance->getScheme(), ['ws', 'wss'])) {
            throw new BadUriException("Invalid URI scheme, must be 'ws' or 'wss'.");
        }
        if (!$uri_instance->getHost()) {
            throw new BadUriException("Invalid URI host.");
        }
        return $uri_instance;
    }
}

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 InvalidArgumentException;
13	use Phrity\Net\{
14	StreamFactory,
15	Uri
16	};
17	use Psr\Http\Message\UriInterface;
18	use Psr\Log\{
19	LoggerAwareInterface,
20	LoggerInterface,
21	NullLogger
22	};
23	use Stringable;
24	use Throwable;
25	use WebSocket\Exception\{
26	BadUriException,
27	ClientException,
28	ConnectionLevelInterface,
29	Exception,
30	HandshakeException,
31	MessageLevelInterface
32	};
33	use WebSocket\Http\{
34	Request,
35	Response
36	};
37	use WebSocket\Message\Message;
38	use WebSocket\Middleware\MiddlewareInterface;
39	use WebSocket\Trait\{
40	ListenerTrait,
41	SendMethodsTrait,
42	StringableTrait
43	};
44
45	/**
46	* WebSocket\Client class.
47	* Entry class for WebSocket client.
48	*/
49	class Client implements LoggerAwareInterface, Stringable
50	{
51	use ListenerTrait;
52	use SendMethodsTrait;
53	use StringableTrait;
54
55	// Settings
56	private $logger;
57	private $timeout = 60;
58	private $frameSize = 4096;
59	private $persistent = false;
60	private $context = [];
61	private $headers = [];
62
63	// Internal resources
64	private $streamFactory;
65	private $socketUri;
66	private $connection;
67	private $middlewares = [];
68	private $streams;
69	private $running = false;
70	private $subProtocol;
71
72
73	/* ---------- Magic methods ------------------------------------------------------------------------------------ */
74
75	/**
76	* @param Psr\Http\Message\UriInterface\|string $uri A ws/wss-URI
77	*/
78	public function __construct(UriInterface\|string $uri)
79	{
80	$this->socketUri = $this->parseUri($uri);	92✔
81	$this->logger = new NullLogger();	86✔
82	$this->setStreamFactory(new StreamFactory());	86✔
83	}
84
85	/**
86	* Get string representation of instance.
87	* @return string String representation
88	*/
89	public function __toString(): string
90	{
91	return $this->stringable('%s', $this->connection ? $this->socketUri->__toString() : 'closed');	2✔
92	}
93
94
95	/* ---------- Configuration ------------------------------------------------------------------------------------ */
96
97	/**
98	* Set stream factory to use.
99	* @param Phrity\Net\StreamFactory $streamFactory
100	* @return self
101	*/
102	public function setStreamFactory(StreamFactory $streamFactory): self
103	{
104	$this->streamFactory = $streamFactory;	86✔
105	return $this;	86✔
106	}
107
108	/**
109	* Set logger.
110	* @param Psr\Log\LoggerInterface $logger Logger implementation
111	* @return self.
112	*/
113	public function setLogger(LoggerInterface $logger): self
114	{
115	$this->logger = $logger;	4✔
116	if ($this->connection) {	4✔
117	$this->connection->setLogger($this->logger);	2✔
118	}
119	return $this;	4✔
120	}
121
122	/**
123	* Set timeout.
124	* @param int $timeout Timeout in seconds
125	* @return self
126	* @throws InvalidArgumentException If invalid timeout provided
127	*/
128	public function setTimeout(int $timeout): self
129	{
130	if ($timeout < 0) {	8✔
131	throw new InvalidArgumentException("Invalid timeout '{$timeout}' provided");	2✔
132	}
133	$this->timeout = $timeout;	6✔
134	if ($this->connection) {	6✔
135	$this->connection->setTimeout($timeout);	2✔
136	}
137	return $this;	6✔
138	}
139
140	/**
141	* Get timeout.
142	* @return int Timeout in seconds
143	*/
144	public function getTimeout(): int
145	{
146	return $this->timeout;	2✔
147	}
148
149	/**
150	* Set frame size.
151	* @param int $frameSize Frame size in bytes
152	* @return self
153	* @throws InvalidArgumentException If invalid frameSize provided
154	*/
155	public function setFrameSize(int $frameSize): self
156	{
157	if ($frameSize < 1) {	12✔
158	throw new InvalidArgumentException("Invalid frameSize '{$frameSize}' provided");	2✔
159	}
160	$this->frameSize = $frameSize;	10✔
161	if ($this->connection) {	10✔
162	$this->connection->setFrameSize($frameSize);	4✔
163	}
164	return $this;	10✔
165	}
166
167	/**
168	* Get frame size.
169	* @return int Frame size in bytes
170	*/
171	public function getFrameSize(): int
172	{
173	return $this->frameSize;	12✔
174	}
175
176	/**
177	* Set connection persistence.
178	* @param bool $persistent True for persistent connection.
179	* @return self.
180	*/
181	public function setPersistent(bool $persistent): self
182	{
183	$this->persistent = $persistent;	2✔
184	return $this;	2✔
185	}
186
187	/**
188	* Set connection context.
189	* @param array $context Context as array, see https://www.php.net/manual/en/context.php
190	* @return self
191	*/
192	public function setContext(array $context): self
193	{
194	$this->context = $context;	2✔
195	return $this;	2✔
196	}
197
198	/**
199	* Add header for handshake.
200	* @param string $name Header name
201	* @param string $content Header content
202	* @return self
203	*/
204	public function addHeader(string $name, string $content): self
205	{
206	$this->headers[$name] = $content;	2✔
207	return $this;	2✔
208	}
209
210	/**
211	* Add a middleware.
212	* @param WebSocket\Middleware\MiddlewareInterface $middleware
213	* @return self
214	*/
215	public function addMiddleware(MiddlewareInterface $middleware): self
216	{
217	$this->middlewares[] = $middleware;	12✔
218	if ($this->connection) {	12✔
219	$this->connection->addMiddleware($middleware);	2✔
220	}
221	return $this;	12✔
222	}
223
224	/**
225	* @return string
226	*/
227	public function getSubProtocol(): string
228	{
NEW 229	return $this->subProtocol;	×
230	}
231
232	/**
233	* @param string $subProtocol
234	* @return Client
235	*/
236	public function setSubProtocol(string $subProtocol): self
237	{
238	$this->subProtocol = $subProtocol;	2✔
239
240	return $this;	2✔
241	}
242
243	/* ---------- Messaging operations ----------------------------------------------------------------------------- */
244
245	/**
246	* Send message.
247	* @param Message $message Message to send.
248	* @return Message Sent message
249	*/
250	public function send(Message $message): Message
251	{
252	if (!$this->isConnected()) {	18✔
253	$this->connect();	2✔
254	}
255	return $this->connection->pushMessage($message);	18✔
256	}
257
258	/**
259	* Receive message.
260	* Note that this operation will block reading.
261	* @return Message\|null
262	*/
263	public function receive(): Message\|null
264	{
265	if (!$this->isConnected()) {	22✔
266	$this->connect();	2✔
267	}
268	return $this->connection->pullMessage();	22✔
269	}
270
271
272	/* ---------- Listener operations ------------------------------------------------------------------------------ */
273
274	/**
275	* Start client listener.
276	* @throws Throwable On low level error
277	*/
278	public function start(): void
279	{
280	// Check if running
281	if ($this->running) {	10✔
282	$this->logger->warning("[client] Client is already running");	2✔
283	return;	2✔
284	}
285	$this->running = true;	10✔
286	$this->logger->info("[client] Client is running");	10✔
287
288	if (!$this->isConnected()) {	10✔
289	$this->connect();	10✔
290	}
291
292	// Run handler
293	while ($this->running) {	10✔
294	try {
295	// Get streams with readable content
296	$readables = $this->streams->waitRead($this->timeout);	8✔
297	foreach ($readables as $key => $readable) {	6✔
298	try {
299	// Read from connection
300	if ($message = $this->connection->pullMessage()) {	6✔
301	$this->dispatch($message->getOpcode(), [$this, $this->connection, $message]);	2✔
302	}
303	} catch (MessageLevelInterface $e) {	6✔
304	// Error, but keep connection open
305	$this->logger->error("[client] {$e->getMessage()}");	2✔
306	$this->dispatch('error', [$this, $this->connection, $e]);	2✔
307	} catch (ConnectionLevelInterface $e) {	4✔
308	// Error, disconnect connection
309	$this->disconnect();	2✔
310	$this->logger->error("[client] {$e->getMessage()}");	2✔
311	$this->dispatch('error', [$this, $this->connection, $e]);	2✔
312	}
313	}
314	if (!$this->connection->isConnected()) {	4✔
315	$this->running = false;	2✔
316	}
317	$this->connection->tick();	4✔
318	$this->dispatch('tick', [$this]);	4✔
319	} catch (Exception $e) {	4✔
320	$this->disconnect();	2✔
321	$this->running = false;	2✔
322
323	// Low-level error
324	$this->logger->error("[client] {$e->getMessage()}");	2✔
325	$this->dispatch('error', [$this, null, $e]);	2✔
326	} catch (Throwable $e) {	2✔
327	$this->disconnect();	2✔
328	$this->running = false;	2✔
329
330	// Crash it
331	$this->logger->error("[client] {$e->getMessage()}");	2✔
332	$this->dispatch('error', [$this, null, $e]);	2✔
333	throw $e;	2✔
334	}
335	gc_collect_cycles(); // Collect garbage	6✔
336	}
337	}
338
339	/**
340	* Stop client listener (resumable).
341	*/
342	public function stop(): void
343	{
344	$this->running = false;	4✔
345	$this->logger->info("[client] Client is stopped");	4✔
346	}
347
348	/**
349	* If client is running (accepting messages).
350	* @return bool
351	*/
352	public function isRunning(): bool
353	{
354	return $this->running;	2✔
355	}
356
357
358	/* ---------- Connection management ---------------------------------------------------------------------------- */
359
360	/**
361	* If Client has active connection.
362	* @return bool True if active connection.
363	*/
364	public function isConnected(): bool
365	{
366	return $this->connection && $this->connection->isConnected();	80✔
367	}
368
369	/**
370	* If Client is readable.
371	* @return bool
372	*/
373	public function isReadable(): bool
374	{
375	return $this->connection && $this->connection->isReadable();	2✔
376	}
377
378	/**
379	* If Client is writable.
380	* @return bool
381	*/
382	public function isWritable(): bool
383	{
384	return $this->connection && $this->connection->isWritable();	2✔
385	}
386
387
388	/**
389	* Connect to server and perform upgrade.
390	* @throws ClientException On failed connection
391	*/
392	public function connect(): void
393	{
394	$this->disconnect();	78✔
395	$this->streams = $this->streamFactory->createStreamCollection();	78✔
396
397	$host_uri = (new Uri())	78✔
398	->withScheme($this->socketUri->getScheme() == 'wss' ? 'ssl' : 'tcp')	78✔
399	->withHost($this->socketUri->getHost(Uri::IDNA))	78✔
400	->withPort($this->socketUri->getPort(Uri::REQUIRE_PORT));	78✔
401
402	$stream = null;	78✔
403
404	try {
405	$client = $this->streamFactory->createSocketClient($host_uri);	78✔
406	$client->setPersistent($this->persistent);	78✔
407	$client->setTimeout($this->timeout);	78✔
408	$client->setContext($this->context);	78✔
409	$stream = $client->connect();	78✔
410	} catch (Throwable $e) {	2✔
411	$error = "Could not open socket to \"{$host_uri}\": {$e->getMessage()}";	2✔
412	$this->logger->error("[client] {$error}", []);	2✔
413	throw new ClientException($error);	2✔
414	}
415	$name = $stream->getRemoteName();	76✔
416	$this->streams->attach($stream, $name);	76✔
417	$this->connection = new Connection($stream, true, false);	76✔
418	$this->connection->setFrameSize($this->frameSize);	76✔
419	$this->connection->setTimeout($this->timeout);	76✔
420	$this->connection->setLogger($this->logger);	76✔
421	foreach ($this->middlewares as $middleware) {	76✔
422	$this->connection->addMiddleware($middleware);	10✔
423	}
424
425	if (!$this->isConnected()) {	76✔
426	$error = "Invalid stream on \"{$host_uri}\".";	2✔
427	$this->logger->error("[client] {$error}");	2✔
428	throw new ClientException($error);	2✔
429	}
430
431	if (!$this->persistent \|\| $stream->tell() == 0) {	74✔
432	$response = $this->performHandshake($host_uri);	74✔
433	}
434
435	$this->logger->info("[client] Client connected to {$this->socketUri}");	66✔
436	$this->dispatch('connect', [$this, $this->connection, $response]);	66✔
437	}
438
439	/**
440	* Disconnect from server.
441	*/
442	public function disconnect(): void
443	{
444	if ($this->isConnected()) {	78✔
445	$this->connection->disconnect();	8✔
446	$this->logger->info('[client] Client disconnected');	8✔
447	$this->dispatch('disconnect', [$this, $this->connection]);	8✔
448	}
449	}
450
451
452	/* ---------- Connection wrapper methods ----------------------------------------------------------------------- */
453
454	/**
455	* Get name of local socket, or null if not connected.
456	* @return string\|null
457	*/
458	public function getName(): string\|null
459	{
460	return $this->isConnected() ? $this->connection->getName() : null;	2✔
461	}
462
463	/**
464	* Get name of remote socket, or null if not connected.
465	* @return string\|null
466	*/
467	public function getRemoteName(): string\|null
468	{
469	return $this->isConnected() ? $this->connection->getRemoteName() : null;	2✔
470	}
471
472	/**
473	* Get Response for handshake procedure.
474	* @return Response\|null Handshake.
475	*/
476	public function getHandshakeResponse(): Response\|null
477	{
478	return $this->connection ? $this->connection->getHandshakeResponse() : null;	2✔
479	}
480
481
482	/* ---------- Internal helper methods -------------------------------------------------------------------------- */
483
484	/**
485	* Perform upgrade handshake on new connections.
486	* @throws HandshakeException On failed handshake
487	*/
488	protected function performHandshake(Uri $host_uri): Response
489	{
490	$http_uri = (new Uri())	74✔
491	->withPath($this->socketUri->getPath(), Uri::ABSOLUTE_PATH)	74✔
492	->withQuery($this->socketUri->getQuery());	74✔
493
494	// Generate the WebSocket key.
495	$key = $this->generateKey();	74✔
496
497	$request = new Request('GET', $http_uri);	74✔
498
499	$request = $request	74✔
500	->withHeader('Host', $host_uri->getAuthority())	74✔
501	->withHeader('User-Agent', 'websocket-client-php')	74✔
502	->withHeader('Connection', 'Upgrade')	74✔
503	->withHeader('Upgrade', 'websocket')	74✔
504	->withHeader('Sec-WebSocket-Key', $key)	74✔
505	->withHeader('Sec-WebSocket-Version', '13');	74✔
506
507	if ($this->subProtocol) {	74✔
508	$request = $request->withHeader('Sec-WebSocket-Protocol', $this->subProtocol);	2✔
509	}
510
511	// Handle basic authentication.
512	if ($userinfo = $this->socketUri->getUserInfo()) {	74✔
513	$request = $request->withHeader('authorization', 'Basic ' . base64_encode($userinfo));	2✔
514	}
515
516	// Add and override with headers.
517	foreach ($this->headers as $name => $content) {	74✔
518	$request = $request->withHeader($name, $content);	2✔
519	}
520
521	$this->connection->pushHttp($request);	74✔
522	$response = $this->connection->pullHttp();	74✔
523
524	try {
525	if ($response->getStatusCode() != 101) {	72✔
526	throw new HandshakeException("Invalid status code {$response->getStatusCode()}.", $response);	2✔
527	}
528
529	if (empty($response->getHeaderLine('Sec-WebSocket-Accept'))) {	70✔
530	throw new HandshakeException(	2✔
531	"Connection to '{$this->socketUri}' failed: Server sent invalid upgrade response.",	2✔
532	$response	2✔
533	);	2✔
534	}
535
536	$response_key = trim($response->getHeaderLine('Sec-WebSocket-Accept'));	68✔
537	$expected_key = base64_encode(	68✔
538	pack('H*', sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'))	68✔
539	);	68✔
540
541	if ($response_key !== $expected_key) {	68✔
542	throw new HandshakeException("Server sent bad upgrade response.", $response);	68✔
543	}
544	} catch (HandshakeException $e) {	6✔
545	$this->logger->error("[client] {$e->getMessage()}");	6✔
546	throw $e;	6✔
547	}
548
549	$this->logger->debug("[client] Handshake on {$http_uri->getPath()}");	66✔
550	$this->connection->setHandshakeRequest($request);	66✔
551	$this->connection->setHandshakeResponse($response);	66✔
552
553	return $response;	66✔
554	}
555
556	/**
557	* Generate a random string for WebSocket key.
558	* @return string Random string
559	*/
560	protected function generateKey(): string
561	{
562	$key = '';	74✔
563	for ($i = 0; $i < 16; $i++) {	74✔
564	$key .= chr(rand(33, 126));	74✔
565	}
566	return base64_encode($key);	74✔
567	}
568
569	/**
570	* Ensure URI insatnce to use in client.
571	* @param UriInterface\|string $uri A ws/wss-URI
572	* @return Uri
573	* @throws BadUriException On invalid URI
574	*/
575	protected function parseUri(UriInterface\|string $uri): Uri
576	{
577	if ($uri instanceof Uri) {	92✔
578	$uri_instance = $uri;	6✔
579	} elseif ($uri instanceof UriInterface) {	86✔
580	$uri_instance = new Uri("{$uri}");	2✔
581	} elseif (is_string($uri)) {	84✔
582	try {
583	$uri_instance = new Uri($uri);	84✔
584	} catch (InvalidArgumentException $e) {	2✔
585	throw new BadUriException("Invalid URI '{$uri}' provided.");	2✔
586	}
587	}
588	if (!in_array($uri_instance->getScheme(), ['ws', 'wss'])) {	90✔
589	throw new BadUriException("Invalid URI scheme, must be 'ws' or 'wss'.");	2✔
590	}
591	if (!$uri_instance->getHost()) {	88✔
592	throw new BadUriException("Invalid URI host.");	2✔
593	}
594	return $uri_instance;	86✔
595	}
596	}

sirn-se / websocket-php / 7557012541

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