forked from phpstan/phpstan-src
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathProcessBase.php
More file actions
107 lines (90 loc) · 2.53 KB
/
Copy pathProcessBase.php
File metadata and controls
107 lines (90 loc) · 2.53 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
<?php declare(strict_types = 1);
namespace PHPStan\Parallel;
use PHPStan\ShouldNotHappenException;
use React\EventLoop\LoopInterface;
use React\EventLoop\TimerInterface;
use React\Stream\ReadableStreamInterface;
use React\Stream\WritableStreamInterface;
use Throwable;
use function sprintf;
/**
* Process-creation-agnostic half of a parallel worker: the TCP/NDJSON
* connection plus the per-request timeout timer. Subclasses only implement
* start() (how the worker process is created) and quit() (how it is torn down).
*/
abstract class ProcessBase implements Process
{
private ?WritableStreamInterface $in = null;
/** @var callable(mixed[] $json) : void */
private $onData;
/** @var callable(Throwable $exception): void */
private $onError;
private ?TimerInterface $timer = null;
public function __construct(
protected LoopInterface $loop,
protected float $timeoutSeconds,
)
{
}
/**
* @param callable(mixed[] $json) : void $onData
* @param callable(Throwable $exception): void $onError
*/
protected function setCallbacks(callable $onData, callable $onError): void
{
$this->onData = $onData;
$this->onError = $onError;
}
protected function cancelTimer(): void
{
if ($this->timer === null) {
return;
}
$this->loop->cancelTimer($this->timer);
$this->timer = null;
}
/** Cancels the timeout timer and ends the writable side of the connection. */
protected function endConnection(): void
{
$this->cancelTimer();
if ($this->in === null) {
return;
}
$this->in->end();
}
/**
* @param mixed[] $data
*/
public function request(array $data): void
{
$this->cancelTimer();
if ($this->in === null) {
throw new ShouldNotHappenException();
}
$this->in->write($data);
$this->timer = $this->loop->addTimer($this->timeoutSeconds, function (): void {
$onError = $this->onError;
$onError(new ProcessTimedOutException(sprintf('Child process timed out after %.1f seconds. Try making it longer with parallel.processTimeout setting.', $this->timeoutSeconds)));
});
}
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);
});
}
}