Master::run() - Code Metrics - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Master::run() F
last analyzed 2025-06-04 13:17 UTC

↳ Parent: Master

Complexity

Conditions	26
Paths	161

Size

Total Lines	84
Code Lines	50

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	34
CRAP Score	32.198

Importance

Changes

Metric	Value
cc	26
eloc	50
nc	161
nop	3
dl	0
loc	84
rs	3.6583
c	0
b	0
f	0
ccs	34
cts	43
cp	0.7907
crap	32.198

How to fix Long Method Complexity

<?php

declare(strict_types=1);

/* (c) Anton Medvedev <[email protected]>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Deployer\Executor;

use Deployer\Deployer;
use Deployer\Host\Host;
use Deployer\Host\HostCollection;
use Deployer\Selector\Selector;
use Deployer\Ssh\IOArguments;
use Deployer\Task\Context;
use Deployer\Task\Task;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Process\PhpExecutableFinder;
use Symfony\Component\Process\Process;

const FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'];

function spinner(string $message = ''): string
{
    $frame = FRAMES[(int) ((int) (new \DateTime())->format('u') / 1e5) % count(FRAMES)];
    return "  $frame $message\r";
}

class Master
{
    private HostCollection $hosts;
    private InputInterface $input;
    private OutputInterface $output;
    private Messenger $messenger;
    private string|false $phpBin;

    public function __construct(
        HostCollection  $hosts,
        InputInterface  $input,
        OutputInterface $output,
        Messenger       $messenger,
    ) {
        $this->hosts = $hosts;
        $this->input = $input;
        $this->output = $output;
        $this->messenger = $messenger;
        $this->phpBin = (new PhpExecutableFinder())->find();
    }

    /**
     * @param Task[] $tasks
     * @param Host[] $hosts
     */
    public function run(array $tasks, array $hosts, ?Planner $plan = null): int
    {
        $globalLimit = (int) $this->input->getOption('limit') ?: count($hosts);

        foreach ($tasks as $task) {
            if (!$plan) {
                $this->messenger->startTask($task);
            }

            $plannedHosts = $hosts;

            $limit = min($globalLimit, $task->getLimit() ?? $globalLimit);

            if ($task->isOnce()) {
                $plannedHosts = [];
                foreach ($hosts as $currentHost) {
                    if (Selector::apply($task->getSelector(), $currentHost)) {
                        $plannedHosts[] = $currentHost;
                        break;
                    }
                }
            } elseif ($task->isOncePerNode()) {
                $plannedHosts = [];
                foreach ($hosts as $currentHost) {
                    if (Selector::apply($task->getSelector(), $currentHost)) {
                        $nodeLabel = $currentHost->getHostname();
                        $labels = $currentHost->config()->get('labels', []);
                        if (is_array($labels) && array_key_exists('node', $labels)) {
                            $nodeLabel = $labels['node'];
                        }
                        if (array_key_exists($nodeLabel, $plannedHosts)) {
                            continue;
                        }
                        $plannedHosts[$nodeLabel] = $currentHost;
                    }
                }
            }

            if ($limit === 1 || count($plannedHosts) === 1) {
                foreach ($plannedHosts as $currentHost) {
                    if (!Selector::apply($task->getSelector(), $currentHost)) {
                        if ($plan) {
                            $plan->commit([], $task);
                        }
                        continue;
                    }

                    if ($plan) {
                        $plan->commit([$currentHost], $task);
                        continue;
                    }

                    $exitCode = $this->runTask($task, [$currentHost]);
                    if ($exitCode !== 0) {
                        return $exitCode;
                    }
                }
            } else {
                foreach (array_chunk($plannedHosts, $limit) as $chunk) {
                    $selectedHosts = [];
                    foreach ($chunk as $currentHost) {
                        if (Selector::apply($task->getSelector(), $currentHost)) {
                            $selectedHosts[] = $currentHost;
                        }
                    }

                    if ($plan) {
                        $plan->commit($selectedHosts, $task);
                        continue;
                    }

                    $exitCode = $this->runTask($task, $selectedHosts);
                    if ($exitCode !== 0) {
                        return $exitCode;
                    }
                }
            }

            if (!$plan) {
                $this->messenger->endTask($task);
            }
        }

        return 0;
    }

    /**
     * @param Host[] $hosts
     */
    private function runTask(Task $task, array $hosts): int
    {
        if (getenv('DEPLOYER_LOCAL_WORKER') === 'true') {
            // This allows to code coverage all recipe,
            // as well as speedup tests by not spawning
            // lots of processes. Also there is a few tests
            // what runs with workers for tests subprocess
            // communications.
            foreach ($hosts as $host) {
                $worker = new Worker(Deployer::get());
                $exitCode = $worker->execute($task, $host);
                if ($exitCode !== 0) {
                    $this->messenger->endTask($task, true);
                    return $exitCode;
                }
            }
            return 0;
        }

        $server = new Server('127.0.0.1', 0, $this->output);
filter:
    dependency_paths: ["lib/*"]

        /** @var Process[] $processes */
        $processes = [];

        $server->afterRun(function (int $port) use (&$processes, $hosts, $task) {
            foreach ($hosts as $host) {
                $processes[] = $this->createProcess($host, $task, $port);
            }

            foreach ($processes as $process) {
                $process->start();
            }
        });

        $echoCallback = function (string $output) {
            $output = preg_replace('/\n$/', '', $output);
            if (strlen($output) !== 0) {
                $this->output->writeln($output);
            }
        };

        $server->ticker(function () use (&$processes, $server, $echoCallback) {
            $this->gatherOutput($processes, $echoCallback);
            if ($this->output->isDecorated() && !getenv('CI')) {
                $this->output->write(spinner());
            }
            if ($this->allFinished($processes)) {
                $server->stop();
            }
        });

        $server->router(function (string $path, array $payload) {
            switch ($path) {
                case '/load':
                    ['host' => $host] = $payload;

                    $host = $this->hosts->get($host);
                    $config = $host->config()->persist();

                    return new Response(200, $config);

                case '/save':
                    ['host' => $host, 'config' => $config] = $payload;

                    $host = $this->hosts->get($host);
                    $host->config()->update($config);

                    return new Response(200, true);

                case '/proxy':
                    ['host' => $host, 'func' => $func, 'arguments' => $arguments] = $payload;

                    Context::push(new Context($this->hosts->get($host)));
                    $answer = call_user_func($func, ...$arguments);
                    Context::pop();

                    return new Response(200, $answer);

                default:
                    return new Response(404, null);
            }
        });

        $server->run();

        if ($this->output->isDecorated() && !getenv('CI')) {
            $this->output->write("    \r"); // clear spinner
        }
        $this->gatherOutput($processes, $echoCallback);

        if ($this->cumulativeExitCode($processes) !== 0) {
            $this->messenger->endTask($task, true);
        }

        return $this->cumulativeExitCode($processes);
    }

    protected function createProcess(Host $host, Task $task, int $port): Process
    {
        $command = [
            $this->phpBin, DEPLOYER_BIN,
            'worker', '--port', $port,
            '--task', $task,
            '--host', $host->getAlias(),
        ];
        $command = array_merge($command, IOArguments::collect($this->input, $this->output));
        if ($task->isVerbose() && $this->output->getVerbosity() === OutputInterface::VERBOSITY_NORMAL) {
            $command[] = '-v';
        }
        if ($this->output->isDebug()) {
            $this->output->writeln("[$host] " . join(' ', $command));
        }
        return new Process($command);
    }

    /**
     * @param Process[] $processes
     */
    protected function allFinished(array $processes): bool
    {
        foreach ($processes as $process) {
            if (!$process->isTerminated()) {
                return false;
            }
        }
        return true;
    }

    /**
     * @param Process[] $processes
     */
    protected function gatherOutput(array $processes, callable $callback): void
    {
        foreach ($processes as $process) {
            $output = $process->getIncrementalOutput();
            if (strlen($output) !== 0) {
                $callback($output);
            }

            $errorOutput = $process->getIncrementalErrorOutput();
            if (strlen($errorOutput) !== 0) {
                $callback($errorOutput);
            }
        }
    }

    /**
     * @param Process[] $processes
     */
    protected function cumulativeExitCode(array $processes): int
    {
        foreach ($processes as $process) {
            if ($process->getExitCode() > 0) {
                return $process->getExitCode();

            }
        }
        return 0;
    }
}


