WIP: ChildProcess

examples/child-process.php

@@ -0,0 +1,37 @@
+<?php
+
+// script to spawn multiple child processes
+
+require __DIR__.'/../vendor/autoload.php';
+
+$loop = new React\EventLoop\StreamSelectLoop();
+$factory = new React\ChildProcess\Factory($loop);
+
+$commands = array(
+    'A' => array('cmd' => 'php', 'args' => array('-r', 'foreach (range(1, 3) as $i) { echo $i, PHP_EOL; sleep(1); } fputs(STDERR, "Bye.");')),
+    'B' => array('cmd' => 'php', 'args' => array('-r', 'foreach (range(1, 6) as $i) { echo $i, PHP_EOL; sleep(1); } fputs(STDERR, "Bye.");')),
+    'C' => array('cmd' => 'php', 'args' => array('-r', 'foreach (range(1, 9) as $i) { echo $i, PHP_EOL; sleep(1); } fputs(STDERR, "Bye.");')),
+);
+
+foreach ($commands as $id => $command) {
+    $idLabel = "[{$id}] ";
+    $process = $factory->spawn($command['cmd'], $command['args']);
+    echo $idLabel, '[PID]    pid is ', (string) $process->getPid(), PHP_EOL;
+    echo $idLabel, '[CMD]    ', $process->getCommand(), PHP_EOL;
+
+    $process->stdout->on('data', function ($data) use ($idLabel) {
+        echo $idLabel, '[STDOUT] ';
+        var_dump($data);
+    });
+
+    $process->stderr->on('data', function ($data) use ($idLabel) {
+        echo $idLabel, '[STDERR] ';
+        var_dump($data);
+    });
+
+    $process->on('exit', function ($status) use ($idLabel) {
+        echo $idLabel, '[EXIT]   exited with status code ', (string) $status, PHP_EOL;
+    });
+}
+
+$loop->run();

src/React/ChildProcess/Factory.php

@@ -0,0 +1,53 @@
+<?php
+
+namespace React\ChildProcess;
+
+use React\EventLoop\LoopInterface;
+use React\ChildProcess\Process;
+use React\ChildProcess\Stream;
+
+class Factory
+{
+    private $loop;
+
+    public function __construct(LoopInterface $loop)
+    {
+        $this->loop = $loop;
+    }
+
+    public function spawn($file, array $args = array(), $cwd = null, $env = null)
+    {
+        $cmd = $this->formatCommandWithArguments($file, $args);
+
+        $fdSpec = array(
+            array('pipe', 'r'),
+            array('pipe', 'w'),
+            array('pipe', 'w'),
+        );
+
+        $process = proc_open($cmd, $fdSpec, $pipes, $cwd, $env);
+
+        $stdin  = new Stream($pipes[0], $this->loop);
+        $stdout = new Stream($pipes[1], $this->loop);
+        $stderr = new Stream($pipes[2], $this->loop);
+
+        $stdin->pause();
+
+        stream_set_blocking($pipes[0], 0);
+        stream_set_blocking($pipes[1], 0);
+        stream_set_blocking($pipes[2], 0);
+
+        return new Process($process, $stdin, $stdout, $stderr);
+    }
+
+    private function formatCommandWithArguments($file, $args)
+    {
+        $command = $file;
+
+        if (count($args) > 0) {
+            $command .= ' ' . join(' ', array_map('escapeshellarg', $args));
+        }
+
+        return $command;
+    }
+}

src/React/ChildProcess/Process.php

@@ -0,0 +1,167 @@
+<?php
+
+namespace React\ChildProcess;
+
+use React\Stream\WritableStreamInterface;
+use React\Stream\ReadableStreamInterface;
+use Evenement\EventEmitter;
+
+class Process extends EventEmitter
+{
+    const SIGNAL_CODE_SIGKILL = 9;
+    const SIGNAL_CODE_SIGTERM = 15;
+
+    public $stdin;
+
+    public $stdout;
+
+    public $stderr;
+
+    private $process;
+
+    private $status = null;
+
+    private $exitCode = null;
+
+    private $signalCode = null;
+
+    private $exited = false;
+
+    public function __construct($process, WritableStreamInterface $stdin, ReadableStreamInterface $stdout, ReadableStreamInterface $stderr)
+    {
+        $this->process = $process;
+        $this->stdin   = $stdin;
+        $this->stdout  = $stdout;
+        $this->stderr  = $stderr;
+
+        $self = $this;
+
+        $this->stdout->on('end', function () use ($self) {
+            $self->updateStatus();
+            $self->observeStatus();
+        });
+
+        $this->stderr->on('end', function () use ($self) {
+            $self->updateStatus();
+            $self->observeStatus();
+        });
+    }
+
+    public function updateStatus()
+    {
+        if ($this->process && is_null($this->signalCode)) {
+            $this->status = proc_get_status($this->process);
+
+            if ($this->status['signaled']) {
+                $this->signalCode = $this->status['termsig'];
+            }
+        }
+    }
+
+    public function observeStatus()
+    {
+        if (!$this->stdout->isReadable() && !$this->stderr->isReadable()) {
+            $this->stdin->close();
+            $this->stdout->close();
+            $this->stderr->close();
+            $this->exits();
+        }
+    }
+
+    public function exits()
+    {
+        $exitCode = proc_close($this->process);
+        $this->process = null;
+
+        if ($this->signalCode) {
+            $this->handleExit(null, $this->signalCode);
+        } else {
+            $this->handleExit($exitCode, null);
+        }
+    }
+
+    public function handleExit($exitCode, $signalCode)
+    {
+        if ($this->exited) {
+            return;
+        }
+
+        $this->exited = true;
+
+        $this->exitCode = $exitCode;
+        $this->signalCode = $signalCode;
+
+        $this->emit('exit', array($exitCode, $signalCode));
+        $this->emit('close', array($exitCode, $signalCode));
+    }
+
+    public function getPid()
+    {
+        $status = $this->getCachedStatus();
+
+        return $status['pid'];
+    }
+
+    public function getCommand()
+    {
+        $status = $this->getCachedStatus();
+
+        return $status['command'];
+    }
+
+    public function isRunning()
+    {
+        if ($this->exited) {
+            return false;
+        } else {
+            $status = $this->getFreshStatus();
+
+            return $status['running'];
+        }
+    }
+
+    public function isSignaled()
+    {
+        $status = $this->getFreshStatus();
+
+        return $status['signaled'];
+    }
+
+    public function isStopped()
+    {
+        $status = $this->getFreshStatus();
+
+        return $status['stopped'];
+    }
+
+    public function terminate($signalCode = self::SIGNAL_CODE_SIGTERM)
+    {
+        proc_terminate($this->process, $signalCode);
+    }
+
+    public function getExitCode()
+    {
+        return $this->exitCode;
+    }
+
+    public function getSignalCode()
+    {
+        return $this->signalCode;
+    }
+
+    private function getCachedStatus()
+    {
+        if (is_null($this->status)) {
+            $this->updateStatus();
+        }
+
+        return $this->status;
+    }
+
+    private function getFreshStatus()
+    {
+        $this->updateStatus();
+
+        return $this->status;
+    }
+}

