16253811376

Committed 13 Jul 2025 09:50PM UTC coverage: 94.794% (+0.001%) from 94.793%

Build # 16253811376

Build Type

push

github

Committed by

web-flow

Commit Message

Merge branch 'master' into sf8

Run Details

2272 of 2348 new or added lines in 338 files covered. (96.76%)

7 existing lines in 5 files now uncovered.

28259 of 29811 relevant lines covered (94.79%)

45.39 hits per line

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

62.3

/src/Console/Command/WorkerCommand.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\Console\Command;

use Clue\React\NDJson\Decoder;
use Clue\React\NDJson\Encoder;
use PhpCsFixer\Cache\NullCacheManager;
use PhpCsFixer\Config;
use PhpCsFixer\Console\ConfigurationResolver;
use PhpCsFixer\Error\ErrorsManager;
use PhpCsFixer\Runner\Event\FileProcessed;
use PhpCsFixer\Runner\Parallel\ParallelAction;
use PhpCsFixer\Runner\Parallel\ParallelConfigFactory;
use PhpCsFixer\Runner\Parallel\ParallelisationException;
use PhpCsFixer\Runner\Runner;
use PhpCsFixer\ToolInfoInterface;
use React\EventLoop\StreamSelectLoop;
use React\Socket\ConnectionInterface;
use React\Socket\TcpConnector;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\ConsoleOutputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\EventDispatcher\EventDispatcher;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

/**
 * @author Greg Korba <greg@codito.dev>
 *
 * @internal
 */
#[AsCommand(name: 'worker', description: 'Internal command for running fixers in parallel', hidden: true)]
final class WorkerCommand extends Command
{
    /** @var string Prefix used before JSON-encoded error printed in the worker's process */
    public const ERROR_PREFIX = 'WORKER_ERROR::';

    protected static $defaultName = 'worker';

    protected static $defaultDescription = 'Internal command for running fixers in parallel';

    private ToolInfoInterface $toolInfo;
    private ConfigurationResolver $configurationResolver;
    private ErrorsManager $errorsManager;
    private EventDispatcherInterface $eventDispatcher;

    /** @var list<FileProcessed> */
    private array $events;

    public function __construct(ToolInfoInterface $toolInfo)
    {
        parent::__construct();

        $this->setHidden(true);
        $this->toolInfo = $toolInfo;
        $this->errorsManager = new ErrorsManager();
        $this->eventDispatcher = new EventDispatcher();
    }

