Messenger::crashed() - Code Metrics - Inspection of "Handle child process unexpectedly dying, emit exce..." - WyriHaximus/reactphp-child-process-messenger - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( a10806...afde92 )

by Cees-Jan

created 2018-03-23 21:26 UTC

Messenger::crashed() 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\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();
        });
        $this->connection->on('close', function () {
            $calls = $this->outstandingRpcCalls->getCalls();
            if (count($calls) === 0) {
                return;
            }
            $error = new CommunicationWithProcessUnexpectedEndException();
            $this->emit('error', [$error, $this]);
            /** @var OutstandingCall $call */
            foreach ($calls as $call) {
                $call->reject($error);
            }
        });
    }

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

    /**
     * @param int|null $exitCode
     * @internal
     */
    public function crashed($exitCode)
    {
        $this->emit('error', [new ProcessUnexpectedEndException($exitCode), $this]);
    }

    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	7	public function __construct(
58		ConnectionInterface $connection,
59		array $options = []
60		) {
61	7	$this->connection = $connection;
62
63	7	$this->options = $this->defaultOptions + $options;
64
65	7	$this->outstandingRpcCalls = new OutstandingCalls();
66
67		$this->connection->on('data', function ($data) {
68	2	$this->buffer .= $data;
69	2	$this->emit('data', [$data]);
70	2	$this->handleData();
71	7	});
72		$this->connection->on('close', function () {
73	1	$calls = $this->outstandingRpcCalls->getCalls();
74	1	if (count($calls) === 0) {
75	1	return;
76		}
77		$error = new CommunicationWithProcessUnexpectedEndException();
78		$this->emit('error', [$error, $this]);
79		/** @var OutstandingCall $call */
80		foreach ($calls as $call) {
81		$call->reject($error);
82		}
83	7	});
84	7	}
85
86		/**
87		* @param string $target
88		* @param callable $listener
89		*/
90	1	public function registerRpc($target, callable $listener)
91		{
92	1	$this->rpcs[$target] = $listener;
93	1	}
94
95		/**
96		* @param string $target
97		*/
98	1	public function deregisterRpc($target)
99		{
100	1	unset($this->rpcs[$target]);
101	1	}
102
103		/**
104		* @param string $target
105		* @return bool
106		*/
107	1	public function hasRpc($target)
108		{
109	1	return isset($this->rpcs[$target]);
110		}
111
112		/**
113		* @param $target
114		* @param $payload
115		* @return React\Promise\PromiseInterface
116		*/
117	1	public function callRpc($target, $payload)
118		{
119		try {
120	1	$promise = $this->rpcs[$target]($payload, $this);
121	1	if ($promise instanceof PromiseInterface) {
122		return $promise;
123		}
124
125	1	throw new \Exception('RPC must return promise');
126	1	} catch (\Exception $exception) {
127	1	return new RejectedPromise($exception);
128		} 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...
129		return new RejectedPromise($exception);
130		}
131		}
132
133		/**
134		* @param Message $message
135		*/
136	1	public function message(Message $message)
137		{
138	1	$this->write($this->createLine($message));
139	1	}
140
141		/**
142		* @param Error $error
143		*/
144	1	public function error(Error $error)
145		{
146	1	$this->write($this->createLine($error));
147	1	}
148
149		/**
150		* @param string $uniqid
151		* @return OutstandingCall
152		*/
153	1	public function getOutstandingCall($uniqid)
154		{
155	1	return $this->outstandingRpcCalls->getCall($uniqid);
156		}
157
158		/**
159		* @param Rpc $rpc
160		* @return \React\Promise\Promise
161		*/
162		public function rpc(Rpc $rpc)
163		{
164	2	$callReference = $this->outstandingRpcCalls->newCall(function () {
165	2	});
166
167	2	$this->write($this->createLine($rpc->setUniqid($callReference->getUniqid())));
168
169	2	return $callReference->getDeferred()->promise();
170		}
171
172		/**
173		* @param ActionableMessageInterface $line
174		* @return LineInterface
175		*/
176	4	public function createLine(ActionableMessageInterface $line)
177		{
178	4	$lineCLass = $this->options['lineClass'];
179
180	4	return (string) new $lineCLass($line, $this->options['lineOptions']);
181		}
182
183		/**
184		* @return \React\Promise\Promise
185		*/
186	1	public function softTerminate()
187		{
188	1	return $this->rpc(MessagesFactory::rpc(static::TERMINATE_RPC));
189		}
190
191		/**
192		* @param string $line
193		*/
194	4	public function write($line)
195		{
196	4	$this->connection->write($line);
197	4	}
198
199		/**
200		* @param int\|null $exitCode
201		* @internal
202		*/
203	1	public function crashed($exitCode)
204		{
205	1	$this->emit('error', [new ProcessUnexpectedEndException($exitCode), $this]);
206	1	}
207
208	2	private function handleData()
209		{
210	2	if (strpos($this->buffer, LineInterface::EOL) === false) {
211	1	return;
212		}
213
214	1	$messages = explode(LineInterface::EOL, $this->buffer);
215	1	$this->buffer = array_pop($messages);
216	1	$this->iterateMessages($messages);
217	1	}
218
219	1	private function iterateMessages(array $messages)
220		{
221	1	foreach ($messages as $message) {
222		try {
223	1	MessagesFactory::fromLine($message, [])->handle($this, 'source');
224		} catch (\Exception $exception) {
225		$this->emit('error', [$exception, $this]);
226		} 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...
227	1	$this->emit('error', [$exception, $this]);
228		}
229		}
230	1	}
231		}
232

WyriHaximus / reactphp-child-process-messenger

GitHub Access Token became invalid

Push — master ( a10806...afde92 )

Messenger::crashed() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like