WebSocketServer::buildMessageProcessor() - Code Metrics - Inspection of "WIP - Add on client disconnect event to MessageHan..." - Nekland/Woketo - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#144)

unknown

created 2018-09-20 20:23 UTC

WebSocketServer::buildMessageProcessor() A

↳ Parent: WebSocketServer

Complexity

Conditions	3
Paths	3

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
dl	0
loc	18
ccs	0
cts	17
cp	0
rs	9.6666
c	0
b	0
f	0
cc	3
nc	3
nop	0
crap	12

<?php

/**
 * This file is a part of Woketo package.
 *
 * (c) Nekland <[email protected]>
 *
 * For the full license, take a look to the LICENSE file
 * on the root directory of this project
 */

namespace Nekland\Woketo\Server;

use Nekland\Woketo\Exception\ConfigException;
use Nekland\Woketo\Exception\RuntimeException;
use Nekland\Woketo\Http\CloseEventDescriptor;
use Nekland\Woketo\Message\MessageHandlerInterface;
use Nekland\Woketo\Rfc6455\FrameFactory;
use Nekland\Woketo\Rfc6455\Handshake\ServerHandshake;
use Nekland\Woketo\Rfc6455\MessageFactory;
use Nekland\Woketo\Rfc6455\FrameHandler\CloseFrameHandler;
use Nekland\Woketo\Rfc6455\FrameHandler\RsvCheckFrameHandler;
use Nekland\Woketo\Rfc6455\FrameHandler\WrongOpcodeFrameHandler;
use Nekland\Woketo\Rfc6455\FrameHandler\PingFrameHandler;
use Nekland\Woketo\Rfc6455\MessageProcessor;
use Nekland\Woketo\Utils\SimpleLogger;
use Psr\Log\LoggerInterface;
use Psr\Log\LogLevel;
use React\EventLoop\LoopInterface;
use React\Socket\ConnectionInterface;
use React\Socket\ServerInterface;

class WebSocketServer
{
    /**
     * @var int
     */
    private $port;

    /**
     * @var string
     */
    private $host;

    /**
     * @var ServerHandshake
     */
    private $handshake;

    /**
     * @var MessageHandlerInterface[]
     */
    private $messageHandlers;

    /**
     * @var Connection[]
     */
    private $connections;

    /**
     * @var LoopInterface
     */
    private $loop;

    /**
     * @var ServerInterface
     */
    private $server;

    /**
     * @var MessageProcessor
     */
    private $messageProcessor;

    /**
     * @var array
     */
    private $config;

    /**
     * @var LoggerInterface
     */
    private $logger;

    /**
     * @var CloseEventDescriptor
     */
    private $statusCodeDescriptor;

    /**
     * @param int $port The number of the port to bind
     * @param string $host The host to listen on (by default 127.0.0.1)
     * @param array $config
     */
    public function __construct($port, $host = '127.0.0.1', $config = [])
    {
        $this->setConfig($config);
        $this->host = $host;
        $this->port = $port;
        $this->handshake = new ServerHandshake();
        $this->connections = [];
        $this->buildMessageProcessor();

        // Some optimization
        \gc_enable();       // As the process never stops, the garbage collector will be usefull, you may need to call it manually sometimes for performance purpose
        \set_time_limit(0); // It's by default on most server for cli apps but better be sure of that fact
    }

    /**
     * @param MessageHandlerInterface|string $messageHandler An instance of a class as string
     * @param string                         $uri            The URI you want to bind on
     */
    public function setMessageHandler($messageHandler, $uri = '*')
    {
        if (!$messageHandler instanceof MessageHandlerInterface &&  !\is_string($messageHandler)) {
            throw new \InvalidArgumentException('The message handler must be an instance of MessageHandlerInterface or a string.');
        }
        if (\is_string($messageHandler)) {
            try {
                $reflection = new \ReflectionClass($messageHandler);
                if(!$reflection->implementsInterface('Nekland\Woketo\Message\MessageHandlerInterface')) {
                    throw new \InvalidArgumentException('The messageHandler must implement MessageHandlerInterface');
                }
            } catch (\ReflectionException $e) {
                throw new \InvalidArgumentException('The messageHandler must be a string representing a class.');
            }
        }
        $this->messageHandlers[$uri] = $messageHandler;
    }

    /**
     * Launch the WebSocket server and an infinite loop that act on event.
     *
     * @throws \Exception
     */
    public function start()
    {
        if ($this->config['prod'] && \extension_loaded('xdebug')) {
            throw new \Exception('xdebug is enabled, it\'s a performance issue. Disable that extension or specify "prod" option to false.');
        }

        $this->loop = $this->loop ?? \React\EventLoop\Factory::create();
        $this->server = $this->server ?? new \React\Socket\TcpServer($this->host . ':' . $this->port, $this->loop);

        if ($this->config['ssl']) {
            $this->server = new \React\Socket\SecureServer($this->server, $this->loop, array_merge([
                'local_cert' => $this->config['certFile'],
                'passphrase' => $this->config['passphrase'],
            ], $this->config['sslContextOptions']));
            $this->getLogger()->info('Enabled ssl');
        }

        $this->server->on('connection', function (ConnectionInterface $socketStream) {
            $this->onNewConnection($socketStream);
        });

        $this->getLogger()->info('Listening on ' . $this->host . ':' . $this->port);

        $this->loop->run();
    }

    /**
     * @param ConnectionInterface $socketStream
     */
    private function onNewConnection(ConnectionInterface $socketStream)
    {
        $connection = new Connection($socketStream, function ($uri, Connection $connection) {
            return $this->getMessageHandler($uri, $connection);
        }, $this->loop, $this->messageProcessor);

        $socketStream->on('end', function () use($connection) {
            $this->onDisconnect($connection);
        });

        $connection->setLogger($this->getLogger());
        $connection->getLogger()->info(sprintf('Ip "%s" establish connection', $connection->getIp()));
        $this->connections[] = $connection;
    }

    /**
     *
     * @param Connection $connection
     */
    private function onDisconnect(Connection $connection)
    {
        $this->removeConnection($connection);
        $connection->getLogger()->info(sprintf('Ip "%s" left connection', $connection->getIp()));

        unset($connection);
    }

    /**
     * Remove a Connection instance by his object id
     * @param Connection $connection
     * @param bool $strict Define if the process should stop if unfindable
     * @throws RuntimeException This method throw an exception if the $connection instance object isn't findable in websocket server's connections
     */
    private function removeConnection(Connection $connection, bool $strict = true)
    {
        $connectionId = spl_object_hash($connection);
        foreach ($this->connections as $index => $connectionIt) {
            if ($connectionId === spl_object_hash($connectionIt)) {
                unset($this->connections[$index]);
                return;
            }
        }

        $this->logger->critical(sprintf('Connection instance with id "%s" is unfindable in websocket server\'s active connections !', $connectionId));
        if ($strict) {
            throw new RuntimeException(sprintf('Impossible to find the connection with id "%d" in the running server.', $connectionId));
        }
    }

    /**
     * @param string $uri
     * @param Connection $connection
     * @return MessageHandlerInterface|null
     */
    private function getMessageHandler(string $uri, Connection $connection)

    {
        $handler = null;

        if (!empty($this->messageHandlers[$uri])) {
            $handler = $this->messageHandlers[$uri];
        }

        if (null === $handler && !empty($this->messageHandlers['*'])) {
            $handler = $this->messageHandlers['*'];
        }

        if (null !== $handler) {
            if (\is_string($handler)) {
                $handler = new $handler;
            }

            return $handler;
        }

        $this->logger->warning('Connection on ' . $uri . ' but no handler found.');
        return null;
    }

    /**
     * Build the message processor with configuration
     */
    private function buildMessageProcessor()
    {
        $this->messageProcessor = new MessageProcessor(
            false,
            new FrameFactory($this->config['frame']),
            new MessageFactory($this->config['message'])
        );
        $this->messageProcessor->addHandler(new PingFrameHandler());
        $this->messageProcessor->addHandler(new CloseFrameHandler());
        $this->messageProcessor->addHandler(new WrongOpcodeFrameHandler());
        $this->messageProcessor->addHandler(new RsvCheckFrameHandler());

        foreach ($this->config['messageHandlers'] as $handler) {
            if (!$handler instanceof MessageHandlerInterface) {
                throw new RuntimeException(sprintf('%s is not an instance of MessageHandlerInterface but must be !', get_class($handler)));
            }
        }
    }