    protected function configure(): void
    {
        $this->setDefinition(
            [
                new InputOption('port', null, InputOption::VALUE_REQUIRED, 'Specifies parallelisation server\'s port.'),
                new InputOption('identifier', null, InputOption::VALUE_REQUIRED, 'Specifies parallelisation process\' identifier.'),
                new InputOption('allow-risky', '', InputOption::VALUE_REQUIRED, 'Are risky fixers allowed (can be `yes` or `no`).'),
                new InputOption('config', '', InputOption::VALUE_REQUIRED, 'The path to a config file.'),
                new InputOption('dry-run', '', InputOption::VALUE_NONE, 'Only shows which files would have been modified.'),
                new InputOption('rules', '', InputOption::VALUE_REQUIRED, 'List of rules that should be run against configured paths.'),
                new InputOption('using-cache', '', InputOption::VALUE_REQUIRED, 'Should cache be used (can be `yes` or `no`).'),
                new InputOption('cache-file', '', InputOption::VALUE_REQUIRED, 'The path to the cache file.'),
                new InputOption('diff', '', InputOption::VALUE_NONE, 'Prints diff for each file.'),
                new InputOption('stop-on-violation', '', InputOption::VALUE_NONE, 'Stop execution on first violation.'),
            ]
        );
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $errorOutput = $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output;
        $identifier = $input->getOption('identifier');
        $port = $input->getOption('port');

        if (null === $identifier || !is_numeric($port)) {
            throw new ParallelisationException('Missing parallelisation options');
        }

        try {
            $runner = $this->createRunner($input);
        } catch (\Throwable $e) {
            throw new ParallelisationException('Unable to create runner: '.$e->getMessage(), 0, $e);
        }

        $loop = new StreamSelectLoop();
        $tcpConnector = new TcpConnector($loop);
        $tcpConnector
            ->connect(\sprintf('127.0.0.1:%d', $port))
            ->then(
                /** @codeCoverageIgnore */
                function (ConnectionInterface $connection) use ($loop, $runner, $identifier): void {
                    $out = new Encoder($connection, \JSON_INVALID_UTF8_IGNORE);
                    $in = new Decoder($connection, true, 512, \JSON_INVALID_UTF8_IGNORE);

                    // [REACT] Initialise connection with the parallelisation operator
                    $out->write(['action' => ParallelAction::WORKER_HELLO, 'identifier' => $identifier]);

                    $handleError = static function (\Throwable $error) use ($out): void {
                        $out->write([
                            'action' => ParallelAction::WORKER_ERROR_REPORT,
                            'class' => \get_class($error),
                            'message' => $error->getMessage(),
                            'file' => $error->getFile(),
                            'line' => $error->getLine(),
                            'code' => $error->getCode(),
                            'trace' => $error->getTraceAsString(),
                        ]);
                    };
                    $out->on('error', $handleError);
                    $in->on('error', $handleError);

                    // [REACT] Listen for messages from the parallelisation operator (analysis requests)
                    $in->on('data', function (array $json) use ($loop, $runner, $out): void {
                        $action = $json['action'] ?? null;

                        // Parallelisation operator does not have more to do, let's close the connection
                        if (ParallelAction::RUNNER_THANK_YOU === $action) {
                            $loop->stop();

                            return;
                        }

                        if (ParallelAction::RUNNER_REQUEST_ANALYSIS !== $action) {
                            // At this point we only expect analysis requests, if any other action happen, we need to fix the code.
                            throw new \LogicException(\sprintf('Unexpected action ParallelAction::%s.', $action));
                        }

                        /** @var iterable<int, string> $files */
                        $files = $json['files'];

                        foreach ($files as $path) {
                            // Reset events because we want to collect only those coming from analysed files chunk
                            $this->events = [];
                            $runner->setFileIterator(new \ArrayIterator([new \SplFileInfo($path)]));
                            $analysisResult = $runner->fix();

                            if (1 !== \count($this->events)) {
                                throw new ParallelisationException('Runner did not report a fixing event or reported too many.');
                            }

                            if (1 < \count($analysisResult)) {
                                throw new ParallelisationException('Runner returned more analysis results than expected.');
                            }

                            $out->write([
                                'action' => ParallelAction::WORKER_RESULT,
                                'file' => $path,
                                'fileHash' => $this->events[0]->getFileHash(),
                                'status' => $this->events[0]->getStatus(),
                                'fixInfo' => array_pop($analysisResult),
                                'errors' => $this->errorsManager->forPath($path),
                            ]);
                        }

                        // Request another file chunk (if available, the parallelisation operator will request new "run" action)
                        $out->write(['action' => ParallelAction::WORKER_GET_FILE_CHUNK]);
                    });
                },
                static function (\Throwable $error) use ($errorOutput): void {
                    // @TODO Verify onRejected behaviour → https://github.com/PHP-CS-Fixer/PHP-CS-Fixer/pull/7777#discussion_r1590399285
                    $errorOutput->writeln($error->getMessage());
                }
            )
        ;

        $loop->run();

        return Command::SUCCESS;
    }

