22042339290

Committed 15 Feb 2026 08:14PM UTC coverage: 92.957% (-0.2%) from 93.171%

Build # 22042339290

Build Type

push

github

Committed by

keradus

Commit Message

test: check PHP env in CI jobs

Run Details

29302 of 31522 relevant lines covered (92.96%)

44.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

12.82

/src/Runner/Parallel/Process.php

<?php

declare(strict_types=1);

/*
 * This file is part of PHP CS Fixer.
 *
 * (c) Fabien Potencier <fabien@symfony.com>
 *     Dariusz Rumiński <dariusz.ruminski@gmail.com>
 *
 * This source file is subject to the MIT license that is bundled
 * with this source code in the file LICENSE.
 */

namespace PhpCsFixer\Runner\Parallel;

use React\ChildProcess\Process as ReactProcess;
use React\EventLoop\LoopInterface;
use React\EventLoop\TimerInterface;
use React\Stream\ReadableStreamInterface;
use React\Stream\WritableStreamInterface;

/**
 * Represents single process that is handled within parallel run.
 * Inspired by:
 *   - https://github.com/phpstan/phpstan-src/blob/9ce425bca5337039fb52c0acf96a20a2b8ace490/src/Parallel/Process.php
 *   - https://github.com/phpstan/phpstan-src/blob/1477e752b4b5893f323b6d2c43591e68b3d85003/src/Process/ProcessHelper.php.
 *
 * @author Greg Korba <greg@codito.dev>
 *
 * @internal
 *
 * @no-named-arguments Parameter names are not covered by the backward compatibility promise.
 */
final class Process
{
    // Properties required for process instantiation
    private string $command;
    private LoopInterface $loop;
    private int $timeoutSeconds;

    // Properties required for process execution
    private ?ReactProcess $process = null;
    private ?WritableStreamInterface $in = null;

    /** @var resource */
    private $stdErr;

    /** @var resource */
    private $stdOut;

    /** @var callable(array<array-key, mixed>): void */
    private $onData;

    /** @var callable(\Throwable): void */
    private $onError;

    private ?TimerInterface $timer = null;

    public function __construct(string $command, LoopInterface $loop, int $timeoutSeconds)
    {
        $this->command = $command;
        $this->loop = $loop;
        $this->timeoutSeconds = $timeoutSeconds;
    }