    /**
     * Sets the configuration
     *
     * @param array $config
     * @throws ConfigException
     */
    private function setConfig(array $config)
    {
        $this->config = \array_merge([
            'frame' => [],
            'message' => [],
            'messageHandlers' => [],
            'prod' => true,
            'ssl' => false,
            'certFile' => '',
            'passphrase' => '',
            'sslContextOptions' => [],
        ], $config);

        if ($this->config['ssl'] && !is_file($this->config['certFile'])) {
            throw new ConfigException('With ssl configuration, you need to specify a certificate file.');
        }
    }

    /**
     * @return SimpleLogger|LoggerInterface
     */
    public function getLogger()
    {
        if (null === $this->logger) {
            return $this->logger = new SimpleLogger(!$this->config['prod']);
        }

        return $this->logger;
    }

    /**
     * Allows you to set a custom logger
     *
     * @param LoggerInterface $logger
     * @return WebSocketServer
     */
    public function setLogger(LoggerInterface $logger)
    {
        $this->logger = $logger;

        return $this;
    }

    /**
     * Allows to specify a loop that will be used instead of the reactphp generated loop.
     *
     * @param LoopInterface $loop
     * @return WebSocketServer
     */
    public function setLoop(LoopInterface $loop)
    {
        $this->loop = $loop;

        return $this;
    }

    /**
     * @param ServerInterface $server
     * @return WebSocketServer
     */
    public function setSocketServer(ServerInterface $server)
    {
        $this->server = $server;

        return $this;
    }
}