1		<?php
2
3		declare(strict_types=1);
4
5		/* (c) Anton Medvedev <[email protected]>
6		*
7		* For the full copyright and license information, please view the LICENSE
8		* file that was distributed with this source code.
9		*/
10
11		namespace Deployer\Executor;
12
13		use Deployer\Deployer;
14		use Deployer\Host\Host;
15		use Deployer\Host\HostCollection;
16		use Deployer\Selector\Selector;
17		use Deployer\Ssh\IOArguments;
18		use Deployer\Task\Context;
19		use Deployer\Task\Task;
20		use Symfony\Component\Console\Input\InputInterface;
21		use Symfony\Component\Console\Output\OutputInterface;
22	1	use Symfony\Component\Process\PhpExecutableFinder;
23		use Symfony\Component\Process\Process;
24
25		const FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'];
26	4
27	4	function spinner(string $message = ''): string
28		{
29		$frame = FRAMES[(int) ((int) (new \DateTime())->format('u') / 1e5) % count(FRAMES)];
30		return " $frame $message\r";
31		}
32
33		class Master
34		{
35		private HostCollection $hosts;
36		private InputInterface $input;
37		private OutputInterface $output;
38		private Messenger $messenger;
39	12	private string\|false $phpBin;
40
41		public function __construct(
42		HostCollection $hosts,
43		InputInterface $input,
44		OutputInterface $output,
45		Messenger $messenger,
46		) {
47		$this->hosts = $hosts;
48	12	$this->input = $input;
49	12	$this->output = $output;
50	12	$this->messenger = $messenger;
51	12	$this->phpBin = (new PhpExecutableFinder())->find();
52	12	}
53	12
54	12	/**
55		* @param Task[] $tasks
56		* @param Host[] $hosts
57		*/
58		public function run(array $tasks, array $hosts, ?Planner $plan = null): int
59		{
60		$globalLimit = (int) $this->input->getOption('limit') ?: count($hosts);
61
62	12	foreach ($tasks as $task) {
63		if (!$plan) {
64	12	$this->messenger->startTask($task);
65	12	}
66
67	12	$plannedHosts = $hosts;
68
69	12	$limit = min($globalLimit, $task->getLimit() ?? $globalLimit);
70	12
71		if ($task->isOnce()) {
72	12	$plannedHosts = [];
73		foreach ($hosts as $currentHost) {
74	12	if (Selector::apply($task->getSelector(), $currentHost)) {
75		$plannedHosts[] = $currentHost;
76	12	break;
77	3	}
78	3	}
79	3	} elseif ($task->isOncePerNode()) {
80	3	$plannedHosts = [];
81	3	foreach ($hosts as $currentHost) {
82		if (Selector::apply($task->getSelector(), $currentHost)) {
83		$nodeLabel = $currentHost->getHostname();
84		$labels = $currentHost->config()->get('labels', []);
85		if (is_array($labels) && array_key_exists('node', $labels)) {
86	12	$nodeLabel = $labels['node'];
87		}
88		if (array_key_exists($nodeLabel, $plannedHosts)) {
89		continue;
90	12	}
91	9	$plannedHosts[$nodeLabel] = $currentHost;
92	9	}
93		}
94		}
95
96		if ($limit === 1 \|\| count($plannedHosts) === 1) {
97		foreach ($plannedHosts as $currentHost) {
98		if (!Selector::apply($task->getSelector(), $currentHost)) {
99	9	if ($plan) {
100		$plan->commit([], $task);
101		}
102		continue;
103		}
104	9
105	9	if ($plan) {
106	2	$plan->commit([$currentHost], $task);
107		continue;
108		}
109
110	4	$exitCode = $this->runTask($task, [$currentHost]);
111	4	if ($exitCode !== 0) {
112	4	return $exitCode;
113	4	}
114	4	}
115	4	} else {
116		foreach (array_chunk($plannedHosts, $limit) as $chunk) {
117		$selectedHosts = [];
118		foreach ($chunk as $currentHost) {
119	4	if (Selector::apply($task->getSelector(), $currentHost)) {
120		$selectedHosts[] = $currentHost;
121		}
122		}
123
124	4	if ($plan) {
125	4	$plan->commit($selectedHosts, $task);
126		continue;
127		}
128
129		$exitCode = $this->runTask($task, $selectedHosts);
130		if ($exitCode !== 0) {
131	12	return $exitCode;
132	12	}
133		}
134		}
135
136	12	if (!$plan) {
137		$this->messenger->endTask($task);
138		}
139		}
140
141		return 0;
142	12	}
143
144		/**
145		* @param Host[] $hosts
146		*/
147		private function runTask(Task $task, array $hosts): int
148		{
149	12	if (getenv('DEPLOYER_LOCAL_WORKER') === 'true') {
150		// This allows to code coverage all recipe,
151		// as well as speedup tests by not spawning
152	12	// lots of processes. Also there is a few tests
153	12	// what runs with workers for tests subprocess
154	12	// communications.
155		foreach ($hosts as $host) {
156		$worker = new Worker(Deployer::get());
157		$exitCode = $worker->execute($task, $host);
158		if ($exitCode !== 0) {
159		$this->messenger->endTask($task, true);
160		return $exitCode;
161		}
162		}
163		return 0;
164		}
165
166		$server = new Server('127.0.0.1', 0, $this->output);
		0 ignored issues – show Bug introduced 2024-10-22 20:22 UTC by Report Bug Copy Issue Report The type `Deployer\Executor\Server` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
167
168		/** @var Process[] $processes */
169	12	$processes = [];
170	12
171		$server->afterRun(function (int $port) use (&$processes, $hosts, $task) {
172		foreach ($hosts as $host) {
173		$processes[] = $this->createProcess($host, $task, $port);
174		}
175
176		foreach ($processes as $process) {
177	12	$process->start();
178		}
179	12	});
180
181		$echoCallback = function (string $output) {
182		$output = preg_replace('/\n$/', '', $output);
183		if (strlen($output) !== 0) {
184		$this->output->writeln($output);
185	8	}
186	8	};
187	8
188	8	$server->ticker(function () use (&$processes, $server, $echoCallback) {
189	2	$this->gatherOutput($processes, $echoCallback);
190		if ($this->output->isDecorated() && !getenv('CI')) {
191		$this->output->write(spinner());
192	8	}
193		if ($this->allFinished($processes)) {
194		$server->stop();
195		}
196	3	});
197	3
198	3	$server->router(function (string $path, array $payload) {
199		switch ($path) {
200	4	case '/load':
201		['host' => $host] = $payload;
202
203	4	$host = $this->hosts->get($host);
204		$config = $host->config()->persist();
205
206	4	return new Response(200, $config);
207	4
208		case '/save':
209		['host' => $host, 'config' => $config] = $payload;
210	4
211	4	$host = $this->hosts->get($host);
212		$host->config()->update($config);
213	4
214		return new Response(200, true);
215
216	4	case '/proxy':
217	4	['host' => $host, 'func' => $func, 'arguments' => $arguments] = $payload;
218	4
219	4	Context::push(new Context($this->hosts->get($host)));
220	4	$answer = call_user_func($func, ...$arguments);
221		Context::pop();
222	4
223		return new Response(200, $answer);
224	4
225		default:
226	4	return new Response(404, null);
227	4	}
228		});
229	4
230		$server->run();
231
232	4	if ($this->output->isDecorated() && !getenv('CI')) {
233		$this->output->write(" \r"); // clear spinner
234	4	}
235	4	$this->gatherOutput($processes, $echoCallback);
236	4
237	4	if ($this->cumulativeExitCode($processes) !== 0) {
238		$this->messenger->endTask($task, true);
239	4	}
240	1
241		return $this->cumulativeExitCode($processes);
242		}
243	4
244		protected function createProcess(Host $host, Task $task, int $port): Process
245		{
246		$command = [
247		$this->phpBin, DEPLOYER_BIN,
248		'worker', '--port', $port,
249		'--task', $task,
250		'--host', $host->getAlias(),
251		];
252		$command = array_merge($command, IOArguments::collect($this->input, $this->output));
253		if ($task->isVerbose() && $this->output->getVerbosity() === OutputInterface::VERBOSITY_NORMAL) {
254		$command[] = '-v';
255		}
256		if ($this->output->isDebug()) {
257		$this->output->writeln("[$host] " . join(' ', $command));
258		}
259		return new Process($command);
260		}
261
262		/**
263		* @param Process[] $processes
264	4	*/
265		protected function allFinished(array $processes): bool
266	4	{
267	4	foreach ($processes as $process) {
268	4	if (!$process->isTerminated()) {
269		return false;
270		}
271	4	}
272		return true;
273		}
274
275		/**
276		* @param Process[] $processes
277		*/
278	4	protected function gatherOutput(array $processes, callable $callback): void
279		{
280	4	foreach ($processes as $process) {
281	4	$output = $process->getIncrementalOutput();
282	4	if (strlen($output) !== 0) {
283	3	$callback($output);
284		}
285
286	4	$errorOutput = $process->getIncrementalErrorOutput();
287	4	if (strlen($errorOutput) !== 0) {
288		$callback($errorOutput);
289		}
290		}
291	4	}
292
293		/**
294		* @param Process[] $processes
295		*/
296		protected function cumulativeExitCode(array $processes): int
297	4	{
298		foreach ($processes as $process) {
299	4	if ($process->getExitCode() > 0) {
300	4	return $process->getExitCode();
		0 ignored issues – show Bug Best Practice introduced 2020-09-29 21:09 UTC by Report Bug Copy Issue Report The expression `return $process->getExitCode()` could return the type `null` which is incompatible with the type-hinted return `integer`. Consider adding an additional type-check to rule them out. Loading history...
301		}
302		}
303		return 0;
304	4	}
305		}
306

deployphp / deployer

GitHub Access Token became invalid

Master::run() F last analyzed 2025-06-04 13:17 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Duplication Side-by-Side

Filter issues like

Master::run() F
last analyzed 2025-06-04 13:17 UTC