danog/process
1
0
Fork 0
mirror of https://github.com/danog/process.git synced 2024-11-30 04:39:04 +01:00
Code Issues Projects Releases Wiki Activity

Fixup after PR #5, add some more tests

Browse Source 
Also removing status() again
This commit is contained in:
Bob Weinand 2015-09-16 19:56:15 +02:00
parent 618e4c9465
commit d2da56638f
12 changed files with 258 additions and 252 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.php_cs
View File

@ -9,7 +9,7 @@ return Symfony\CS\Config\Config::create()
])
->finder(
Symfony\CS\Finder\DefaultFinder::create()
->in(__DIR__ . "/lib")
->in(__DIR__ . "/tests")
->in(__DIR__)
->exclude(__DIR__."/vendor")
)
;

181
Process.php Normal file
View File

@ -0,0 +1,181 @@
<?php
namespace Amp;
class Process {
private $cmd;
private $options;
private $proc;
private $stdin;
private $stdout;
private $stderr;
private $deferred;
private $writeDeferreds = [];
private $writeBuf;
private $writeTotal;
private $writeCur;
const BUFFER_NONE = 0;
const BUFFER_STDOUT = 1;
const BUFFER_STDERR = 2;
const BUFFER_ALL = 3;
/* $options are passed directly to proc_open(), "cwd" and "env" entries are passed as fourth respectively fifth parameters to proc_open() */
public function __construct($cmd, array $options = []) {
$this->cmd = $cmd;
$this->options = $options;
}
/**
* @param int $buffer one of the self::BUFFER_* constants. Determines whether it will buffer the stdout and/or stderr data internally
* @return Promise is updated with ["out", $data] or ["err", $data] for data received on stdout or stderr
* That Promise will be resolved to a stdClass object with stdout, stderr (when $buffer is true), exit (holding exit code) and signal (only present when terminated via signal) properties
*/
public function exec($buffer = self::BUFFER_NONE) {
if ($this->proc) {
throw new \RuntimeException("Process was already launched");
}
$fds = [["pipe", "r"], ["pipe", "w"], ["pipe", "w"]];
$cwd = isset($this->options["cwd"]) ? $this->options["cwd"] : NULL;
$env = isset($this->options["env"]) ? $this->options["env"] : NULL;
if (!$this->proc = @proc_open($this->cmd, $fds, $pipes, $cwd, $env, $this->options)) {
return new Failure(new \RuntimeException("Failed executing command: $this->cmd"));
}
$this->writeBuf = "";
$this->writeTotal = 0;
$this->writeCur = 0;
stream_set_blocking($pipes[0], false);
stream_set_blocking($pipes[1], false);
stream_set_blocking($pipes[2], false);
$this->deferred = new Deferred;
$result = new \stdClass;
if ($buffer & self::BUFFER_STDOUT) {
$result->stdout = "";
}
if ($buffer & self::BUFFER_STDERR) {
$result->stderr = "";
}
$this->stdout = \Amp\onReadable($pipes[1], function($watcher, $sock) use ($result) {
if ("" == $data = @fread($sock, 8192)) {
\Amp\cancel($watcher);
\Amp\cancel($this->stdin);
\Amp\immediately(function() use ($result) {
$status = proc_get_status($this->proc);
assert($status["running"] === false);
if ($status["signaled"]) {
$result->signal = $status["termsig"];
}
$result->exit = $status["exitcode"];
$this->deferred->succeed($result);
foreach ($this->writeDeferreds as $deferred) {
$deferred->fail(new \Exception("Write could not be completed, process finished"));
}
$this->writeDeferreds = [];
});
} else {
if (isset($result->stdout)) {
$result->stdout .= $data;
}
$this->deferred->update(["out", $data]);
}
});
$this->stderr = \Amp\onReadable($pipes[2], function($watcher, $sock) use ($result) {
if ("" == $data = @fread($sock, 8192)) {
\Amp\cancel($watcher);
} else {
if (isset($result->stderr)) {
$result->stderr .= $data;
}
$this->deferred->update(["err", $data]);
}
});
$this->stdin = \Amp\onWritable($pipes[0], function($watcher, $sock) {
$this->writeCur += @fwrite($sock, $this->writeBuf);
if ($this->writeCur == $this->writeTotal) {
\Amp\disable($watcher);
}
while (($next = key($this->writeDeferreds)) !== null && $next <= $this->writeCur) {
$this->writeDeferreds[$next]->succeed($this->writeCur);
unset($this->writeDeferreds[$next]);
}
}, ["enable" => false]);
return $this->deferred->promise();
}
/* Only kills process, Promise returned by exec() will succeed in the next tick */
public function kill($signal = 15) {
if ($this->proc) {
return proc_terminate($this->proc, $signal);
}
return false;
}
/* Aborts all watching completely and immediately */
public function cancel($signal = 9) {
if (!$this->proc) {
return;
}
$this->kill($signal);
\Amp\cancel($this->stdout);
\Amp\cancel($this->stderr);
\Amp\cancel($this->stdin);
$this->deferred->fail(new \RuntimeException("Process watching was cancelled"));
foreach ($this->writeDeferreds as $deferred) {
$deferred->fail(new \Exception("Write could not be completed, process watching was cancelled"));
}
$this->writeDeferreds = [];
}
/**
* @return Promise which will succeed after $str was written. It will contain the total number of already written bytes to the process
*/
public function write($str) {
assert(strlen($str) > 0);
if (!$this->proc) {
throw new \RuntimeException("Process was not yet launched");
}
$this->writeBuf .= $str;
\Amp\enable($this->stdin);
$this->writeTotal += strlen($str);
$deferred = $this->writeDeferreds[$this->writeTotal] = new Deferred;
return $deferred->promise();
}
/* Returns the process identifier (PID) of the executed process, if applicable. */
public function pid() {
if ($this->proc === null) {
return null;
}
return proc_get_status($this->proc)["pid"];
}
/* Returns the command to execute. */
public function command() {
return $this->cmd;
}
/* Returns the options the process is run with. */
public function options() {
return $this->options;
}
}

