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 40f0f10

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

File tree

5 files changed

+173
-23
lines changed
Filter options

5 files changed

+173
-23
lines changed

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

Copy file name to clipboard
+90Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
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+
* @param resource|scalar|\Traversable|null The input to append as stream resource, scalar or \Traversable
39+
*/
40+
public function write($input)
41+
{
42+
if (null === $input) {
43+
return;
44+
}
45+
if ($this->isClosed()) {
46+
throw new RuntimeException(sprintf('%s is closed', static::class));
47+
}
48+
$this->input[] = ProcessUtils::validateInput(__METHOD__, $input);
49+
}
50+
51+
/**
52+
* Closes the write buffer.
53+
*/
54+
public function close()
55+
{
56+
$this->open = false;
57+
}
58+
59+
/**
60+
* Tells whether the write buffer is closed or not.
61+
*/
62+
public function isClosed()
63+
{
64+
return !$this->open;
65+
}
66+
67+
public function getIterator()
68+
{
69+
$this->open = true;
70+
71+
while ($this->open || $this->input) {
72+
if (!$this->input) {
73+
yield '';
74+
continue;
75+
}
76+
$current = array_shift($this->input);
77+
78+
if ($current instanceof \Iterator) {
79+
foreach ($current as $cur) {
80+
yield $cur;
81+
}
82+
} else {
83+
yield $current;
84+
}
85+
if (!$this->input && $this->open && null !== $onEmpty = $this->onEmpty) {
86+
$this->write($onEmpty($this));
87+
}
88+
}
89+
}
90+
}

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

Copy file name to clipboardExpand all lines: src/Symfony/Component/Process/Pipes/AbstractPipes.php
+15-3Lines changed: 15 additions & 3 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
*
@@ -23,7 +25,7 @@ abstract class AbstractPipes implements PipesInterface
2325

2426
/** @var string */
2527
private $inputBuffer = '';
26-
/** @var resource|\Iterator|null */
28+
/** @var resource|scalar|\Iterator|null */
2729
private $input;
2830
/** @var bool */
2931
private $blocked = true;
@@ -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/Process.php

Copy file name to clipboardExpand all lines: src/Symfony/Component/Process/Process.php
+3-3Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,7 @@ class Process
132132
* @param string $commandline The command line to run
133133
* @param string|null $cwd The working directory or null to use the working dir of the current PHP process
134134
* @param array|null $env The environment variables or null to use the same environment as the current PHP process
135-
* @param string|null $input The input
135+
* @param mixed|null $input The input as stream resource, scalar or \Traversable, or null for no input
136136
* @param int|float|null $timeout The timeout in seconds or null to disable
137137
* @param array $options An array of options for proc_open
138138
*
@@ -1027,7 +1027,7 @@ public function setEnv(array $env)
10271027
/**
10281028
* Gets the Process input.
10291029
*
1030-
* @return null|string The Process input
1030+
* @return resource|string|\Iterator|null The Process input
10311031
*/
10321032
public function getInput()
10331033
{
@@ -1039,7 +1039,7 @@ public function getInput()
10391039
*
10401040
* This content will be passed to the underlying process standard input.
10411041
*
1042-
* @param mixed $input The content
1042+
* @param resource|scalar|\Traversable|null $input The content
10431043
*
10441044
* @return self The current Process instance
10451045
*

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

Copy file name to clipboardExpand all lines: src/Symfony/Component/Process/ProcessBuilder.php
+1-1Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,7 @@ public function addEnvironmentVariables(array $variables)
167167
/**
168168
* Sets the input of the process.
169169
*
170-
* @param mixed $input The input as a string
170+
* @param resource|scalar|\Traversable|null $input The input content
171171
*
172172
* @return ProcessBuilder
173173
*

‎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.