Messenger::onData() - Code Metrics - Inspection of "react/child-process 0.4 && 0.5 && 0.6" - WyriHaximus/reactphp-child-process-messenger - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#13)

by Cees-Jan

created 2018-02-20 22:12 UTC

Messenger::onData() A

↳ Parent: Project

Complexity

Conditions	2
Paths	2

Size

Total Lines	12
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	2

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	12
ccs	8
cts	8
cp	1
rs	9.4285
cc	2
eloc	7
nc	2
nop	2
crap	2

<?php

namespace WyriHaximus\React\ChildProcess\Messenger;

use Evenement\EventEmitter;
use React\Promise\PromiseInterface;
use React\Promise\RejectedPromise;
use React\Socket\ConnectionInterface;
use WyriHaximus\React\ChildProcess\Messenger\Messages\ActionableMessageInterface;
use WyriHaximus\React\ChildProcess\Messenger\Messages\Error;
use WyriHaximus\React\ChildProcess\Messenger\Messages\Factory as MessagesFactory;
use WyriHaximus\React\ChildProcess\Messenger\Messages\LineInterface;
use WyriHaximus\React\ChildProcess\Messenger\Messages\Message;
use WyriHaximus\React\ChildProcess\Messenger\Messages\Rpc;

class Messenger extends EventEmitter
{
    const INTERVAL = 0.1;
    const TERMINATE_RPC = 'wyrihaximus.react.child-process.messenger.terminate';

    /**
     * @var ConnectionInterface
     */
    protected $connection;

    /**
     * @var OutstandingCalls
     */
    protected $outstandingRpcCalls;

    /**
     * @var array
     */
    protected $rpcs = [];

    /**
     * @var array
     */
    protected $options = [];

    /**
     * @var string[]
     */
    protected $buffer = '';

    protected $defaultOptions = [
        'lineClass' => 'WyriHaximus\React\ChildProcess\Messenger\Messages\Line',
        'messageFactoryClass' => 'WyriHaximus\React\ChildProcess\Messenger\Messages\Factory',
        'lineOptions' => [],
    ];

    /**
     * Messenger constructor.
     * @param ConnectionInterface $connection
     * @param array               $options
     */
    public function __construct(
        ConnectionInterface $connection,
        array $options = []
    ) {
        $this->connection = $connection;

        $this->options = $this->defaultOptions + $options;

        $this->outstandingRpcCalls = new OutstandingCalls();

        $this->connection->on('data', function ($data) {
            $this->buffer .= $data;
            $this->emit('data', [$data]);
            $this->handleData();
        });
    }

    /**
     * @param string   $target
     * @param callable $listener
     */
    public function registerRpc($target, callable $listener)
    {
        $this->rpcs[$target] = $listener;
    }

    /**
     * @param string $target
     */
    public function deregisterRpc($target)
    {
        unset($this->rpcs[$target]);
    }

    /**
     * @param  string $target
     * @return bool
     */
    public function hasRpc($target)
    {
        return isset($this->rpcs[$target]);
    }

    /**
     * @param $target
     * @param $payload
     * @return React\Promise\PromiseInterface
     */
    public function callRpc($target, $payload)
    {
        try {
            $promise = $this->rpcs[$target]($payload, $this);
            if ($promise instanceof PromiseInterface) {
                return $promise;
            }

            throw new \Exception('RPC must return promise');
        } catch (\Exception $exception) {
            return new RejectedPromise($exception);
        } catch (\Throwable $exception) {

            return new RejectedPromise($exception);
        }
    }

    /**
     * @param Message $message
     */
    public function message(Message $message)
    {
        $this->write($this->createLine($message));
    }

    /**
     * @param Error $error
     */
    public function error(Error $error)
    {
        $this->write($this->createLine($error));
    }

    /**
     * @param  string          $uniqid
     * @return OutstandingCall
     */
    public function getOutstandingCall($uniqid)
    {
        return $this->outstandingRpcCalls->getCall($uniqid);
    }

    /**
     * @param  Rpc                    $rpc
     * @return \React\Promise\Promise
     */
    public function rpc(Rpc $rpc)
    {
        $callReference = $this->outstandingRpcCalls->newCall(function () {
        });

        $this->write($this->createLine($rpc->setUniqid($callReference->getUniqid())));

        return $callReference->getDeferred()->promise();
    }