    /**
     * @param callable(array<array-key, mixed> $json): void  $onData  callback to be called when data is received from the parallelisation operator
     * @param callable(\Throwable $exception): void          $onError callback to be called when an exception occurs
     * @param callable(?int $exitCode, string $output): void $onExit  callback to be called when the process exits
     */
    public function start(callable $onData, callable $onError, callable $onExit): void
    {
        $sysTempDir = sys_get_temp_dir();
        if (!is_writable($sysTempDir)) {
            throw new ParallelisationException(\sprintf(
                'Failed creating temp file as sys_get_temp_dir="%s" is not writable.',
                $sysTempDir,
            ));
        }

        $stdOut = tmpfile();
        if (false === $stdOut) {
            throw new ParallelisationException('Failed creating temp file for stdOut.');
        }
        $this->stdOut = $stdOut;

        $stdErr = tmpfile();
        if (false === $stdErr) {
            throw new ParallelisationException('Failed creating temp file for stdErr.');
        }
        $this->stdErr = $stdErr;

        $this->onData = $onData;
        $this->onError = $onError;

        $this->process = new ReactProcess($this->command, null, null, [
            1 => $this->stdOut,
            2 => $this->stdErr,
        ]);
        $this->process->start($this->loop);
        $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);
        });
    }

    /**
     * Handles requests from parallelisation operator to its worker (spawned process).
     *
     * @param array<array-key, mixed> $data
     */
    public function request(array $data): void
    {
        $this->cancelTimer(); // Configured process timeout actually means "chunk timeout" (each request resets timer)

        if (null === $this->in) {
            throw new ParallelisationException(
                'Process not connected with parallelisation operator, ensure `bindConnection()` was called',
            );
        }

        $this->in->write($data);
        $this->timer = $this->loop->addTimer($this->timeoutSeconds, function (): void {
            ($this->onError)(
                new \Exception(
                    \sprintf(
                        'Child process timed out after %d seconds. Try making it longer using `ParallelConfig`.',
                        $this->timeoutSeconds,
                    ),
                )
            );
        });
    }

    public function quit(): void
    {
        $this->cancelTimer();
        if (null === $this->process || !$this->process->isRunning()) {
            return;
        }

        foreach ($this->process->pipes as $pipe) {
            $pipe->close();
        }

        if (null === $this->in) {
            return;
        }

        $this->in->end();
    }

    public function bindConnection(ReadableStreamInterface $out, WritableStreamInterface $in): void
    {
        $this->in = $in;

        $in->on('error', function (\Throwable $error): void {
            ($this->onError)($error);
        });

        $out->on('data', function (array $json): void {
            $this->cancelTimer();

            // Pass everything to the parallelisation operator, it should decide how to handle the data
            ($this->onData)($json);
        });
        $out->on('error', function (\Throwable $error): void {
            ($this->onError)($error);
        });
    }

    private function cancelTimer(): void
    {
        if (null === $this->timer) {
            return;
        }

        $this->loop->cancelTimer($this->timer);
        $this->timer = null;
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	/*
6	* This file is part of PHP CS Fixer.
7	*
8	* (c) Fabien Potencier <fabien@symfony.com>
9	* Dariusz Rumiński <dariusz.ruminski@gmail.com>
10	*
11	* This source file is subject to the MIT license that is bundled
12	* with this source code in the file LICENSE.
13	*/
14
15	namespace PhpCsFixer\Runner\Parallel;
16
17	use React\ChildProcess\Process as ReactProcess;
18	use React\EventLoop\LoopInterface;
19	use React\EventLoop\TimerInterface;
20	use React\Stream\ReadableStreamInterface;
21	use React\Stream\WritableStreamInterface;
22
23	/**
24	* Represents single process that is handled within parallel run.
25	* Inspired by:
26	* - https://github.com/phpstan/phpstan-src/blob/9ce425bca5337039fb52c0acf96a20a2b8ace490/src/Parallel/Process.php
27	* - https://github.com/phpstan/phpstan-src/blob/1477e752b4b5893f323b6d2c43591e68b3d85003/src/Process/ProcessHelper.php.
28	*
29	* @author Greg Korba <greg@codito.dev>
30	*
31	* @internal
32	*
33	* @no-named-arguments Parameter names are not covered by the backward compatibility promise.
34	*/
35	final class Process
36	{
37	// Properties required for process instantiation
38	private string $command;
39	private LoopInterface $loop;
40	private int $timeoutSeconds;
41
42	// Properties required for process execution
43	private ?ReactProcess $process = null;
44	private ?WritableStreamInterface $in = null;
45
46	/** @var resource */
47	private $stdErr;
48
49	/** @var resource */
50	private $stdOut;
51
52	/** @var callable(array<array-key, mixed>): void */
53	private $onData;
54
55	/** @var callable(\Throwable): void */
56	private $onError;
57
58	private ?TimerInterface $timer = null;
59
60	public function __construct(string $command, LoopInterface $loop, int $timeoutSeconds)
61	{
62	$this->command = $command;	1✔
63	$this->loop = $loop;	1✔
64	$this->timeoutSeconds = $timeoutSeconds;	1✔
65	}
66
67	/**
68	* @param callable(array<array-key, mixed> $json): void $onData callback to be called when data is received from the parallelisation operator
69	* @param callable(\Throwable $exception): void $onError callback to be called when an exception occurs
70	* @param callable(?int $exitCode, string $output): void $onExit callback to be called when the process exits
71	*/
72	public function start(callable $onData, callable $onError, callable $onExit): void
73	{
74	$sysTempDir = sys_get_temp_dir();	×
75	if (!is_writable($sysTempDir)) {	×
76	throw new ParallelisationException(\sprintf(	×
77	'Failed creating temp file as sys_get_temp_dir="%s" is not writable.',	×
78	$sysTempDir,	×
79	));	×
80	}
81
82	$stdOut = tmpfile();	×
83	if (false === $stdOut) {	×
84	throw new ParallelisationException('Failed creating temp file for stdOut.');	×
85	}
86	$this->stdOut = $stdOut;	×
87
88	$stdErr = tmpfile();	×
89	if (false === $stdErr) {	×
90	throw new ParallelisationException('Failed creating temp file for stdErr.');	×
91	}
92	$this->stdErr = $stdErr;	×
93
94	$this->onData = $onData;	×
95	$this->onError = $onError;	×
96
97	$this->process = new ReactProcess($this->command, null, null, [	×
98	1 => $this->stdOut,	×
99	2 => $this->stdErr,	×
100	]);	×
101	$this->process->start($this->loop);	×
102	$this->process->on('exit', function ($exitCode) use ($onExit): void {	×
103	$this->cancelTimer();	×
104
105	$output = '';	×
106	rewind($this->stdOut);	×
107	$stdOut = stream_get_contents($this->stdOut);	×
108	if (\is_string($stdOut)) {	×
109	$output .= $stdOut;	×
110	}
111
112	rewind($this->stdErr);	×
113	$stdErr = stream_get_contents($this->stdErr);	×
114	if (\is_string($stdErr)) {	×
115	$output .= $stdErr;	×
116	}
117
118	$onExit($exitCode, $output);	×
119
120	fclose($this->stdOut);	×
121	fclose($this->stdErr);	×
122	});	×
123	}
124
125	/**
126	* Handles requests from parallelisation operator to its worker (spawned process).
127	*
128	* @param array<array-key, mixed> $data
129	*/
130	public function request(array $data): void
131	{
132	$this->cancelTimer(); // Configured process timeout actually means "chunk timeout" (each request resets timer)	1✔
133
134	if (null === $this->in) {	1✔
135	throw new ParallelisationException(	1✔
136	'Process not connected with parallelisation operator, ensure `bindConnection()` was called',	1✔
137	);	1✔
138	}
139
140	$this->in->write($data);	×
141	$this->timer = $this->loop->addTimer($this->timeoutSeconds, function (): void {	×
142	($this->onError)(	×
143	new \Exception(	×
144	\sprintf(	×
145	'Child process timed out after %d seconds. Try making it longer using `ParallelConfig`.',	×
146	$this->timeoutSeconds,	×
147	),	×
148	)	×
149	);	×
150	});	×
151	}
152
153	public function quit(): void
154	{
155	$this->cancelTimer();	×
156	if (null === $this->process \|\| !$this->process->isRunning()) {	×
157	return;	×
158	}
159
160	foreach ($this->process->pipes as $pipe) {	×
161	$pipe->close();	×
162	}
163
164	if (null === $this->in) {	×
165	return;	×
166	}
167
168	$this->in->end();	×
169	}
170
171	public function bindConnection(ReadableStreamInterface $out, WritableStreamInterface $in): void
172	{
173	$this->in = $in;	×
174
175	$in->on('error', function (\Throwable $error): void {	×
176	($this->onError)($error);	×
177	});	×
178
179	$out->on('data', function (array $json): void {	×
180	$this->cancelTimer();	×
181
182	// Pass everything to the parallelisation operator, it should decide how to handle the data
183	($this->onData)($json);	×
184	});	×
185	$out->on('error', function (\Throwable $error): void {	×
186	($this->onError)($error);	×
187	});	×
188	}
189
190	private function cancelTimer(): void
191	{
192	if (null === $this->timer) {	1✔
193	return;	1✔
194	}
195
196	$this->loop->cancelTimer($this->timer);	×
197	$this->timer = null;	×
198	}
199	}

keradus / PHP-CS-Fixer / 22042339290

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous