Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit fd984ef

Browse filesBrowse files
nicolas-grekasnicolas-grekas
committed
[Process] Add InputStream to seamlessly feed running processes
1 parent 6ed73d5 commit fd984ef
Copy full SHA for fd984ef

File tree

3 files changed

+166
-18
lines changed
Filter options

3 files changed

+166
-18
lines changed

‎src/Symfony/Component/Process/InputStream.php

Copy file name to clipboard
+88Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Process;
13+
14+
use Symfony\Component\Process\Exception\RuntimeException;
15+
16+
/**
17+
* Provides a way to continuously write to the input of a Process until the InputStream is closed.
18+
*
19+
* @author Nicolas Grekas <p@tchwork.com>
20+
*/
21+
class InputStream implements \IteratorAggregate
22+
{
23+
private $onEmpty = null;
24+
private $input = array();
25+
private $open = true;
26+
27+
/**
28+
* Sets a callback that is called when the write buffer becomes empty.
29+
*/
30+
public function onEmpty(callable $onEmpty = null)
31+
{
32+
$this->onEmpty = $onEmpty;
33+
}
34+
35+
/**
36+
* Appends an input to the write buffer.
37+
*/
38+
public function write($input)
39+
{
40+
if (null === $input) {
41+
return;
42+
}
43+
if ($this->isClosed()) {
44+
throw new RuntimeException(sprintf('%s is closed', static::class));
45+
}
46+
$this->input[] = ProcessUtils::validateInput(__METHOD__, $input);
47+
}
48+
49+
/**
50+
* Closes the write buffer.
51+
*/
52+
public function close()
53+
{
54+
$this->open = false;
55+
}
56+
57+
/**
58+
* Tells whether the write buffer is closed or not.
59+
*/
60+
public function isClosed()
61+
{
62+
return !$this->open;
63+
}
64+
65+
public function getIterator()
66+
{
67+
$this->open = true;
68+
69+
while ($this->open || $this->input) {
70+
if (!$this->input) {
71+
yield '';
72+
continue;
73+
}
74+
$current = array_shift($this->input);
75+
76+
if ($current instanceof \Iterator) {
77+
foreach ($current as $cur) {
78+
yield $cur;
79+
}
80+
} else {
81+
yield $current;
82+
}
83+
if (!$this->input && $this->open && null !== $onEmpty = $this->onEmpty) {
84+
$this->write($onEmpty($this));
85+
}
86+
}
87+
}
88+
}

‎src/Symfony/Component/Process/Pipes/AbstractPipes.php

Copy file name to clipboardExpand all lines: src/Symfony/Component/Process/Pipes/AbstractPipes.php
+14-2Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111

1212
namespace Symfony\Component\Process\Pipes;
1313

14+
use Symfony\Component\Process\Exception\InvalidArgumentException;
15+
1416
/**
1517
* @author Romain Neutron <imprec@gmail.com>
1618
*
@@ -84,6 +86,8 @@ protected function unblock()
8486

8587
/**
8688
* Writes input to stdin.
89+
*
90+
* @throws InvalidArgumentException When an input iterator yields a non supported value
8791
*/
8892
protected function write()
8993
{
@@ -97,10 +101,18 @@ protected function write()
97101
$input = null;
98102
} elseif (is_resource($input = $input->current())) {
99103
stream_set_blocking($input, 0);
100-
} else {
101-
$this->inputBuffer .= $input;
104+
} elseif (!isset($this->inputBuffer[0])) {
105+
if (!is_string($input)) {
106+
if (!is_scalar($input)) {
107+
throw new InvalidArgumentException(sprintf('%s yielded a value of type "%s", but only scalars and stream resources are supported', get_class($this->input), gettype($input)));
108+
}
109+
$input = (string) $input;
110+
}
111+
$this->inputBuffer = $input;
102112
$this->input->next();
103113
$input = null;
114+
} else {
115+
$input = null;
104116
}
105117
}
106118

‎src/Symfony/Component/Process/Tests/ProcessTest.php

Copy file name to clipboardExpand all lines: src/Symfony/Component/Process/Tests/ProcessTest.php
+64-16Lines changed: 64 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
use Symfony\Component\Process\Exception\LogicException;
1515
use Symfony\Component\Process\Exception\ProcessTimedOutException;
1616
use Symfony\Component\Process\Exception\RuntimeException;
17+
use Symfony\Component\Process\InputStream;
1718
use Symfony\Component\Process\PhpExecutableFinder;
1819
use Symfony\Component\Process\Pipes\PipesInterface;
1920
use Symfony\Component\Process\Process;
@@ -1176,29 +1177,76 @@ public function provideVariousIncrementals() {
11761177

11771178
public function testIteratorInput()
11781179
{
1179-
$nextData = 'ping';
1180-
$input = function () use (&$nextData) {
1181-
while (false !== $nextData) {
1182-
yield $nextData;
1183-
yield $nextData = '';
1184-
}
1180+
$input = function () {
1181+
yield 'ping';
1182+
yield 'pong';
11851183
};
1186-
$input = $input();
11871184

1188-
$process = new Process(self::$phpBin.' -r '.escapeshellarg('stream_copy_to_stream(STDIN, STDOUT);'));
1185+
$process = new Process(self::$phpBin.' -r '.escapeshellarg('stream_copy_to_stream(STDIN, STDOUT);'), null, null, $input());
1186+
$process->run();
1187+
$this->assertSame('pingpong', $process->getOutput());
1188+
}
1189+
1190+
public function testSimpleInputStream()
1191+
{
1192+
$input = new InputStream();
1193+
1194+
$process = new Process(self::$phpBin.' -r '.escapeshellarg('echo \'ping\'; stream_copy_to_stream(STDIN, STDOUT);'));
11891195
$process->setInput($input);
1190-
$process->start(function ($type, $data) use ($input, &$nextData) {
1196+
1197+
$process->start(function ($type, $data) use ($input) {
11911198
if ('ping' === $data) {
1192-
$h = fopen('php://memory', 'r+');
1193-
fwrite($h, 'pong');
1194-
rewind($h);
1195-
$nextData = $h;
1196-
$input->next();
1197-
} else {
1198-
$nextData = false;
1199+
$input->write('pang');
1200+
} elseif (!$input->isClosed()) {
1201+
$input->write('pong');
1202+
$input->close();
11991203
}
12001204
});
12011205

1206+
$process->wait();
1207+
$this->assertSame('pingpangpong', $process->getOutput());
1208+
}
1209+
1210+
public function testInputStreamWithCallable()
1211+
{
1212+
$i = 0;
1213+
$stream = fopen('php://memory', 'w+');
1214+
$stream = function () use ($stream, &$i) {
1215+
if ($i < 3) {
1216+
rewind($stream);
1217+
fwrite($stream, ++$i);
1218+
rewind($stream);
1219+
1220+
return $stream;
1221+
}
1222+
};
1223+
1224+
$input = new InputStream();
1225+
$input->onEmpty($stream);
1226+
$input->write($stream());
1227+
1228+
$process = new Process(self::$phpBin.' -r '.escapeshellarg('stream_copy_to_stream(STDIN, STDOUT);'));
1229+
$process->setInput($input);
1230+
$process->start(function ($type, $data) use ($input) {
1231+
$input->close();
1232+
});
1233+
1234+
$process->wait();
1235+
$this->assertSame('123', $process->getOutput());
1236+
}
1237+
1238+
public function testInputStreamWithGenerator()
1239+
{
1240+
$input = new InputStream();
1241+
$input->onEmpty(function ($input) {
1242+
yield 'pong';
1243+
$input->close();
1244+
});
1245+
1246+
$process = new Process(self::$phpBin.' -r '.escapeshellarg('stream_copy_to_stream(STDIN, STDOUT);'));
1247+
$process->setInput($input);
1248+
$process->start();
1249+
$input->write('ping');
12021250
$process->wait();
12031251
$this->assertSame('pingpong', $process->getOutput());
12041252
}

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.