12
composer.json
View File

@ -6,23 +6,19 @@
"amphp/amp": "^1"
},
"require-dev": {
"phpunit/phpunit": "^4.8",
"fabpot/php-cs-fixer": "~1.9"
"phpunit/phpunit": "^4.8",
"fabpot/php-cs-fixer": "~1.9"
},
"license": "MIT",
"authors": [
{
"name": "Bob Weinand",
"email": "bobwei9@hotmail.com"
},
{
"name": "Edoardo Biraghi",
"email": "edoardo.biraghi@gmail.com"
}
],
"autoload": {
"classmap": [
{"Amp\\Process": "lib/Process.php"}
{"Amp\\Process": "Process.php"}
]
}
}
}

8
examples/001_basic_command.php
View File

@ -5,8 +5,8 @@ use Amp\Process;
include __DIR__."/../vendor/autoload.php";
\Amp\run(function() {
$proc = new Process("echo 1");
$result = (yield $proc->exec(Process::BUFFER_ALL));
$proc = new Process("echo 1");
$result = (yield $proc->exec(Process::BUFFER_ALL));
var_dump($result->stdout); // "1"
});
var_dump($result->stdout); // "1"
});

16
examples/002_write_command.php
View File

@ -5,14 +5,14 @@ use Amp\Process;
include __DIR__."/../vendor/autoload.php";
\Amp\run(function() {
$proc = new Process('read ; echo "$REPLY"');
$promise = $proc->exec(Process::BUFFER_ALL);
$proc = new Process('read ; echo "$REPLY"');
$promise = $proc->exec(Process::BUFFER_ALL);
/* send to stdin */
$proc->write("abc\n");
/* send to stdin */
$proc->write("abc\n");
/* wait for process end */
$result = (yield $promise);
/* wait for process end */
$result = (yield $promise);
var_dump($result->stdout); // "abc"
});
var_dump($result->stdout); // "abc"
});

24
examples/003_watch_live.php
View File

@ -5,18 +5,18 @@ use Amp\Process;
include __DIR__."/../vendor/autoload.php";
\Amp\run(function() {
$proc = new Process("echo 1; sleep 1; echo 2; sleep 1; echo 3");
$promise = $proc->exec();
$proc = new Process("echo 1; sleep 1; echo 2; sleep 1; echo 3");
$promise = $proc->exec();
$promise->watch(function($data) {
// $data[0] is either "out" or "err", $data[1] the actual data
list($type, $msg) = $data;
// "1" ... 2 seconds ... "2" ... 2 seconds ... "3"
print "$type: $msg";
});
$promise->watch(function($data) {
// $data[0] is either "out" or "err", $data[1] the actual data
list($type, $msg) = $data;
// "1" ... 2 seconds ... "2" ... 2 seconds ... "3"
print "$type: $msg";
});
$result = (yield $promise);
$result = (yield $promise);
// we aren't buffering by default (Process::BUFFER_NONE is default) ... so only exit code present and eventually the killing signal
var_dump($result);
});
// we aren't buffering by default (Process::BUFFER_NONE is default) ... so only exit code present and eventually the killing signal
var_dump($result);
});

188
lib/Process.php
View File