    /**
     * @param  ActionableMessageInterface $line
     * @return LineInterface
     */
    public function createLine(ActionableMessageInterface $line)
    {
        $lineCLass = $this->options['lineClass'];

        return (string) new $lineCLass($line, $this->options['lineOptions']);
    }

    /**
     * @return \React\Promise\Promise
     */
    public function softTerminate()
    {
        return $this->rpc(MessagesFactory::rpc(static::TERMINATE_RPC));
    }

    /**
     * @param string $line
     */
    public function write($line)
    {
        $this->connection->write($line);
    }

    private function handleData()
    {
        if (strpos($this->buffer, LineInterface::EOL) === false) {
            return;
        }

        $messages = explode(LineInterface::EOL, $this->buffer);
        $this->buffer = array_pop($messages);
        $this->iterateMessages($messages);
    }

    private function iterateMessages(array $messages)
    {
        foreach ($messages as $message) {
            try {
                MessagesFactory::fromLine($message, [])->handle($this, 'source');
            } catch (\Exception $exception) {
                $this->emit('error', [$exception, $this]);
            } catch (\Throwable $exception) {

                $this->emit('error', [$exception, $this]);
            }
        }
    }
}


1		<?php
2
3		namespace WyriHaximus\React\ChildProcess\Messenger;
4
5		use Evenement\EventEmitter;
6		use React\Promise\PromiseInterface;
7		use React\Promise\RejectedPromise;
8		use React\Socket\ConnectionInterface;
9		use WyriHaximus\React\ChildProcess\Messenger\Messages\ActionableMessageInterface;
10		use WyriHaximus\React\ChildProcess\Messenger\Messages\Error;
11		use WyriHaximus\React\ChildProcess\Messenger\Messages\Factory as MessagesFactory;
12		use WyriHaximus\React\ChildProcess\Messenger\Messages\LineInterface;
13		use WyriHaximus\React\ChildProcess\Messenger\Messages\Message;
14		use WyriHaximus\React\ChildProcess\Messenger\Messages\Rpc;
15
16		class Messenger extends EventEmitter
17		{
18		const INTERVAL = 0.1;
19		const TERMINATE_RPC = 'wyrihaximus.react.child-process.messenger.terminate';
20
21		/**
22		* @var ConnectionInterface
23		*/
24		protected $connection;
25
26		/**
27		* @var OutstandingCalls
28		*/
29		protected $outstandingRpcCalls;
30
31		/**
32		* @var array
33		*/
34		protected $rpcs = [];
35
36		/**
37		* @var array
38		*/
39		protected $options = [];
40
41		/**
42		* @var string[]
43		*/
44		protected $buffer = '';
45
46		protected $defaultOptions = [
47		'lineClass' => 'WyriHaximus\React\ChildProcess\Messenger\Messages\Line',
48		'messageFactoryClass' => 'WyriHaximus\React\ChildProcess\Messenger\Messages\Factory',
49		'lineOptions' => [],
50		];
51
52		/**
53		* Messenger constructor.
54		* @param ConnectionInterface $connection
55		* @param array $options
56		*/
57	8	public function __construct(
58		ConnectionInterface $connection,
59		array $options = []
60		) {
61	8	$this->connection = $connection;
62
63	8	$this->options = $this->defaultOptions + $options;
64
65	8	$this->outstandingRpcCalls = new OutstandingCalls();
66
67		$this->connection->on('data', function ($data) {
68	3	$this->buffer .= $data;
69	3	$this->emit('data', [$data]);
70	3	$this->handleData();
71	8	});
72	8	}
73
74		/**
75		* @param string $target
76		* @param callable $listener
77		*/
78	2	public function registerRpc($target, callable $listener)
79		{
80	2	$this->rpcs[$target] = $listener;
81	2	}
82
83		/**
84		* @param string $target
85		*/
86	1	public function deregisterRpc($target)
87		{
88	1	unset($this->rpcs[$target]);
89	1	}
90
91		/**
92		* @param string $target
93		* @return bool
94		*/
95	1	public function hasRpc($target)
96		{
97	1	return isset($this->rpcs[$target]);
98		}
99
100		/**
101		* @param $target
102		* @param $payload
103		* @return React\Promise\PromiseInterface
104		*/
105	2	public function callRpc($target, $payload)
106		{
107		try {
108	2	$promise = $this->rpcs[$target]($payload, $this);
109	2	if ($promise instanceof PromiseInterface) {
110	1	return $promise;
111		}
112
113	1	throw new \Exception('RPC must return promise');
114	1	} catch (\Exception $exception) {
115	1	return new RejectedPromise($exception);
116		} catch (\Throwable $exception) {
		0 ignored issues – show Bug introduced 2017-07-13 17:53 UTC by Report Bug Copy Issue Report The class `Throwable` does not exist. Did you forget a USE statement, or did you not list all dependencies? Scrutinizer analyzes your `composer.json`/`composer.lock` file if available to determine the classes, and functions that are defined by your dependencies. It seems like the listed class was neither found in your dependencies, nor was it found in the analyzed files in your repository. If you are using some other form of dependency management, you might want to disable this analysis. Loading history...
117		return new RejectedPromise($exception);
118		}
119		}
120
121		/**
122		* @param Message $message
123		*/
124	1	public function message(Message $message)
125		{
126	1	$this->write($this->createLine($message));
127	1	}
128
129		/**
130		* @param Error $error
131		*/
132	1	public function error(Error $error)
133		{
134	1	$this->write($this->createLine($error));
135	1	}
136
137		/**
138		* @param string $uniqid
139		* @return OutstandingCall
140		*/
141	1	public function getOutstandingCall($uniqid)
142		{
143	1	return $this->outstandingRpcCalls->getCall($uniqid);
144		}
145
146		/**
147		* @param Rpc $rpc
148		* @return \React\Promise\Promise
149		*/
150		public function rpc(Rpc $rpc)
151		{
152	2	$callReference = $this->outstandingRpcCalls->newCall(function () {
153	2	});
154
155	2	$this->write($this->createLine($rpc->setUniqid($callReference->getUniqid())));
156
157	2	return $callReference->getDeferred()->promise();
158		}
159
160		/**
161		* @param ActionableMessageInterface $line
162		* @return LineInterface
163		*/
164	4	public function createLine(ActionableMessageInterface $line)
165		{
166	4	$lineCLass = $this->options['lineClass'];
167
168	4	return (string) new $lineCLass($line, $this->options['lineOptions']);
169		}
170
171		/**
172		* @return \React\Promise\Promise
173		*/
174	1	public function softTerminate()
175		{
176	1	return $this->rpc(MessagesFactory::rpc(static::TERMINATE_RPC));
177		}
178
179		/**
180		* @param string $line
181		*/
182	4	public function write($line)
183		{
184	4	$this->connection->write($line);
185	4	}
186
187	3	private function handleData()
188		{
189	3	if (strpos($this->buffer, LineInterface::EOL) === false) {
190	1	return;
191		}
192
193	2	$messages = explode(LineInterface::EOL, $this->buffer);
194	2	$this->buffer = array_pop($messages);
195	2	$this->iterateMessages($messages);
196	2	}
197
198	2	private function iterateMessages(array $messages)
199		{
200	2	foreach ($messages as $message) {
201		try {
202	2	MessagesFactory::fromLine($message, [])->handle($this, 'source');
203		} catch (\Exception $exception) {
204		$this->emit('error', [$exception, $this]);
205		} catch (\Throwable $exception) {
		0 ignored issues – show Bug introduced 2017-07-13 17:53 UTC by Report Bug Copy Issue Report The class `Throwable` does not exist. Did you forget a USE statement, or did you not list all dependencies? Scrutinizer analyzes your `composer.json`/`composer.lock` file if available to determine the classes, and functions that are defined by your dependencies. It seems like the listed class was neither found in your dependencies, nor was it found in the analyzed files in your repository. If you are using some other form of dependency management, you might want to disable this analysis. Loading history...
206	2	$this->emit('error', [$exception, $this]);
207		}
208		}
209	2	}
210		}
211

WyriHaximus / reactphp-child-process-messenger

GitHub Access Token became invalid

Pull Request — master (#13)

Messenger::onData() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like