    private function createRunner(InputInterface $input): Runner
    {
        $passedConfig = $input->getOption('config');
        $passedRules = $input->getOption('rules');

        if (null !== $passedConfig && null !== $passedRules) {
            throw new \RuntimeException('Passing both `--config` and `--rules` options is not allowed');
        }

        // There's no one single source of truth when it comes to fixing single file, we need to collect statuses from events.
        $this->eventDispatcher->addListener(FileProcessed::NAME, function (FileProcessed $event): void {
            $this->events[] = $event;
        });

        $this->configurationResolver = new ConfigurationResolver(
            new Config(),
            [
                'allow-risky' => $input->getOption('allow-risky'),
                'config' => $passedConfig,
                'dry-run' => $input->getOption('dry-run'),
                'rules' => $passedRules,
                'path' => [],
                'path-mode' => ConfigurationResolver::PATH_MODE_OVERRIDE, // IMPORTANT! WorkerCommand is called with file that already passed filtering, so here we can rely on PATH_MODE_OVERRIDE.
                'using-cache' => $input->getOption('using-cache'),
                'cache-file' => $input->getOption('cache-file'),
                'diff' => $input->getOption('diff'),
                'stop-on-violation' => $input->getOption('stop-on-violation'),
            ],
            getcwd(), // @phpstan-ignore-line
            $this->toolInfo
        );

        return new Runner(
            null, // Paths are known when parallelisation server requests new chunk, not now
            $this->configurationResolver->getFixers(),
            $this->configurationResolver->getDiffer(),
            $this->eventDispatcher,
            $this->errorsManager,
            $this->configurationResolver->getLinter(),
            $this->configurationResolver->isDryRun(),
            new NullCacheManager(), // IMPORTANT! We pass null cache, as cache is read&write in main process and we do not need to do it again.
            $this->configurationResolver->getDirectory(),
            $this->configurationResolver->shouldStopOnViolation(),
            ParallelConfigFactory::sequential(), // IMPORTANT! Worker must run in sequential mode.
            null,
            $this->configurationResolver->getConfigFile()
        );
    }
}

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\Console\Command;
16
17	use Clue\React\NDJson\Decoder;
18	use Clue\React\NDJson\Encoder;
19	use PhpCsFixer\Cache\NullCacheManager;
20	use PhpCsFixer\Config;
21	use PhpCsFixer\Console\ConfigurationResolver;
22	use PhpCsFixer\Error\ErrorsManager;
23	use PhpCsFixer\Runner\Event\FileProcessed;
24	use PhpCsFixer\Runner\Parallel\ParallelAction;
25	use PhpCsFixer\Runner\Parallel\ParallelConfigFactory;
26	use PhpCsFixer\Runner\Parallel\ParallelisationException;
27	use PhpCsFixer\Runner\Runner;
28	use PhpCsFixer\ToolInfoInterface;
29	use React\EventLoop\StreamSelectLoop;
30	use React\Socket\ConnectionInterface;
31	use React\Socket\TcpConnector;
32	use Symfony\Component\Console\Attribute\AsCommand;
33	use Symfony\Component\Console\Command\Command;
34	use Symfony\Component\Console\Input\InputInterface;
35	use Symfony\Component\Console\Input\InputOption;
36	use Symfony\Component\Console\Output\ConsoleOutputInterface;
37	use Symfony\Component\Console\Output\OutputInterface;
38	use Symfony\Component\EventDispatcher\EventDispatcher;
39	use Symfony\Component\EventDispatcher\EventDispatcherInterface;
40
41	/**
42	* @author Greg Korba <greg@codito.dev>
43	*
44	* @internal
45	*/
46	#[AsCommand(name: 'worker', description: 'Internal command for running fixers in parallel', hidden: true)]
47	final class WorkerCommand extends Command
48	{
49	/** @var string Prefix used before JSON-encoded error printed in the worker's process */
50	public const ERROR_PREFIX = 'WORKER_ERROR::';
51
52	protected static $defaultName = 'worker';
53
54	protected static $defaultDescription = 'Internal command for running fixers in parallel';
55
56	private ToolInfoInterface $toolInfo;
57	private ConfigurationResolver $configurationResolver;
58	private ErrorsManager $errorsManager;
59	private EventDispatcherInterface $eventDispatcher;
60
61	/** @var list<FileProcessed> */
62	private array $events;
63
64	public function __construct(ToolInfoInterface $toolInfo)
65	{
66	parent::__construct();	5✔
67
68	$this->setHidden(true);	5✔
69	$this->toolInfo = $toolInfo;	5✔
70	$this->errorsManager = new ErrorsManager();	5✔
71	$this->eventDispatcher = new EventDispatcher();	5✔
72	}
73
74	protected function configure(): void
75	{
76	$this->setDefinition(	5✔
77	[	5✔
78	new InputOption('port', null, InputOption::VALUE_REQUIRED, 'Specifies parallelisation server\'s port.'),	5✔
79	new InputOption('identifier', null, InputOption::VALUE_REQUIRED, 'Specifies parallelisation process\' identifier.'),	5✔
80	new InputOption('allow-risky', '', InputOption::VALUE_REQUIRED, 'Are risky fixers allowed (can be `yes` or `no`).'),	5✔
81	new InputOption('config', '', InputOption::VALUE_REQUIRED, 'The path to a config file.'),	5✔
82	new InputOption('dry-run', '', InputOption::VALUE_NONE, 'Only shows which files would have been modified.'),	5✔
83	new InputOption('rules', '', InputOption::VALUE_REQUIRED, 'List of rules that should be run against configured paths.'),	5✔
84	new InputOption('using-cache', '', InputOption::VALUE_REQUIRED, 'Should cache be used (can be `yes` or `no`).'),	5✔
85	new InputOption('cache-file', '', InputOption::VALUE_REQUIRED, 'The path to the cache file.'),	5✔
86	new InputOption('diff', '', InputOption::VALUE_NONE, 'Prints diff for each file.'),	5✔
87	new InputOption('stop-on-violation', '', InputOption::VALUE_NONE, 'Stop execution on first violation.'),	5✔
88	]	5✔
89	);	5✔
90	}
91
92	protected function execute(InputInterface $input, OutputInterface $output): int
93	{
94	$errorOutput = $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output;	3✔
95	$identifier = $input->getOption('identifier');	3✔
96	$port = $input->getOption('port');	3✔
97
98	if (null === $identifier \|\| !is_numeric($port)) {	3✔
99	throw new ParallelisationException('Missing parallelisation options');	2✔
100	}
101
102	try {
103	$runner = $this->createRunner($input);	1✔
104	} catch (\Throwable $e) {	×
105	throw new ParallelisationException('Unable to create runner: '.$e->getMessage(), 0, $e);	×
106	}
107
108	$loop = new StreamSelectLoop();	1✔
109	$tcpConnector = new TcpConnector($loop);	1✔
110	$tcpConnector	1✔
111	->connect(\sprintf('127.0.0.1:%d', $port))	1✔
112	->then(	1✔
113	/** @codeCoverageIgnore */
114	function (ConnectionInterface $connection) use ($loop, $runner, $identifier): void {	1✔
NEW 115	$out = new Encoder($connection, \JSON_INVALID_UTF8_IGNORE);	×
NEW 116	$in = new Decoder($connection, true, 512, \JSON_INVALID_UTF8_IGNORE);	×
117
118	// [REACT] Initialise connection with the parallelisation operator
119	$out->write(['action' => ParallelAction::WORKER_HELLO, 'identifier' => $identifier]);	×
120
121	$handleError = static function (\Throwable $error) use ($out): void {	×
122	$out->write([	×
123	'action' => ParallelAction::WORKER_ERROR_REPORT,	×
124	'class' => \get_class($error),	×
125	'message' => $error->getMessage(),	×
126	'file' => $error->getFile(),	×
127	'line' => $error->getLine(),	×
128	'code' => $error->getCode(),	×
129	'trace' => $error->getTraceAsString(),	×
130	]);	×
131	};	×
132	$out->on('error', $handleError);	×
133	$in->on('error', $handleError);	×
134
135	// [REACT] Listen for messages from the parallelisation operator (analysis requests)
136	$in->on('data', function (array $json) use ($loop, $runner, $out): void {	×
137	$action = $json['action'] ?? null;	×
138
139	// Parallelisation operator does not have more to do, let's close the connection
140	if (ParallelAction::RUNNER_THANK_YOU === $action) {	×
141	$loop->stop();	×
142
143	return;	×
144	}
145
146	if (ParallelAction::RUNNER_REQUEST_ANALYSIS !== $action) {	×
147	// At this point we only expect analysis requests, if any other action happen, we need to fix the code.
148	throw new \LogicException(\sprintf('Unexpected action ParallelAction::%s.', $action));	×
149	}
150
151	/** @var iterable<int, string> $files */
152	$files = $json['files'];	×
153
154	foreach ($files as $path) {	×
155	// Reset events because we want to collect only those coming from analysed files chunk
156	$this->events = [];	×
157	$runner->setFileIterator(new \ArrayIterator([new \SplFileInfo($path)]));	×
158	$analysisResult = $runner->fix();	×
159
160	if (1 !== \count($this->events)) {	×
161	throw new ParallelisationException('Runner did not report a fixing event or reported too many.');	×
162	}
163
164	if (1 < \count($analysisResult)) {	×
165	throw new ParallelisationException('Runner returned more analysis results than expected.');	×
166	}
167
168	$out->write([	×
169	'action' => ParallelAction::WORKER_RESULT,	×
170	'file' => $path,	×
171	'fileHash' => $this->events[0]->getFileHash(),	×
172	'status' => $this->events[0]->getStatus(),	×
173	'fixInfo' => array_pop($analysisResult),	×
174	'errors' => $this->errorsManager->forPath($path),	×
175	]);	×
176	}
177
178	// Request another file chunk (if available, the parallelisation operator will request new "run" action)
179	$out->write(['action' => ParallelAction::WORKER_GET_FILE_CHUNK]);	×
180	});	×
181	},	1✔
182	static function (\Throwable $error) use ($errorOutput): void {	1✔
183	// @TODO Verify onRejected behaviour → https://github.com/PHP-CS-Fixer/PHP-CS-Fixer/pull/7777#discussion_r1590399285
184	$errorOutput->writeln($error->getMessage());	1✔
185	}	1✔
186	)	1✔
187	;	1✔
188
189	$loop->run();	1✔
190
191	return Command::SUCCESS;	1✔
192	}
193
194	private function createRunner(InputInterface $input): Runner
195	{
196	$passedConfig = $input->getOption('config');	1✔
197	$passedRules = $input->getOption('rules');	1✔
198
199	if (null !== $passedConfig && null !== $passedRules) {	1✔
200	throw new \RuntimeException('Passing both `--config` and `--rules` options is not allowed');	×
201	}
202
203	// There's no one single source of truth when it comes to fixing single file, we need to collect statuses from events.
204	$this->eventDispatcher->addListener(FileProcessed::NAME, function (FileProcessed $event): void {	1✔
205	$this->events[] = $event;	×
206	});	1✔
207
208	$this->configurationResolver = new ConfigurationResolver(	1✔
209	new Config(),	1✔
210	[	1✔
211	'allow-risky' => $input->getOption('allow-risky'),	1✔
212	'config' => $passedConfig,	1✔
213	'dry-run' => $input->getOption('dry-run'),	1✔
214	'rules' => $passedRules,	1✔
215	'path' => [],	1✔
216	'path-mode' => ConfigurationResolver::PATH_MODE_OVERRIDE, // IMPORTANT! WorkerCommand is called with file that already passed filtering, so here we can rely on PATH_MODE_OVERRIDE.	1✔
217	'using-cache' => $input->getOption('using-cache'),	1✔
218	'cache-file' => $input->getOption('cache-file'),	1✔
219	'diff' => $input->getOption('diff'),	1✔
220	'stop-on-violation' => $input->getOption('stop-on-violation'),	1✔
221	],	1✔
222	getcwd(), // @phpstan-ignore-line	1✔
223	$this->toolInfo	1✔
224	);	1✔
225
226	return new Runner(	1✔
227	null, // Paths are known when parallelisation server requests new chunk, not now	1✔
228	$this->configurationResolver->getFixers(),	1✔
229	$this->configurationResolver->getDiffer(),	1✔
230	$this->eventDispatcher,	1✔
231	$this->errorsManager,	1✔
232	$this->configurationResolver->getLinter(),	1✔
233	$this->configurationResolver->isDryRun(),	1✔
234	new NullCacheManager(), // IMPORTANT! We pass null cache, as cache is read&write in main process and we do not need to do it again.	1✔
235	$this->configurationResolver->getDirectory(),	1✔
236	$this->configurationResolver->shouldStopOnViolation(),	1✔
237	ParallelConfigFactory::sequential(), // IMPORTANT! Worker must run in sequential mode.	1✔
238	null,	1✔
239	$this->configurationResolver->getConfigFile()	1✔
240	);	1✔
241	}
242	}

keradus / PHP-CS-Fixer / 16253811376

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