src/React/ChildProcess/Stream.php

@@ -0,0 +1,19 @@
+<?php
+
+namespace React\ChildProcess;
+
+use React\Stream\Stream as DefaultStream;
+
+class Stream extends DefaultStream
+{
+    public function handleData($stream)
+    {
+        $data = fread($stream, $this->bufferSize);
+
+        if ('' === $data || false === $data) {
+            $this->end();
+        } else {
+            $this->emit('data', array($data, $this));
+        }
+    }
+}

tests/React/Tests/ChildProcess/FactoryTest.php

@@ -0,0 +1,102 @@
+<?php
+
+namespace React\Tests\ChildProcess;
+
+use React\ChildProcess\Factory;
+use React\EventLoop\StreamSelectLoop;
+
+class FactoryTest extends \PHPUnit_Framework_TestCase
+{
+    public function testSpawn()
+    {
+        $loop    = $this->createLoop();
+        $factory = $this->createFactory($loop);
+        $process = $factory->spawn('php', array('-r', 'echo "cwd = ", getcwd(), ", count of env = ", count($_ENV);'));
+
+        $capturedData = '';
+
+        $process->stdout->on('data', function ($data) use (&$capturedData) {
+            $capturedData .= $data;
+        });
+
+        $loop->run();
+
+        $cwd = getcwd();
+        $this->assertSame("cwd = {$cwd}, count of env = 0", $capturedData);
+    }
+
+    public function testSpawnWithPwd()
+    {
+        $loop    = $this->createLoop();
+        $factory = $this->createFactory($loop);
+        $process = $factory->spawn('php', array('-r', 'echo "cwd = ", getcwd();'), '/');
+
+        $capturedData = '';
+
+        $process->stdout->on('data', function ($data) use (&$capturedData) {
+            $capturedData .= $data;
+        });
+
+        $loop->run();
+
+        $this->assertSame('cwd = /', $capturedData);
+    }
+
+    public function testSpawnWithEnv()
+    {
+        if (isset($_SERVER['TRAVIS']) && 'true' === $_SERVER['TRAVIS']) {
+            $this->markTestSkipped();
+        }
+
+        $loop    = $this->createLoop();
+        $factory = $this->createFactory($loop);
+        $process = $factory->spawn('php', array('-r', 'echo "foo = ", $_ENV["foo"];'), null, array('foo' => 'FOO'));
+
+        $capturedData = '';
+
+        $process->stdout->on('data', function ($data) use (&$capturedData) {
+            $capturedData .= $data;
+        });
+
+        $loop->run();
+
+        $this->assertSame('foo = FOO', $capturedData);
+    }
+
+    public function testCommand()
+    {
+        $loop    = $this->createLoop();
+        $factory = $this->createFactory($loop);
+        $process = $factory->spawn('echo');
+
+        $this->assertSame('echo', $process->getCommand());
+    }
+
+    public function testCommandWithOneArgument()
+    {
+        $loop    = $this->createLoop();
+        $factory = $this->createFactory($loop);
+        $process = $factory->spawn('echo', array('foo'));
+
+        $this->assertSame("echo 'foo'", $process->getCommand());
+    }
+
+    public function testCommandWithManyArguments()
+    {
+        $loop    = $this->createLoop();
+        $factory = $this->createFactory($loop);
+        $process = $factory->spawn('echo', array('foo', 'bar', 'foo bar'));
+
+        $this->assertSame("echo 'foo' 'bar' 'foo bar'", $process->getCommand());
+    }
+
+    private function createFactory($loop)
+    {
+        return new Factory($loop);
+    }
+
+    private function createLoop()
+    {
+        return new StreamSelectLoop();
+    }
+}