@ -1,188 +0,0 @@
<?php
namespace Amp;
class Process {
private $cmd;
private $options;
private $proc;
private $stdin;
private $stdout;
private $stderr;
private $deferred;
private $writeDeferreds = [];
private $writeBuf;
private $writeTotal;
private $writeCur;
const BUFFER_NONE = 0;
const BUFFER_STDOUT = 1;
const BUFFER_STDERR = 2;
const BUFFER_ALL = 3;
/* $options are passed directly to proc_open(), "cwd" and "env" entries are passed as fourth respectively fifth parameters to proc_open() */
public function __construct($cmd, array $options = []) {
$this->cmd = $cmd;
$this->options = $options;
}
/**
* @param int $buffer one of the self::BUFFER_* constants. Determines whether it will buffer the stdout and/or stderr data internally
* @return Promise is updated with ["out", $data] or ["err", $data] for data received on stdout or stderr
* That Promise will be resolved to a stdClass object with stdout, stderr (when $buffer is true), exit (holding exit code) and signal (only present when terminated via signal) properties
*/
public function exec($buffer = self::BUFFER_NONE) {
if ($this->proc) {
throw new \RuntimeException("Process was already launched");
}
$fds = [["pipe", "r"], ["pipe", "w"], ["pipe", "w"]];
$cwd = isset($this->options["cwd"]) ? $this->options["cwd"] : NULL;
$env = isset($this->options["env"]) ? $this->options["env"] : NULL;
if (!$this->proc = @proc_open($this->cmd, $fds, $pipes, $cwd, $env, $this->options)) {
return new Failure(new \RuntimeException("Failed executing command: $this->cmd"));
}
$this->writeBuf = "";
$this->writeTotal = 0;
$this->writeCur = 0;
stream_set_blocking($pipes[0], false);
stream_set_blocking($pipes[1], false);
stream_set_blocking($pipes[2], false);
$this->deferred = new Deferred;
$result = new \stdClass;
if ($buffer & self::BUFFER_STDOUT) {
$result->stdout = "";
}
if ($buffer & self::BUFFER_STDERR) {
$result->stderr = "";
}
$this->stdout = \Amp\onReadable($pipes[1], function($watcher, $sock) use ($result) {
if ("" == $data = @fread($sock, 8192)) {
\Amp\cancel($watcher);
\Amp\cancel($this->stdin);
\Amp\immediately(function() use ($result) {
$status = proc_get_status($this->proc);
assert($status["running"] === false);
if ($status["signaled"]) {
$result->signal = $status["termsig"];
}
$result->exit = $status["exitcode"];
$this->deferred->succeed($result);
foreach ($this->writeDeferreds as $deferred) {
$deferred->fail(new \Exception("Write could not be completed, process finished"));
}
$this->writeDeferreds = [];
});
} else {
isset($result->stdout) && $result->stdout .= $data;
$this->deferred->update(["out", $data]);
}
});
$this->stderr = \Amp\onReadable($pipes[2], function($watcher, $sock) use ($result) {
if ("" == $data = @fread($sock, 8192)) {
\Amp\cancel($watcher);
} else {
isset($result->stderr) && $result->stderr .= $data;
$this->deferred->update(["err", $data]);
}
});
$this->stdin = \Amp\onWritable($pipes[0], function($watcher, $sock) {
$this->writeCur += @fwrite($sock, $this->writeBuf);
if ($this->writeCur == $this->writeTotal) {
\Amp\disable($watcher);
}
while (($next = key($this->writeDeferreds)) !== null && $next <= $this->writeCur) {
$this->writeDeferreds[$next]->succeed($this->writeCur);
unset($this->writeDeferreds[$next]);
}
}, ["enable" => false]);
return $this->deferred->promise();
}
/* Only kills process, Promise returned by exec() will succeed in the next tick */
public function kill($signal = 15) {
if ($this->proc) {
return proc_terminate($this->proc, $signal);
}
return false;
}
/* Aborts all watching completely and immediately */
public function cancel($signal = 9) {
if (!$this->proc) {
return;
}
$this->kill($signal);
\Amp\cancel($this->stdout);
\Amp\cancel($this->stderr);
\Amp\cancel($this->stdin);
$this->deferred->fail(new \RuntimeException("Process watching was cancelled"));
foreach ($this->writeDeferreds as $deferred) {
$deferred->fail(new \Exception("Write could not be completed, process watching was cancelled"));
}
$this->writeDeferreds = [];
}
/**
* @return Promise which will succeed after $str was written. It will contain the total number of already written bytes to the process
*/
public function write($str) {
assert(strlen($str) > 0);
if (!$this->proc) {
throw new \RuntimeException("Process was not yet launched");
}
$this->writeBuf .= $str;
\Amp\enable($this->stdin);
$this->writeTotal += strlen($str);
$deferred = $this->writeDeferreds[$this->writeTotal] = new Deferred;
return $deferred->promise();
}
/**
* Alias for \proc_get_status()
*
* @return array|null
*/
public function status() {
if ($this->proc === null) {
return null;
}
return proc_get_status($this->proc);
}
/**
* Returns the process identifier (PID) of the executed process, if applicable.
*
* @return int|null
*/
public function pid() {
return $this->status()["pid"];
}
/**
* Returns the command of the current executed process.
*
* @return string
*/
public function getCommand() {
return $this->cmd;
}
}

