Messenger::message() - Code Metrics - Inspection of "Code style" - WyriHaximus/reactphp-child-process-messenger - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 1c2b2d...9313fa )

by Cees-Jan

created 2017-08-10 20:25 UTC

Messenger::message() A

↳ Parent: Messenger

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
dl	0
loc	4
ccs	3
cts	3
cp	1
rs	10
c	0
b	0
f	0
cc	1
eloc	2
nc	1
nop	1
crap	1

<?php

namespace WyriHaximus\React\ChildProcess\Messenger;

use Evenement\EventEmitter;
use React\Promise\PromiseInterface;
use React\Promise\RejectedPromise;
use React\Stream\Stream;
use WyriHaximus\React\ChildProcess\Messenger\Messages\ActionableMessageInterface;
use WyriHaximus\React\ChildProcess\Messenger\Messages\Error;
use WyriHaximus\React\ChildProcess\Messenger\Messages\Factory as MessageFactory;
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 Stream
     */
    protected $stdin;

    /**
     * @var Stream
     */
    protected $stdout;

    /**
     * @var Stream
     */
    protected $stderr;

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

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

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

    /**
     * @var string[]
     */
    protected $buffers = [
        'stdin' => '',
        'stdout' => '',
        'stderr' => '',
    ];

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

    /**
     * @param Stream $stdin
     * @param Stream $stdout
     * @param Stream $stderr
     * @param array  $options
     */
    public function __construct(Stream $stdin, Stream $stdout, Stream $stderr, array $options)
    {
        $this->stdin  = $stdin;
        $this->stdout = $stdout;
        $this->stderr = $stderr;
        $this->options = $this->defaultOptions + $options;

        $this->outstandingRpcCalls = new OutstandingCalls();

        $this->attachMessenger();
    }

    /**
     * Forward any unknown calls when there is a call forward possible.
     *
     * @param string $name
     * @param array  $arguments
     *
     * @return mixed
     */
    public function __call($name, array $arguments)
    {
        if (isset($this->options['callForward'])) {
            $call = $this->options['callForward'];

            return $call($name, $arguments);
        }
    }

    /**
     * @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->writeErr($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(MessageFactory::rpc(static::TERMINATE_RPC));
    }

    /**
     * @return Stream
     */
    public function getStdin()
    {
        return $this->stdin;
    }

    /**
     * @return Stream
     */
    public function getStdout()
    {
        return $this->stdout;
    }

    /**
     * @return Stream
     */
    public function getStderr()
    {
        return $this->stderr;
    }

    protected function attachMessenger()
    {
        /**
         * @todo duplicated code much?
         */
        if (isset($this->options['read_err'])) {

            $streamName = $this->options['read_err'];
            $this->$streamName->on('data', function ($data) use ($streamName) {
                $this->onData($data, $streamName);
            });
            unset($streamName);
        }

        if (isset($this->options['read'])) {

            $streamName = $this->options['read'];
            $this->$streamName->on('data', function ($data) use ($streamName) {
                $this->onData($data, $streamName);
            });
            unset($streamName);
        }
    }

    /**
     * @param string $line
     */
    protected function write($line)

    {
        if (isset($this->options['write'])) {
            $streamName = $this->options['write'];
            $this->$streamName->write($line);
            unset($streamName);
        }
    }

    /**
     * @param string $line
     */
    protected function writeErr($line)

    {
        if (isset($this->options['write_err'])) {
            $streamName = $this->options['write_err'];
            $this->$streamName->write($line);
            unset($streamName);
        }
    }

    /**
     * @param string $data
     * @param string $source
     */
    protected function onData($data, $source)
    {
        $this->emit('data', [$source, $data]);

        $this->buffers[$source] .= $data;

        if (strpos($this->buffers[$source], LineInterface::EOL) !== false) {
            $messages = explode(LineInterface::EOL, $this->buffers[$source]);
            $this->buffers[$source] = array_pop($messages);
            $this->iterateMessages($messages, $source);
        }
    }

    /**
     * @param array  $messages
     * @param string $source
     */
    protected function iterateMessages(array $messages, $source)
    {
        $messageFactory = $this->options['messageFactoryClass'];
        foreach ($messages as $message) {
            try {
                $messageFactory::fromLine($message, $this->options['lineOptions'])->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\Stream\Stream;
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 MessageFactory;
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 Stream
23			*/
24			protected $stdin;
25
26			/**
27			* @var Stream
28			*/
29			protected $stdout;
30
31			/**
32			* @var Stream
33			*/
34			protected $stderr;
35
36			/**
37			* @var OutstandingCalls
38			*/
39			protected $outstandingRpcCalls;
40
41			/**
42			* @var array
43			*/
44			protected $rpcs = [];
45
46			/**
47			* @var array
48			*/
49			protected $options = [];
50
51			/**
52			* @var string[]
53			*/
54			protected $buffers = [
55			'stdin' => '',
56			'stdout' => '',
57			'stderr' => '',
58			];
59
60			protected $defaultOptions = [
61			'lineClass' => 'WyriHaximus\React\ChildProcess\Messenger\Messages\Line',
62			'messageFactoryClass' => 'WyriHaximus\React\ChildProcess\Messenger\Messages\Factory',
63			'lineOptions' => [],
64			];
65
66			/**
67			* @param Stream $stdin
68			* @param Stream $stdout
69			* @param Stream $stderr
70			* @param array $options
71			*/
72	10		public function __construct(Stream $stdin, Stream $stdout, Stream $stderr, array $options)
73			{
74	10		$this->stdin = $stdin;
75	10		$this->stdout = $stdout;
76	10		$this->stderr = $stderr;
77	10		$this->options = $this->defaultOptions + $options;
78
79	10		$this->outstandingRpcCalls = new OutstandingCalls();
80
81	10		$this->attachMessenger();
82	10		}
83
84			/**
85			* Forward any unknown calls when there is a call forward possible.
86			*
87			* @param string $name
88			* @param array $arguments
89			*
90			* @return mixed
91			*/
92	1		public function __call($name, array $arguments)
93			{
94	1		if (isset($this->options['callForward'])) {
95	1		$call = $this->options['callForward'];
96
97	1		return $call($name, $arguments);
98			}
99			}
100
101			/**
102			* @param string $target
103			* @param callable $listener
104			*/
105	2		public function registerRpc($target, callable $listener)
106			{
107	2		$this->rpcs[$target] = $listener;
108	2		}
109
110			/**
111			* @param string $target
112			*/
113	1		public function deregisterRpc($target)
114			{
115	1		unset($this->rpcs[$target]);
116	1		}
117
118			/**
119			* @param string $target
120			* @return bool
121			*/
122	1		public function hasRpc($target)
123			{
124	1		return isset($this->rpcs[$target]);
125			}
126
127			/**
128			* @param $target
129			* @param $payload
130			* @return React\Promise\PromiseInterface
131			*/
132	2		public function callRpc($target, $payload)
133			{
134			try {
135	2		$promise = $this->rpcs[$target]($payload, $this);
136	2		if ($promise instanceof PromiseInterface) {
137	1		return $promise;
138			}
139
140	1		throw new \Exception('RPC must return promise');
141	1		} catch (\Exception $exception) {
142	1		return new RejectedPromise($exception);
143			} 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...
144			return new RejectedPromise($exception);
145			}
146			}
147
148			/**
149			* @param Message $message
150			*/
151	1		public function message(Message $message)
152			{
153	1		$this->write($this->createLine($message));
154	1		}
155
156			/**
157			* @param Error $error
158			*/
159	1		public function error(Error $error)
160			{
161	1		$this->writeErr($this->createLine($error));
162	1		}
163
164			/**
165			* @param string $uniqid
166			* @return OutstandingCall
167			*/
168	1		public function getOutstandingCall($uniqid)
169			{
170	1		return $this->outstandingRpcCalls->getCall($uniqid);
171			}
172
173			/**
174			* @param Rpc $rpc
175			* @return \React\Promise\Promise
176			*/
177	2		public function rpc(Rpc $rpc)
178			{
179			$callReference = $this->outstandingRpcCalls->newCall(function () {
180	2		});
181
182	2		$this->write($this->createLine($rpc->setUniqid($callReference->getUniqid())));
183
184	2		return $callReference->getDeferred()->promise();
185			}
186
187			/**
188			* @param ActionableMessageInterface $line
189			* @return LineInterface
190			*/
191	4		public function createLine(ActionableMessageInterface $line)
192			{
193	4		$lineCLass = $this->options['lineClass'];
194
195	4		return (string) new $lineCLass($line, $this->options['lineOptions']);
196			}
197
198			/**
199			* @return \React\Promise\Promise
200			*/
201	1		public function softTerminate()
202			{
203	1		return $this->rpc(MessageFactory::rpc(static::TERMINATE_RPC));
204			}
205
206			/**
207			* @return Stream
208			*/
209	3		public function getStdin()
210			{
211	3		return $this->stdin;
212			}
213
214			/**
215			* @return Stream
216			*/
217	3		public function getStdout()
218			{
219	3		return $this->stdout;
220			}
221
222			/**
223			* @return Stream
224			*/
225	3		public function getStderr()
226			{
227	3		return $this->stderr;
228			}
229
230	10		protected function attachMessenger()
231			{
232			/**
233			* @todo duplicated code much?
234			*/
235	10	View Code Duplication	if (isset($this->options['read_err'])) {
			0 ignored issues – show Duplication introduced 2015-11-21 19:02 UTC by Report Bug Copy Issue Report This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
236	2		$streamName = $this->options['read_err'];
237			$this->$streamName->on('data', function ($data) use ($streamName) {
238			$this->onData($data, $streamName);
239	2		});
240	2		unset($streamName);
241			}
242
243	10	View Code Duplication	if (isset($this->options['read'])) {
			0 ignored issues – show Duplication introduced 2015-11-21 19:02 UTC by Report Bug Copy Issue Report This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
244	6		$streamName = $this->options['read'];
245	6		$this->$streamName->on('data', function ($data) use ($streamName) {
246	3		$this->onData($data, $streamName);
247	6		});
248	6		unset($streamName);
249			}
250	10		}
251
252			/**
253			* @param string $line
254			*/
255	3	View Code Duplication	protected function write($line)
			0 ignored issues – show Duplication introduced 2017-08-02 21:05 UTC by Report Bug Copy Issue Report This method seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
256			{
257	3		if (isset($this->options['write'])) {
258	3		$streamName = $this->options['write'];
259	3		$this->$streamName->write($line);
260	3		unset($streamName);
261			}
262	3		}
263
264			/**
265			* @param string $line
266			*/
267	1	View Code Duplication	protected function writeErr($line)
			0 ignored issues – show Duplication introduced 2017-08-02 21:05 UTC by Report Bug Copy Issue Report This method seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
268			{
269	1		if (isset($this->options['write_err'])) {
270	1		$streamName = $this->options['write_err'];
271	1		$this->$streamName->write($line);
272	1		unset($streamName);
273			}
274	1		}
275
276			/**
277			* @param string $data
278			* @param string $source
279			*/
280	3		protected function onData($data, $source)
281			{
282	3		$this->emit('data', [$source, $data]);
283
284	3		$this->buffers[$source] .= $data;
285
286	3		if (strpos($this->buffers[$source], LineInterface::EOL) !== false) {
287	2		$messages = explode(LineInterface::EOL, $this->buffers[$source]);
288	2		$this->buffers[$source] = array_pop($messages);
289	2		$this->iterateMessages($messages, $source);
290			}
291	3		}
292
293			/**
294			* @param array $messages
295			* @param string $source
296			*/
297	2		protected function iterateMessages(array $messages, $source)
298			{
299	2		$messageFactory = $this->options['messageFactoryClass'];
300	2		foreach ($messages as $message) {
301			try {
302	2		$messageFactory::fromLine($message, $this->options['lineOptions'])->handle($this, $source);
303			} catch (\Exception $exception) {
304			$this->emit('error', [$exception, $this]);
305			} 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...
306	2		$this->emit('error', [$exception, $this]);
307			}
308			}
309	2		}
310			}
311

WyriHaximus / reactphp-child-process-messenger

GitHub Access Token became invalid

Push — master ( 1c2b2d...9313fa )

Messenger::message() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like