WIP: ChildProcess

examples/child-process.php

@@ -0,0 +1,53 @@
+<?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, blue('[PID]') . '    pid is ', (string) $process->getPid(), PHP_EOL;
+    echo $idLabel, blue('[CMD]') . '    ', $process->getCommand(), PHP_EOL;
+
+    $process->stdout->on('data', function ($data) use ($idLabel) {
+        echo $idLabel, green('[STDOUT]') . ' ';
+        var_dump($data);
+    });
+
+    $process->stderr->on('data', function ($data) use ($idLabel) {
+        echo $idLabel, red('[STDERR]') . ' ';
+        var_dump($data);
+    });
+
+    $process->on('exit', function ($status) use ($idLabel) {
+        echo $idLabel, yellow('[EXIT]') . '   exited with status code ', (string) $status, PHP_EOL;
+    });
+}
+
+$loop->run();
+
+function red($str) {
+    return "\033[31m{$str}\033[0m";
+}
+
+function green($str) {
+    return "\033[32m{$str}\033[0m";
+}
+
+function yellow($str) {
+    return "\033[33m{$str}\033[0m";
+}
+
+function blue($str) {
+    return "\033[34m{$str}\033[0m";
+}

src/React/ChildProcess/Factory.php

@@ -0,0 +1,55 @@
+<?php
+
+namespace React\ChildProcess;
+
+use React\EventLoop\LoopInterface;
+use React\Stream\Stream;
+use React\ChildProcess\Process;
+
+class Factory
+{
+    private $loop;
+
+    public function __construct(LoopInterface $loop)
+    {
+        $this->loop = $loop;
+    }
+
+    public function spawn($file, $args = NULL, $options = NULL)
+    {
+        $args = $args ? $args : array();
+
+        $cmd = $this->createCommand($file, $args);
+        $cwd = isset($options['cwd']) ? $options['cwd'] : NULL;
+        $env = isset($options['env']) ? $options['env'] : NULL;
+
+        $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 createCommand($file, $args)
+    {
+        $command = $file;
+        if (count($args) > 0) {
+            $command .= ' ' . join(' ', array_map('escapeshellarg', $args));
+        }
+        return $command;
+    }
+}

src/React/ChildProcess/Process.php

@@ -0,0 +1,91 @@
+<?php
+
+namespace React\ChildProcess;
+
+use React\Stream\WritableStreamInterface;
+use React\Stream\ReadableStreamInterface;
+use Evenement\EventEmitter;
+
+class Process extends EventEmitter
+{
+    public $stdin;
+
+    public $stdout;
+
+    public $stderr;
+
+    private $status = NULL;
+
+    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, $stderr) {
+            if ($stderr->isReadable() === false) {
+                $self->handleExit();
+            }
+        });
+
+        $this->stderr->on('end', function () use ($self, $stdout) {
+            if ($stdout->isReadable() === false) {
+                $self->handleExit();
+            }
+        });
+    }
+
+    public function handleExit()
+    {
+        $status = proc_close($this->process);
+        $this->emit('exit', array($status));
+        $this->emit('close', array($status));
+    }
+
+    public function getPid()
+    {
+        $status = $this->getCachedStatus();
+        return $status['pid'];
+    }
+
+    public function getCommand()
+    {
+        $status = $this->getCachedStatus();
+        return $status['command'];
+    }
+
+    public function isRunning()
+    {
+        $status = $this->getFreshStatus();
+        return $status['running'];
+    }
+
+    public function isSignaled()
+    {
+        $status = $this->getFreshStatus();
+        return $status['signaled'];
+    }
+
+    public function isStopped()
+    {
+        $status = $this->getFreshStatus();
+        return $status['stopped'];
+    }
+
+    private function getCachedStatus()
+    {
+        if (is_null($this->status)) {
+            $this->status = proc_get_status($this->process);
+        }
+        return $this->status;
+    }
+
+    private function getFreshStatus()
+    {
+        $this->status = proc_get_status($this->process);
+        return $this->status;
+    }
+}