Connection::processHandcheck() - Code Metrics - Inspection of "Woketo Client" - Nekland/Woketo - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#100)

by Maxime

created 2017-02-14 00:53 UTC

Connection::processHandcheck() A

↳ Parent: Project

Complexity

Conditions	2
Paths	2

Size

Total Lines	16
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	11
CRAP Score	2.0023

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
dl	0
loc	16
ccs	11
cts	12
cp	0.9167
rs	9.4285
c	1
b	0
f	0
cc	2
eloc	11
nc	2
nop	1
crap	2.0023

<?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\Core\AbstractConnection;
use Nekland\Woketo\Exception\NoHandlerException;
use Nekland\Woketo\Exception\RuntimeException;
use Nekland\Woketo\Exception\WebsocketException;
use Nekland\Woketo\Http\Request;
use Nekland\Woketo\Http\Response;
use Nekland\Woketo\Message\MessageHandlerInterface;
use Nekland\Woketo\Rfc6455\Frame;
use Nekland\Woketo\Rfc6455\Message;
use Nekland\Woketo\Rfc6455\MessageProcessor;
use Nekland\Woketo\Rfc6455\Handshake\ServerHandshake;
use React\EventLoop\LoopInterface;
use React\EventLoop\Timer\TimerInterface;
use React\Socket\ConnectionInterface;

class Connection extends AbstractConnection
{
    /**
     * 5 seconds
     */
    const DEFAULT_TIMEOUT = 5;

    /**
     * @var Message
     */
    private $currentMessage;

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

    /**
     * @var TimerInterface
     */
    private $timeout;

    public function __construct(
        ConnectionInterface $socketStream,
        \Closure $messageHandler,
        LoopInterface $loop,
        MessageProcessor $messageProcessor,
        ServerHandshake $handshake = null
    ) {
        parent::__construct($messageProcessor, $handshake ?: new ServerHandshake);
        $this->stream = $socketStream;
class Alien {}

class Dalek extends Alien {}

class Plot
{
    /** @var  Dalek */
    public $villain;
}

$alien = new Alien();
$plot = new Plot();
if ($alien instanceof Dalek) {
    $plot->villain = $alien;
}
        $this->initListeners();
        $this->handler = $messageHandler;
        $this->loop = $loop;
    }

    private function initListeners()
    {
        $this->stream->on('data', function ($data) {
            $this->processData($data);
        });
        $this->stream->on('error', function ($data) {
            $this->error($data);
        });
    }

    private function processData($data)
    {
        try {
            if (!$this->handshakeDone) {
                $this->processHandshake($data);
            } else {
                $this->processMessage($data);
            }

            return;
        } catch (WebsocketException $e) {
            $this->messageProcessor->close($this->stream);
            $this->logger->notice('Connection to ' . $this->getIp() . ' closed with error : ' . $e->getMessage());
            $this->getHandler()->onError($e, $this);
        } catch (NoHandlerException $e) {
            $this->getLogger()->info(sprintf('No handler found for uri %s. Connection closed.', $this->uri));
            $this->close();
        }
    }

    /**
     * This method build a message and buffer data in case of incomplete data.
     *
     * @param string $data
     */
    protected function processMessage(string $data)
    {
        // It may be a timeout going (we were waiting for data), let's clear it.
        if ($this->timeout !== null) {
            $this->timeout->cancel();
            $this->timeout = null;
        }

        foreach ($this->messageProcessor->onData($data, $this->stream, $this->currentMessage) as $message) {
            $this->currentMessage = $message;
            if ($this->currentMessage->isComplete()) {
                // Sending the message through the woketo API.
                switch($this->currentMessage->getOpcode()) {
                    case Frame::OP_TEXT:
                        $this->getHandler()->onMessage($this->currentMessage->getContent(), $this);
                        break;
                    case Frame::OP_BINARY:
                        $this->getHandler()->onBinary($this->currentMessage->getContent(), $this);
                        break;
                }
                $this->currentMessage = null;

            } else {
                // We wait for more data so we start a timeout.
                $this->timeout = $this->loop->addTimer(Connection::DEFAULT_TIMEOUT, function () {
                    $this->logger->notice('Connection to ' . $this->getIp() . ' timed out.');
                    $this->messageProcessor->timeout($this->stream);
                });
            }
        }
    }

