Handle child process unexpectedly dying, emit exceptions when that ha…

…ppens, and also when the connection with the child process gets cut off. Run overflow example on CI's to ensure that behavior

Closes #14

.travis.yml

@@ -80,6 +80,7 @@ script:
   - if [ "$qaExtended" = "true" ]; then make ci-extended; fi;
   - php benchmark/memory.php
   - php examples/error/error.php
+  - php examples/overflow/overflow.php
 
 ## Gather coverage and set it to coverage servers
 after_script: if [ "$qaExtended" = "true" ]; then make ci-coverage; fi;

appveyor.yml

@@ -74,3 +74,4 @@ test_script:
     - vendor/bin/phpunit -c phpunit.xml.dist
     - php benchmark/memory.php
     - php example/error/error.php
+    - php example/overflow/overflow.php

examples/ExamplesChildProcess.php

@@ -29,6 +29,14 @@ public static function create(Messenger $messenger, LoopInterface $loop)
                 'formattedTime' => (new DateTime('@' . $payload['unixTime']))->format('c'),
             ]);
         });
+        $messenger->registerRpc('overflow', function () {
+            ini_set('memory_limit', '20M');
+
+            $string = '';
+            while (true) {
+                $string .= '0123456789';
+            }
+        });
     }
 
     public static function isPrime($n)

examples/overflow/overflow.php

@@ -0,0 +1,25 @@
+<?php
+
+require dirname(dirname(__DIR__)) . '/vendor/autoload.php';
+
+use React\EventLoop\Factory;
+use WyriHaximus\React\ChildProcess\Messenger\Factory as MessengerFactory;
+use WyriHaximus\React\ChildProcess\Messenger\Messages\Factory as MessageFactory;
+use WyriHaximus\React\ChildProcess\Messenger\Messages\Payload;
+use WyriHaximus\React\ChildProcess\Messenger\Messenger;
+
+$loop = Factory::create();
+
+MessengerFactory::parentFromClass(ExamplesChildProcess::class, $loop)->then(function (Messenger $messenger) {
+    return $messenger->rpc(
+        MessageFactory::rpc('overflow')
+    )->always(function () use ($messenger) {
+        $messenger->softTerminate();
+    });
+})->done(function (Payload $result) {
+    throw new Exception('Should never reach this!');
+}, function ($et) {
+    echo (string)$et;
+});
+
+$loop->run();

src/CommunicationWithProcessUnexpectedEndException.php

@@ -0,0 +1,13 @@
+<?php
+
+namespace WyriHaximus\React\ChildProcess\Messenger;
+
+use Exception;
+
+final class CommunicationWithProcessUnexpectedEndException extends Exception
+{
+    public function __construct()
+    {
+        parent::__construct('Communication with process stopped unexpectedly');
+    }
+}

src/Factory.php

@@ -137,7 +137,7 @@ private static function startParent(
         LoopInterface $loop,
         array $options
     ) {
-        return new Promise\Promise(function ($resolve, $reject) use ($process, $server, $loop, $options) {
+        return (new Promise\Promise(function ($resolve, $reject) use ($process, $server, $loop, $options) {
             $server->on(
                 'connection',
                 function (ConnectionInterface $connection) use ($server, $resolve, $reject, $options) {
@@ -162,6 +162,21 @@ function (ConnectionInterface $connection) use ($server, $resolve, $reject, $opt
             });
 
             $process->start($loop);
+        }))->then(function (Messenger $messenger) use ($loop, $process) {
+            $loop->addPeriodicTimer(self::INTERVAL, function ($timer) use ($messenger, $loop, $process) {
+                if (!$process->isRunning()) {
+                    $loop->cancelTimer($timer);
+
+                    $exitCode = $process->getExitCode();
+                    if ($exitCode === 0) {
+                        return;
+                    }
+
+                    $messenger->crashed($exitCode);
+                }
+            });
+
+            return $messenger;
         });
     }
 }

src/Messenger.php

@@ -69,6 +69,18 @@ public function __construct(
             $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);
+            }
+        });
     }
 
     /**
@@ -184,6 +196,15 @@ 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) {

src/OutstandingCalls.php

@@ -34,6 +34,14 @@ public function getCall($uniqid)
         return $this->calls[$uniqid];
     }
 
+    /**
+     * @return array
+     */
+    public function getCalls()
+    {
+        return array_values($this->calls);
+    }
+
     /**
      * @return string
      */

