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

Completed

Push — master ( 05ba71...78c5cf )

by Cees-Jan

created 2015-12-26 20:17 UTC

Messenger::error() A

↳ Parent: Messenger

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes	1
Bugs	0	Features	1

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

<?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();
    }

    /**
     * @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);
        }
    }

    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 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 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]);
            }
        }
    }

    /**
     * @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;
    }

    /**
     * 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);
        }
    }
}


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	8		* @param array $options
71			*/
72	8		public function __construct(Stream $stdin, Stream $stdout, Stream $stderr, array $options)
73	8		{
74	8		$this->stdin = $stdin;
75	8		$this->stdout = $stdout;
76			$this->stderr = $stderr;
77	8		$this->options = $this->defaultOptions + $options;
78
79	8		$this->outstandingRpcCalls = new OutstandingCalls();
80	8
81			$this->attachMessenger();
82			}
83
84			/**
85			* @param string $target
86	2		* @param callable $listener
87			*/
88	2		public function registerRpc($target, callable $listener)
89	2		{
90			$this->rpcs[$target] = $listener;
91			}
92
93			/**
94			* @param string $target
95			*/
96			public function deregisterRpc($target)
97			{
98			unset($this->rpcs[$target]);
99			}
100
101			/**
102			* @param string $target
103	1		* @return bool
104			*/
105	1		public function hasRpc($target)
106			{
107			return isset($this->rpcs[$target]);
108			}
109
110			/**
111			* @param $target
112			* @param $payload
113	2		* @return React\Promise\PromiseInterface
114			*/
115			public function callRpc($target, $payload)
116	2		{
117	2		try {
118	1		$promise = $this->rpcs[$target]($payload, $this);
119			if ($promise instanceof PromiseInterface) {
120			return $promise;
121	1		}
122	1
123	1		throw new \Exception('RPC must return promise');
124			} catch (\Exception $exception) {
125			return new RejectedPromise($exception);
126			}
127	8		}
128
129			protected function attachMessenger()
130			{
131			/**
132	8		* @todo duplicated code much?
133	1		*/
134		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...
135			$streamName = $this->options['read_err'];
136	1		$this->$streamName->on('data', function ($data) use ($streamName) {
137	1		$this->onData($data, $streamName);
138	1		});
139			unset($streamName);
140	8		}
141	5
142		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...
143	2		$streamName = $this->options['read'];
144	5		$this->$streamName->on('data', function ($data) use ($streamName) {
145	5		$this->onData($data, $streamName);
146	5		});
147	8		unset($streamName);
148			}
149			}
150
151			/**
152	2		* @param string $line
153			*/
154	2	View Code Duplication	protected function write($line)
			0 ignored issues – show Duplication introduced 2015-11-21 19:02 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...
155	2		{
156	2		if (isset($this->options['write'])) {
157	2		$streamName = $this->options['write'];
158	2		$this->$streamName->write($line);
159	2		unset($streamName);
160			}
161			}
162
163			/**
164			* @param string $line
165			*/
166		View Code Duplication	protected function writeErr($line)
			0 ignored issues – show Duplication introduced 2015-11-21 19:02 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...
167			{
168			if (isset($this->options['write_err'])) {
169			$streamName = $this->options['write_err'];
170			$this->$streamName->write($line);
171			unset($streamName);
172			}
173			}
174
175			/**
176	1		* @param Message $message
177			*/
178	1		public function message(Message $message)
179	1		{
180			$this->write($this->createLine($message));
181			}
182
183			/**
184			* @param Error $error
185			*/
186			public function error(Error $error)
187			{
188			$this->writeErr($this->createLine($error));
189			}
190
191			/**
192			* @param string $uniqid
193			* @return OutstandingCall
194			*/
195			public function getOutstandingCall($uniqid)
196	1		{
197			return $this->outstandingRpcCalls->getCall($uniqid);
198	1		}
199
200	1		/**
201			* @param Rpc $rpc
202	1		* @return \React\Promise\Promise
203			*/
204			public function rpc(Rpc $rpc)
205			{
206			$callReference = $this->outstandingRpcCalls->newCall(function () {
207
208			});
209	2
210			$this->write($this->createLine($rpc->setUniqid($callReference->getUniqid())));
211	2
212			return $callReference->getDeferred()->promise();
213	2		}
214
215	2		/**
216	1		* @param string $data
217	1		* @param string $source
218	1		*/
219	1		protected function onData($data, $source)
220	2		{
221			$this->emit('data', [$source, $data]);
222
223			$this->buffers[$source] .= $data;
224
225			if (strpos($this->buffers[$source], LineInterface::EOL) !== false) {
226	1		$messages = explode(LineInterface::EOL, $this->buffers[$source]);
227			$this->buffers[$source] = array_pop($messages);
228	1		$this->iterateMessages($messages, $source);
229	1		}
230			}
231	1
232	1		/**
233			* @param array $messages
234			* @param string $source
235	1		*/
236	1		protected function iterateMessages(array $messages, $source)
237			{
238			$messageFactory = $this->options['messageFactoryClass'];
239			foreach ($messages as $message) {
240			try {
241			$messageFactory::fromLine($message, $this->options['lineOptions'])->handle($this, $source);
242	2		} catch (\Exception $exception) {
243			$this->emit('error', [$exception, $this]);
244	2		}
245	2		}
246			}
247
248			/**
249			* @param ActionableMessageInterface $line
250			* @return LineInterface
251			*/
252			public function createLine(ActionableMessageInterface $line)
253			{
254			$lineCLass = $this->options['lineClass'];
255			return (string) new $lineCLass($line, $this->options['lineOptions']);
256			}
257
258			/**
259	3		* @return \React\Promise\Promise
260			*/
261	3		public function softTerminate()
262			{
263			return $this->rpc(MessageFactory::rpc(static::TERMINATE_RPC));
264			}
265
266			/**
267	3		* @return Stream
268			*/
269	3		public function getStdin()
270			{
271			return $this->stdin;
272			}
273
274			/**
275	3		* @return Stream
276			*/
277	3		public function getStdout()
278			{
279			return $this->stdout;
280			}
281
282			/**
283			* @return Stream
284			*/
285			public function getStderr()
286			{
287			return $this->stderr;
288	1		}
289
290	1		/**
291	1		* Forward any unknown calls when there is a call forward possible.
292	1		*
293			* @param string $name
294			* @param array $arguments
295			*
296			* @return mixed
297			*/
298			public function __call($name, array $arguments)
299			{
300			if (isset($this->options['callForward'])) {
301			$call = $this->options['callForward'];
302			return $call($name, $arguments);
303			}
304			}
305			}
306

WyriHaximus / reactphp-child-process-messenger

GitHub Access Token became invalid

Push — master ( 05ba71...78c5cf )

Messenger::error() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like