    /**
     * @param string|Frame $frame
     * @param int          $opCode An int representing binary or text data (const of Frame class)
     * @throws \Nekland\Woketo\Exception\RuntimeException
     */
    public function write($frame, int $opCode = Frame::OP_TEXT)
    {
        try {
            $this->messageProcessor->write($frame, $this->stream, $opCode);
        } catch (WebsocketException $e) {
            throw new RuntimeException($e);
        }
    }

    /**
     * @param mixed $data
     */
    protected function error($data)
    {
        $message = "A connectivity error occurred: " . $data;
        $this->logger->error($message);
        $this->getHandler()->onError(new WebsocketException($message), $this);
    }

    /**
     * If it's a new client, we need to make some special actions named the handshake.
     *
     * @param string $data
     */
    protected function processHandshake(string $data)
    {
        if ($this->handshakeDone) {
            return;
        }

        $request = Request::create($data);
        $this->handshake->verify($request);
        $this->uri = $request->getUri();
        $response = Response::createSwitchProtocolResponse();
        $this->handshake->sign($request, $response);
        $response->send($this->stream);
        
        $this->handshakeDone = true;
        $this->getHandler()->onConnection($this);
    }

    /**
     * Close the connection with normal close.
     */
    public function close()
    {
        $this->messageProcessor->close($this->stream);
    }

