1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
|
<?php
namespace React\Stream;
use Evenement\EventEmitter;
use React\EventLoop\LoopInterface;
use InvalidArgumentException;
final class DuplexResourceStream extends EventEmitter implements DuplexStreamInterface
{
private $stream;
private $loop;
/**
* Controls the maximum buffer size in bytes to read at once from the stream.
*
* This can be a positive number which means that up to X bytes will be read
* at once from the underlying stream resource. Note that the actual number
* of bytes read may be lower if the stream resource has less than X bytes
* currently available.
*
* This can be `-1` which means read everything available from the
* underlying stream resource.
* This should read until the stream resource is not readable anymore
* (i.e. underlying buffer drained), note that this does not neccessarily
* mean it reached EOF.
*
* @var int
*/
private $bufferSize;
private $buffer;
private $readable = true;
private $writable = true;
private $closing = false;
private $listening = false;
public function __construct($stream, LoopInterface $loop, $readChunkSize = null, WritableStreamInterface $buffer = null)
{
if (!is_resource($stream) || get_resource_type($stream) !== "stream") {
throw new InvalidArgumentException('First parameter must be a valid stream resource');
}
// ensure resource is opened for reading and wrting (fopen mode must contain "+")
$meta = stream_get_meta_data($stream);
if (isset($meta['mode']) && $meta['mode'] !== '' && strpos($meta['mode'], '+') === false) {
throw new InvalidArgumentException('Given stream resource is not opened in read and write mode');
}
// this class relies on non-blocking I/O in order to not interrupt the event loop
// e.g. pipes on Windows do not support this: https://bugs.php.net/bug.php?id=47918
if (stream_set_blocking($stream, 0) !== true) {
throw new \RuntimeException('Unable to set stream resource to non-blocking mode');
}
// Use unbuffered read operations on the underlying stream resource.
// Reading chunks from the stream may otherwise leave unread bytes in
// PHP's stream buffers which some event loop implementations do not
// trigger events on (edge triggered).
// This does not affect the default event loop implementation (level
// triggered), so we can ignore platforms not supporting this (HHVM).
// Pipe streams (such as STDIN) do not seem to require this and legacy
// PHP versions cause SEGFAULTs on unbuffered pipe streams, so skip this.
if (function_exists('stream_set_read_buffer') && !$this->isLegacyPipe($stream)) {
stream_set_read_buffer($stream, 0);
}
if ($buffer === null) {
$buffer = new WritableResourceStream($stream, $loop);
}
$this->stream = $stream;
$this->loop = $loop;
$this->bufferSize = ($readChunkSize === null) ? 65536 : (int)$readChunkSize;
$this->buffer = $buffer;
$that = $this;
$this->buffer->on('error', function ($error) use ($that) {
$that->emit('error', array($error));
});
$this->buffer->on('close', array($this, 'close'));
$this->buffer->on('drain', function () use ($that) {
$that->emit('drain');
});
$this->resume();
}
public function isReadable()
{
return $this->readable;
}
public function isWritable()
{
return $this->writable;
}
public function pause()
{
if ($this->listening) {
$this->loop->removeReadStream($this->stream);
$this->listening = false;
}
}
public function resume()
{
if (!$this->listening && $this->readable) {
$this->loop->addReadStream($this->stream, array($this, 'handleData'));
$this->listening = true;
}
}
public function write($data)
{
if (!$this->writable) {
return false;
}
return $this->buffer->write($data);
}
public function close()
{
if (!$this->writable && !$this->closing) {
return;
}
$this->closing = false;
$this->readable = false;
$this->writable = false;
$this->emit('close');
$this->pause();
$this->buffer->close();
$this->removeAllListeners();
if (is_resource($this->stream)) {
fclose($this->stream);
}
}
public function end($data = null)
{
if (!$this->writable) {
return;
}
$this->closing = true;
$this->readable = false;
$this->writable = false;
$this->pause();
$this->buffer->end($data);
}
public function pipe(WritableStreamInterface $dest, array $options = array())
{
return Util::pipe($this, $dest, $options);
}
/** @internal */
public function handleData($stream)
{
$error = null;
set_error_handler(function ($errno, $errstr, $errfile, $errline) use (&$error) {
$error = new \ErrorException(
$errstr,
0,
$errno,
$errfile,
$errline
);
});
$data = stream_get_contents($stream, $this->bufferSize);
restore_error_handler();
if ($error !== null) {
$this->emit('error', array(new \RuntimeException('Unable to read from stream: ' . $error->getMessage(), 0, $error)));
$this->close();
return;
}
if ($data !== '') {
$this->emit('data', array($data));
} else{
// no data read => we reached the end and close the stream
$this->emit('end');
$this->close();
}
}
/**
* Returns whether this is a pipe resource in a legacy environment
*
* This works around a legacy PHP bug (#61019) that was fixed in PHP 5.4.28+
* and PHP 5.5.12+ and newer.
*
* @param resource $resource
* @return bool
* @link https://github.com/reactphp/child-process/issues/40
*
* @codeCoverageIgnore
*/
private function isLegacyPipe($resource)
{
if (PHP_VERSION_ID < 50428 || (PHP_VERSION_ID >= 50500 && PHP_VERSION_ID < 50512)) {
$meta = stream_get_meta_data($resource);
if (isset($meta['stream_type']) && $meta['stream_type'] === 'STDIO') {
return true;
}
}
return false;
}
}
|