src/ProcessUnexpectedEndException.php

@@ -0,0 +1,16 @@
+<?php
+
+namespace WyriHaximus\React\ChildProcess\Messenger;
+
+use Exception;
+
+final class ProcessUnexpectedEndException extends Exception
+{
+    /**
+     * @param int $exitCode
+     */
+    public function __construct($exitCode)
+    {
+        parent::__construct('Process stopped unexpectedly', $exitCode);
+    }
+}

tests/MessengerTest.php

@@ -4,6 +4,7 @@
 
 use PHPUnit\Framework\TestCase;
 use Prophecy\Argument;
+use React\EventLoop\Factory;
 use WyriHaximus\React\ChildProcess\Messenger\Messenger;
 use WyriHaximus\React\Tests\ChildProcess\Messenger\Stub\ConnectionStub;
 
@@ -13,6 +14,7 @@ public function testSetAndHasRpc()
     {
         $connection = $this->prophesize('React\Socket\ConnectionInterface');
         $connection->on('data', Argument::type('callable'))->shouldBeCalled();
+        $connection->on('close', Argument::type('callable'))->shouldBeCalled();
         $messenger = new Messenger($connection->reveal());
 
         $payload = [
@@ -42,6 +44,7 @@ public function testMessage()
     {
         $connection = $this->prophesize('React\Socket\ConnectionInterface');
         $connection->on('data', Argument::type('callable'))->shouldBeCalled();
+        $connection->on('close', Argument::type('callable'))->shouldBeCalled();
         $connection->write(Argument::type('string'))->shouldBeCalled();
 
         $messenger = new Messenger($connection->reveal());
@@ -55,6 +58,7 @@ public function testError()
     {
         $connection = $this->prophesize('React\Socket\ConnectionInterface');
         $connection->on('data', Argument::type('callable'))->shouldBeCalled();
+        $connection->on('close', Argument::type('callable'))->shouldBeCalled();
         $connection->write(Argument::type('string'))->shouldBeCalled();
 
         $messenger = new Messenger($connection->reveal());
@@ -68,6 +72,7 @@ public function testRpc()
     {
         $connection = $this->prophesize('React\Socket\ConnectionInterface');
         $connection->on('data', Argument::type('callable'))->shouldBeCalled();
+        $connection->on('close', Argument::type('callable'))->shouldBeCalled();
         $connection->write(Argument::type('string'))->shouldBeCalled();
 
         $messenger = new Messenger($connection->reveal());
@@ -91,4 +96,19 @@ public function testEmitOnData()
 
         $this->assertTrue($cbCalled);
     }
+
+    public function testCrashed()
+    {
+        $this->setExpectedException('WyriHaximus\React\ChildProcess\Messenger\ProcessUnexpectedEndException');
+
+        $loop = Factory::create();
+        $connection = new ConnectionStub();
+
+        $messenger = new Messenger($connection);
+        $loop->futureTick(function () use ($messenger) {
+            $messenger->crashed(123);
+        });
+
+        throw \Clue\React\Block\await(\React\Promise\Stream\first($messenger, 'error'), $loop);
+    }
 }

tests/OutstandingCallsTest.php

@@ -12,6 +12,7 @@ public function testBasic()
         $oc = new OutstandingCalls();
         $call = $oc->newCall();
         $this->assertInstanceOf('WyriHaximus\React\ChildProcess\Messenger\OutstandingCall', $call);
+        $this->assertEquals([$call], $oc->getCalls());
         $this->assertEquals($call, $oc->getCall($call->getUniqid()));
     }
 }