-
Notifications
You must be signed in to change notification settings - Fork 559
Expand file tree
/
Copy pathProcess.php
More file actions
152 lines (129 loc) · 3.48 KB
/
Process.php
File metadata and controls
152 lines (129 loc) · 3.48 KB
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
<?php declare(strict_types = 1);
namespace PHPStan\Parallel;
use Exception;
use PHPStan\ShouldNotHappenException;
use React\EventLoop\LoopInterface;
use React\EventLoop\TimerInterface;
use React\Stream\ReadableStreamInterface;
use React\Stream\WritableStreamInterface;
use Throwable;
use function fclose;
use function is_string;
use function rewind;
use function sprintf;
use function stream_get_contents;
use function tmpfile;
class Process
{
public \React\ChildProcess\Process $process;
private WritableStreamInterface $in;
/** @var resource */
private $stdOut;
/** @var resource */
private $stdErr;
/** @var callable(mixed[] $json) : void */
private $onData;
/** @var callable(Throwable $exception): void */
private $onError;
private ?TimerInterface $timer = null;
public function __construct(
private string $command,
private LoopInterface $loop,
private float $timeoutSeconds,
)
{
}
/**
* @param callable(mixed[] $json) : void $onData
* @param callable(Throwable $exception): void $onError
* @param callable(?int $exitCode, string $output) : void $onExit
*/
public function start(callable $onData, callable $onError, callable $onExit): void
{
$tmpStdOut = tmpfile();
if ($tmpStdOut === false) {
throw new ShouldNotHappenException('Failed creating temp file for stdout.');
}
$tmpStdErr = tmpfile();
if ($tmpStdErr === false) {
throw new ShouldNotHappenException('Failed creating temp file for stderr.');
}
$this->stdOut = $tmpStdOut;
$this->stdErr = $tmpStdErr;
$this->process = new \React\ChildProcess\Process($this->command, null, null, [
1 => $this->stdOut,
2 => $this->stdErr,
]);
$this->process->start($this->loop);
$this->onData = $onData;
$this->onError = $onError;
$this->process->on('exit', function ($exitCode) use ($onExit): void {
$this->cancelTimer();
$output = '';
rewind($this->stdOut);
$stdOut = stream_get_contents($this->stdOut);
if (is_string($stdOut)) {
$output .= $stdOut;
}
rewind($this->stdErr);
$stdErr = stream_get_contents($this->stdErr);
if (is_string($stdErr)) {
$output .= $stdErr;
}
$onExit($exitCode, $output);
fclose($this->stdOut);
fclose($this->stdErr);
});
}
private function cancelTimer(): void
{
if ($this->timer === null) {
return;
}
$this->loop->cancelTimer($this->timer);
$this->timer = null;
}
/**
* @param mixed[] $data
*/
public function request(array $data): void
{
$this->cancelTimer();
$this->in->write($data);
$this->timer = $this->loop->addTimer($this->timeoutSeconds, function (): void {
$onError = $this->onError;
$onError(new Exception(sprintf('Child process timed out after %.1f seconds. Try making it longer with parallel.processTimeout setting.', $this->timeoutSeconds)));
});
}
public function quit(): void
{
$this->cancelTimer();
if (!$this->process->isRunning()) {
return;
}
foreach ($this->process->pipes as $pipe) {
$pipe->close();
}
$this->in->end();
}
public function bindConnection(ReadableStreamInterface $out, WritableStreamInterface $in): void
{
$out->on('data', function (array $json): void {
$this->cancelTimer();
if ($json['action'] !== 'result') {
return;
}
$onData = $this->onData;
$onData($json['result']);
});
$this->in = $in;
$out->on('error', function (Throwable $error): void {
$onError = $this->onError;
$onError($error);
});
$in->on('error', function (Throwable $error): void {
$onError = $this->onError;
$onError($error);
});
}
}