69
tests/AbstractProcessTest.php
View File

@ -4,12 +4,7 @@ namespace Amp\Process\Test;
use Amp\Process;
/**
* Class AbstractProcessTest
* @package Amp\Process\Test
*/
abstract class AbstractProcessTest extends \PHPUnit_Framework_TestCase {
const CMD_PROCESS = 'echo foo';
abstract function testReactor();
@ -26,50 +21,76 @@ abstract class AbstractProcessTest extends \PHPUnit_Framework_TestCase {
}
public function testCommandCanRun() {
\Amp\run(function(){
\Amp\run(function() {
$process = new Process(self::CMD_PROCESS);
$this->assertNull($process->status());
$this->assertNull($process->pid());
$promise = $process->exec();
$this->assertArrayHasKey('running', $process->status());
$this->assertArrayHasKey('pid', $process->status());
$this->assertTrue($process->status()['running']);
$completed = false;
$promise->when(function() use (&$completed) { $completed = true; });
$this->assertFalse($completed);
$this->assertInternalType('int', $process->pid());
});
}
public function testProcessResolvePromise()
{
\Amp\run(function(){
public function testProcessResolvePromise() {
\Amp\run(function() {
$process = new Process(self::CMD_PROCESS);
$promise = $process->exec();
$promise = $process->exec(Process::BUFFER_ALL);
$this->assertInstanceOf('\Amp\Promise', $promise);
$return = (yield $promise);
$this->assertObjectHasAttribute('exit', $return);
$this->assertInternalType('int', $return->exit);
$this->assertObjectHasAttribute('stdout', $return);
$this->assertSame("foo\n", $return->stdout);
$this->assertObjectHasAttribute('stderr', $return);
$this->assertSame("", $return->stderr);
});
}
public function testKillSignals()
{
\Amp\run(function(){
public function testKillSignals() {
\Amp\run(function() {
$process = new Process(self::CMD_PROCESS);
$promise = $process->exec();
$process->kill();
$return = (yield $promise);
$this->assertObjectHasAttribute('signal', $return);
$this->assertObjectHasAttribute('exit', $return);
$this->assertInternalType('int', $return->signal);
$this->assertInternalType('int', $return->exit);
$this->assertEquals(15, $return->signal);
$this->assertEquals(-1, $return->exit);
$this->assertSame(15, $return->signal);
$this->assertSame(-1, $return->exit);
});
}
public function testWatch() {
\Amp\run(function() {
$process = new Process(self::CMD_PROCESS);
$this->assertNull($process->pid());
$promise = $process->exec();
$msg = "";
$promise->watch(function($update) use (&$msg) {
list($type, $partMsg) = $update;
$this->assertSame("out", $type);
$msg .= $partMsg;
});
yield $promise;
$this->assertSame("foo\n", $msg);
});
public function testGetCommand() {
$process = new Process(self::CMD_PROCESS);
$this->assertSame(self::CMD_PROCESS, $process->getCommand());
}
public function testCommand() {
$process = new Process(self::CMD_PROCESS);
$this->assertSame(self::CMD_PROCESS, $process->command());
}
public function testOptions() {
$process = new Process(self::CMD_PROCESS);
$this->assertSame([], $process->options());
}
}

1
tests/LibeventReactorProcessTest.php
View File

@ -5,7 +5,6 @@ namespace Amp\Process\Test;
use Amp\LibeventReactor;
class LibeventReactorProcessTest extends AbstractProcessTest {
protected function setUp() {
if (extension_loaded("libevent")) {
\Amp\reactor($assign = new LibeventReactor);

2
tests/NativeReactorProcessTest.php
View File

@ -2,11 +2,9 @@
namespace Amp\Process\Test;
use Amp\NativeReactor;
class NativeReactorProcessTest extends AbstractProcessTest {
public function setUp(){
\Amp\reactor(new NativeReactor());
}

1
tests/UvReactorProcessTest.php
View File

@ -5,7 +5,6 @@ namespace Amp\Process\Test;
use Amp\UvReactor;
class UvReactorProcessTest extends AbstractProcessTest {
public function setUp(){
if (extension_loaded("uv")) {
\Amp\reactor($assign = new UvReactor);

4
tests/bootstrap.php
View File

@ -1,5 +1,5 @@
<?php
namespace Amp\Test;
require __DIR__ . "/../vendor/autoload.php";
error_reporting(E_ALL);
error_reporting(E_ALL);