    /**
     * @return MessageHandlerInterface
     * @throws NoHandlerException
     */
    private function getHandler() : MessageHandlerInterface
    {
        if ($this->handler instanceof \Closure) {
            $handler = $this->handler;
            $handler = $handler($this->uri, $this);

            if (null === $handler) {
                throw new NoHandlerException(sprintf('No handler for request URI %s.', $this->uri));
            }

            $this->handler = $handler;
        }

        return $this->handler;
    }
}


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\Core\AbstractConnection;
15		use Nekland\Woketo\Exception\NoHandlerException;
16		use Nekland\Woketo\Exception\RuntimeException;
17		use Nekland\Woketo\Exception\WebsocketException;
18		use Nekland\Woketo\Http\Request;
19		use Nekland\Woketo\Http\Response;
20		use Nekland\Woketo\Message\MessageHandlerInterface;
21		use Nekland\Woketo\Rfc6455\Frame;
22		use Nekland\Woketo\Rfc6455\Message;
23		use Nekland\Woketo\Rfc6455\MessageProcessor;
24		use Nekland\Woketo\Rfc6455\Handshake\ServerHandshake;
25		use React\EventLoop\LoopInterface;
26		use React\EventLoop\Timer\TimerInterface;
27		use React\Socket\ConnectionInterface;
28
29		class Connection extends AbstractConnection
30		{
31		/**
32		* 5 seconds
33		*/
34		const DEFAULT_TIMEOUT = 5;
35
36		/**
37		* @var Message
38		*/
39		private $currentMessage;
40
41		/**
42		* @var LoopInterface
43		*/
44		private $loop;
45
46		/**
47		* @var TimerInterface
48		*/
49		private $timeout;
50
51	5	public function __construct(
52		ConnectionInterface $socketStream,
53		\Closure $messageHandler,
54		LoopInterface $loop,
55		MessageProcessor $messageProcessor,
56		ServerHandshake $handshake = null
57		) {
58	5	parent::__construct($messageProcessor, $handshake ?: new ServerHandshake);
59	5	$this->stream = $socketStream;
		0 ignored issues – show Documentation Bug introduced 2017-01-06 22:26 UTC by Report Bug Copy Issue Report `$socketStream` is of type `object<React\Socket\ConnectionInterface>`, but the property `$stream` was declared to be of type `object<React\Socket\Connection>`. Are you sure that you always receive this specific sub-class here, or does it make sense to add an `instanceof` check? Our type inference engine has found a suspicous assignment of a value to a property. This check raises an issue when a value that can be of a given class or a super-class is assigned to a property that is type hinted more strictly. Either this assignment is in error or an instanceof check should be added for that assignment. class Alien {} class Dalek extends Alien {} class Plot { /** @var Dalek */ public $villain; } $alien = new Alien(); $plot = new Plot(); if ($alien instanceof Dalek) { $plot->villain = $alien; } Loading history...
60	5	$this->initListeners();
61	5	$this->handler = $messageHandler;
62	5	$this->loop = $loop;
63	5	}
64
65	5	private function initListeners()
66		{
67		$this->stream->on('data', function ($data) {
68	5	$this->processData($data);
69	5	});
70		$this->stream->on('error', function ($data) {
71		$this->error($data);
72	5	});
73	5	}
74
75	5	private function processData($data)
76		{
77		try {
78	5	if (!$this->handshakeDone) {
79	5	$this->processHandshake($data);
80		} else {
81	2	$this->processMessage($data);
82		}
83
84	4	return;
85	1	} catch (WebsocketException $e) {
86		$this->messageProcessor->close($this->stream);
87		$this->logger->notice('Connection to ' . $this->getIp() . ' closed with error : ' . $e->getMessage());
88		$this->getHandler()->onError($e, $this);
89	1	} catch (NoHandlerException $e) {
90	1	$this->getLogger()->info(sprintf('No handler found for uri %s. Connection closed.', $this->uri));
91	1	$this->close();
92		}
93	1	}
94
95		/**
96		* This method build a message and buffer data in case of incomplete data.
97		*
98		* @param string $data
99		*/
100	2	protected function processMessage(string $data)
101		{
102		// It may be a timeout going (we were waiting for data), let's clear it.
103	2	if ($this->timeout !== null) {
104		$this->timeout->cancel();
105		$this->timeout = null;
106		}
107
108	2	foreach ($this->messageProcessor->onData($data, $this->stream, $this->currentMessage) as $message) {
109	2	$this->currentMessage = $message;
110	2	if ($this->currentMessage->isComplete()) {
111		// Sending the message through the woketo API.
112	2	switch($this->currentMessage->getOpcode()) {
113	2	case Frame::OP_TEXT:
114	1	$this->getHandler()->onMessage($this->currentMessage->getContent(), $this);
115	1	break;
116	1	case Frame::OP_BINARY:
117	1	$this->getHandler()->onBinary($this->currentMessage->getContent(), $this);
118	1	break;
119		}
120	2	$this->currentMessage = null;
121
122		} else {
123		// We wait for more data so we start a timeout.
124	2	$this->timeout = $this->loop->addTimer(Connection::DEFAULT_TIMEOUT, function () {
125		$this->logger->notice('Connection to ' . $this->getIp() . ' timed out.');
126		$this->messageProcessor->timeout($this->stream);
127	2	});
128		}
129		}
130	2	}
131
132		/**
133		* @param string\|Frame $frame
134		* @param int $opCode An int representing binary or text data (const of Frame class)
135		* @throws \Nekland\Woketo\Exception\RuntimeException
136		*/
137		public function write($frame, int $opCode = Frame::OP_TEXT)
138		{
139		try {
140		$this->messageProcessor->write($frame, $this->stream, $opCode);
141		} catch (WebsocketException $e) {
142		throw new RuntimeException($e);
143		}
144		}
145
146		/**
147		* @param mixed $data
148		*/
149		protected function error($data)
150		{
151		$message = "A connectivity error occurred: " . $data;
152		$this->logger->error($message);
153		$this->getHandler()->onError(new WebsocketException($message), $this);
154		}
155
156		/**
157		* If it's a new client, we need to make some special actions named the handshake.
158		*
159		* @param string $data
160		*/
161	5	protected function processHandshake(string $data)
162		{
163	5	if ($this->handshakeDone) {
164		return;
165		}
166
167	5	$request = Request::create($data);
168	5	$this->handshake->verify($request);
169	5	$this->uri = $request->getUri();
170	5	$response = Response::createSwitchProtocolResponse();
171	5	$this->handshake->sign($request, $response);
172	5	$response->send($this->stream);
173
174	5	$this->handshakeDone = true;
175	5	$this->getHandler()->onConnection($this);
176	4	}
177
178		/**
179		* Close the connection with normal close.
180		*/
181	1	public function close()
182		{
183	1	$this->messageProcessor->close($this->stream);
184	1	}
185
186		/**
187		* @return MessageHandlerInterface
188		* @throws NoHandlerException
189		*/
190	5	private function getHandler() : MessageHandlerInterface
191		{
192	5	if ($this->handler instanceof \Closure) {
193	5	$handler = $this->handler;
194	5	$handler = $handler($this->uri, $this);
195
196	5	if (null === $handler) {
197	1	throw new NoHandlerException(sprintf('No handler for request URI %s.', $this->uri));
198		}
199
200	4	$this->handler = $handler;
201		}
202
203	4	return $this->handler;
204		}
205		}
206

Nekland / Woketo

Pull Request — master (#100)

Connection::processHandcheck() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like