1			<?php
2
3			/**
4			* This file is a part of Woketo package.
5			*
6			* (c) Nekland <[email protected]>
7			*
8			* For the full license, take a look to the LICENSE file
9			* on the root directory of this project
10			*/
11
12			namespace Nekland\Woketo\Server;
13
14			use Nekland\Woketo\Exception\ConfigException;
15			use Nekland\Woketo\Exception\RuntimeException;
16			use Nekland\Woketo\Http\CloseEventDescriptor;
17			use Nekland\Woketo\Message\MessageHandlerInterface;
18			use Nekland\Woketo\Rfc6455\FrameFactory;
19			use Nekland\Woketo\Rfc6455\Handshake\ServerHandshake;
20			use Nekland\Woketo\Rfc6455\MessageFactory;
21			use Nekland\Woketo\Rfc6455\FrameHandler\CloseFrameHandler;
22			use Nekland\Woketo\Rfc6455\FrameHandler\RsvCheckFrameHandler;
23			use Nekland\Woketo\Rfc6455\FrameHandler\WrongOpcodeFrameHandler;
24			use Nekland\Woketo\Rfc6455\FrameHandler\PingFrameHandler;
25			use Nekland\Woketo\Rfc6455\MessageProcessor;
26			use Nekland\Woketo\Utils\SimpleLogger;
27			use Psr\Log\LoggerInterface;
28			use Psr\Log\LogLevel;
29			use React\EventLoop\LoopInterface;
30			use React\Socket\ConnectionInterface;
31			use React\Socket\ServerInterface;
32
33			class WebSocketServer
34			{
35			/**
36			* @var int
37			*/
38			private $port;
39
40			/**
41			* @var string
42			*/
43			private $host;
44
45			/**
46			* @var ServerHandshake
47			*/
48			private $handshake;
49
50			/**
51			* @var MessageHandlerInterface[]
52			*/
53			private $messageHandlers;
54
55			/**
56			* @var Connection[]
57			*/
58			private $connections;
59
60			/**
61			* @var LoopInterface
62			*/
63			private $loop;
64
65			/**
66			* @var ServerInterface
67			*/
68			private $server;
69
70			/**
71			* @var MessageProcessor
72			*/
73			private $messageProcessor;
74
75			/**
76			* @var array
77			*/
78			private $config;
79
80			/**
81			* @var LoggerInterface
82			*/
83			private $logger;
84
85			/**
86			* @var CloseEventDescriptor
87			*/
88			private $statusCodeDescriptor;
89
90			/**
91			* @param int $port The number of the port to bind
92			* @param string $host The host to listen on (by default 127.0.0.1)
93			* @param array $config
94			*/
95			public function __construct($port, $host = '127.0.0.1', $config = [])
96			{
97			$this->setConfig($config);
98			$this->host = $host;
99			$this->port = $port;
100			$this->handshake = new ServerHandshake();
101			$this->connections = [];
102			$this->buildMessageProcessor();
103
104			// Some optimization
105			\gc_enable(); // As the process never stops, the garbage collector will be usefull, you may need to call it manually sometimes for performance purpose
106			\set_time_limit(0); // It's by default on most server for cli apps but better be sure of that fact
107			}
108
109			/**
110			* @param MessageHandlerInterface\|string $messageHandler An instance of a class as string
111			* @param string $uri The URI you want to bind on
112			*/
113			public function setMessageHandler($messageHandler, $uri = '*')
114			{
115			if (!$messageHandler instanceof MessageHandlerInterface && !\is_string($messageHandler)) {
116			throw new \InvalidArgumentException('The message handler must be an instance of MessageHandlerInterface or a string.');
117			}
118			if (\is_string($messageHandler)) {
119			try {
120			$reflection = new \ReflectionClass($messageHandler);
121			if(!$reflection->implementsInterface('Nekland\Woketo\Message\MessageHandlerInterface')) {
122			throw new \InvalidArgumentException('The messageHandler must implement MessageHandlerInterface');
123			}
124			} catch (\ReflectionException $e) {
125			throw new \InvalidArgumentException('The messageHandler must be a string representing a class.');
126			}
127			}
128			$this->messageHandlers[$uri] = $messageHandler;
129			}
130
131			/**
132			* Launch the WebSocket server and an infinite loop that act on event.
133			*
134			* @throws \Exception
135			*/
136			public function start()
137			{
138			if ($this->config['prod'] && \extension_loaded('xdebug')) {
139			throw new \Exception('xdebug is enabled, it\'s a performance issue. Disable that extension or specify "prod" option to false.');
140			}
141
142			$this->loop = $this->loop ?? \React\EventLoop\Factory::create();
143			$this->server = $this->server ?? new \React\Socket\TcpServer($this->host . ':' . $this->port, $this->loop);
144
145			if ($this->config['ssl']) {
146			$this->server = new \React\Socket\SecureServer($this->server, $this->loop, array_merge([
147			'local_cert' => $this->config['certFile'],
148			'passphrase' => $this->config['passphrase'],
149			], $this->config['sslContextOptions']));
150			$this->getLogger()->info('Enabled ssl');
151			}
152
153			$this->server->on('connection', function (ConnectionInterface $socketStream) {
154			$this->onNewConnection($socketStream);
155			});
156
157			$this->getLogger()->info('Listening on ' . $this->host . ':' . $this->port);
158
159			$this->loop->run();
160			}
161
162			/**
163			* @param ConnectionInterface $socketStream
164			*/
165			private function onNewConnection(ConnectionInterface $socketStream)
166			{
167			$connection = new Connection($socketStream, function ($uri, Connection $connection) {
168			return $this->getMessageHandler($uri, $connection);
169			}, $this->loop, $this->messageProcessor);
170
171			$socketStream->on('end', function () use($connection) {
172			$this->onDisconnect($connection);
173			});
174
175			$connection->setLogger($this->getLogger());
176			$connection->getLogger()->info(sprintf('Ip "%s" establish connection', $connection->getIp()));
177			$this->connections[] = $connection;
178			}
179
180			/**
181			*
182			* @param Connection $connection
183			*/
184			private function onDisconnect(Connection $connection)
185			{
186			$this->removeConnection($connection);
187			$connection->getLogger()->info(sprintf('Ip "%s" left connection', $connection->getIp()));
188
189			unset($connection);
190			}
191
192			/**
193			* Remove a Connection instance by his object id
194			* @param Connection $connection
195			* @param bool $strict Define if the process should stop if unfindable
196			* @throws RuntimeException This method throw an exception if the $connection instance object isn't findable in websocket server's connections
197			*/
198			private function removeConnection(Connection $connection, bool $strict = true)
199			{
200			$connectionId = spl_object_hash($connection);
201			foreach ($this->connections as $index => $connectionIt) {
202			if ($connectionId === spl_object_hash($connectionIt)) {
203			unset($this->connections[$index]);
204			return;
205			}
206			}
207
208			$this->logger->critical(sprintf('Connection instance with id "%s" is unfindable in websocket server\'s active connections !', $connectionId));
209			if ($strict) {
210			throw new RuntimeException(sprintf('Impossible to find the connection with id "%d" in the running server.', $connectionId));
211			}
212			}
213
214			/**
215			* @param string $uri
216			* @param Connection $connection
217			* @return MessageHandlerInterface\|null
218			*/
219			private function getMessageHandler(string $uri, Connection $connection)
			0 ignored issues – show Unused Code introduced 2017-08-02 15:20 UTC by Report Bug Copy Issue Report The parameter `$connection` is not used and could be removed. This check looks from parameters that have been defined for a function or method, but which are not used in the method body. Loading history...
220			{
221			$handler = null;
222
223			if (!empty($this->messageHandlers[$uri])) {
224			$handler = $this->messageHandlers[$uri];
225			}
226
227			if (null === $handler && !empty($this->messageHandlers['*'])) {
228			$handler = $this->messageHandlers['*'];
229			}
230
231			if (null !== $handler) {
232			if (\is_string($handler)) {
233			$handler = new $handler;
234			}
235
236			return $handler;
237			}
238
239			$this->logger->warning('Connection on ' . $uri . ' but no handler found.');
240			return null;
241			}
242
243			/**
244			* Build the message processor with configuration
245			*/
246			private function buildMessageProcessor()
247			{
248			$this->messageProcessor = new MessageProcessor(
249			false,
250			new FrameFactory($this->config['frame']),
251			new MessageFactory($this->config['message'])
252			);
253			$this->messageProcessor->addHandler(new PingFrameHandler());
254			$this->messageProcessor->addHandler(new CloseFrameHandler());
255			$this->messageProcessor->addHandler(new WrongOpcodeFrameHandler());
256			$this->messageProcessor->addHandler(new RsvCheckFrameHandler());
257
258			foreach ($this->config['messageHandlers'] as $handler) {
259			if (!$handler instanceof MessageHandlerInterface) {
260			throw new RuntimeException(sprintf('%s is not an instance of MessageHandlerInterface but must be !', get_class($handler)));
261			}
262			}
263			}
264
265			/**
266			* Sets the configuration
267			*
268			* @param array $config
269			* @throws ConfigException
270			*/
271			private function setConfig(array $config)
272			{
273			$this->config = \array_merge([
274			'frame' => [],
275			'message' => [],
276			'messageHandlers' => [],
277			'prod' => true,
278			'ssl' => false,
279			'certFile' => '',
280			'passphrase' => '',
281			'sslContextOptions' => [],
282			], $config);
283
284			if ($this->config['ssl'] && !is_file($this->config['certFile'])) {
285			throw new ConfigException('With ssl configuration, you need to specify a certificate file.');
286			}
287			}
288
289			/**
290			* @return SimpleLogger\|LoggerInterface
291			*/
292			public function getLogger()
293			{
294			if (null === $this->logger) {
295			return $this->logger = new SimpleLogger(!$this->config['prod']);
296			}
297
298			return $this->logger;
299			}
300
301			/**
302			* Allows you to set a custom logger
303			*
304			* @param LoggerInterface $logger
305			* @return WebSocketServer
306			*/
307			public function setLogger(LoggerInterface $logger)
308			{
309			$this->logger = $logger;
310
311			return $this;
312			}
313
314			/**
315			* Allows to specify a loop that will be used instead of the reactphp generated loop.
316			*
317			* @param LoopInterface $loop
318			* @return WebSocketServer
319			*/
320			public function setLoop(LoopInterface $loop)
321			{
322			$this->loop = $loop;
323
324			return $this;
325			}
326
327			/**
328			* @param ServerInterface $server
329			* @return WebSocketServer
330			*/
331			public function setSocketServer(ServerInterface $server)
332			{
333			$this->server = $server;
334
335			return $this;
336			}
337			}
338

Nekland / Woketo

Pull Request — master (#144)

WebSocketServer::buildMessageProcessor() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like