Master::allFinished() - Code Metrics - Inspection of "v7" - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( e2875b...35ced9 )

by Anton

created 2020-09-29 20:56 UTC

Master::allFinished() A

↳ Parent: Master

Complexity

Conditions	3
Paths	3

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	3

Importance

Changes

Metric	Value
cc	3
nc	3
nop	1
dl	0
loc	9
ccs	5
cts	5
cp	1
crap	3
rs	9.9666
c	0
b	0
f	0

<?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\Component\Ssh\Client;
use Deployer\Configuration\Configuration;
use Deployer\Deployer;
use Deployer\Exception\Exception;
use Deployer\Host\Host;
use Deployer\Host\Localhost;
use Deployer\Selector\Selector;
use Deployer\Task\Task;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Process\Process;

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

function spinner($message = '')
{
    $frame = FRAMES[(int)(microtime(true) * 10) % count(FRAMES)];
    return "  $frame $message\r";
}

class Master
{
    private $input;
    private $output;
    private $server;
    private $messenger;
    private $client;
    private $config;

    public function __construct(
        InputInterface $input,
        OutputInterface $output,
        Server $server,
        Messenger $messenger,
        Client $client,
        Configuration $config
    )
    {
        $this->input = $input;
        $this->output = $output;
        $this->server = $server;
        $this->messenger = $messenger;
        $this->client = $client;
        $this->config = $config;
    }

    /**
     * @param Task[] $tasks
     * @param Host[] $hosts
     * @param Planner|null $plan
     * @return int
     */
    public function run(array $tasks, array $hosts, $plan = null): int
    {
        $plan || $this->server->start();
        $plan || $this->connect($hosts);

        $globalLimit = (int)$this->input->getOption('limit') ?: count($hosts);

        foreach ($tasks as $task) {
            $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;
                    }
                }
            }

            if ($task->isLocal()) {
                $plannedHosts = [new Localhost('localhost')];
            }

            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($hosts, $limit) as $chunk) {
                    $selector = $task->getSelector();
                    $selectedHosts = [];
                    foreach ($chunk as $currentHost) {
                        if ($selector === null || Selector::apply($selector, $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 connect(array $hosts)
    {
        $callback = function (string $output) {
            $output = preg_replace('/\n$/', '', $output);
            if (strlen($output) !== 0) {
                $this->output->writeln($output);
            }
        };

        // Connect to each host sequentially, to prevent getting locked.
        foreach ($hosts as $host) {
            if ($host instanceof Localhost) {
                continue;
            }
            $process = $this->createConnectProcess($host);
            $process->start();

            while ($process->isRunning()) {
                $this->gatherOutput([$process], $callback);
                if ($this->output->isDecorated()) {
                    $this->output->write(spinner(str_pad("connect {$host->getTag()}", intval(getenv('COLUMNS')) - 1)));
                }
                usleep(1000);
            }
        }

        // Clear spinner.
        $this->output->write(str_repeat(' ', intval(getenv('COLUMNS')) - 1) . "\r");
    }

    /**
     * @param Task $task
     * @param Host[] $hosts
     * @return int
     */
    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) {
                    return $exitCode;
                }
            }
            return 0;
        }

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

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

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

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

        $this->server->addPeriodicTimer(0.03, function ($timer) use (&$processes, $callback) {
            $this->gatherOutput($processes, $callback);
            $this->output->write(spinner());
            if ($this->allFinished($processes)) {
                $this->server->stop();
                $this->server->cancelTimer($timer);
            }
        });

        $this->server->run();

        $this->output->write("    \r"); // clear spinner
        $this->gatherOutput($processes, $callback);

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

    protected function createProcess(Host $host, Task $task): Process
    {
        $dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
        $decorated = $this->output->isDecorated() ? '--decorated' : '';
        $verbosity = self::stringifyVerbosity($this->output->getVerbosity());
        $command = "$dep worker $task {$host->getAlias()} {$this->server->getPort()} {$this->input} $decorated $verbosity";

        if ($this->output->isDebug()) {
            $this->output->writeln("[{$host->getTag()}] $command");
        }

        return Process::fromShellCommandline($command);
    }

    protected function createConnectProcess(Host $host): Process
    {
        $dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
        $decorated = $this->output->isDecorated() ? '--decorated' : '';
        $verbosity = self::stringifyVerbosity($this->output->getVerbosity());
        $command = "$dep connect {$host->getAlias()} $decorated $verbosity";

        if ($this->output->isDebug()) {
            $this->output->writeln("[{$host->getTag()}] $command");
        }

        return Process::fromShellCommandline($command);
    }

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

    /**
     * @param Process[] $processes
     * @param callable $callback
     */
    protected function gatherOutput(array $processes, callable $callback)
    {
        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
     * @return int
     */
    protected function cumulativeExitCode(array $processes): int
    {
        foreach ($processes as $process) {
            if ($process->getExitCode() > 0) {
                return $process->getExitCode();
            }
        }
        return 0;
    }

    private static function stringifyVerbosity(int $verbosity): string
    {
        switch ($verbosity) {
            case OutputInterface::VERBOSITY_QUIET:
                return '-q';
            case OutputInterface::VERBOSITY_NORMAL:
                return '';
            case OutputInterface::VERBOSITY_VERBOSE:
                return '-v';
            case OutputInterface::VERBOSITY_VERY_VERBOSE:
                return '-vv';
            case OutputInterface::VERBOSITY_DEBUG:
                return '-vvv';
            default:
                throw new Exception('Unknown verbosity level: ' . $verbosity);
        }
    }
}


1			<?php declare(strict_types=1);
2			/* (c) Anton Medvedev <[email protected]>
3			*
4			* For the full copyright and license information, please view the LICENSE
5			* file that was distributed with this source code.
6			*/
7
8			namespace Deployer\Executor;
9
10			use Deployer\Component\Ssh\Client;
11			use Deployer\Configuration\Configuration;
12			use Deployer\Deployer;
13			use Deployer\Exception\Exception;
14			use Deployer\Host\Host;
15			use Deployer\Host\Localhost;
16			use Deployer\Selector\Selector;
17			use Deployer\Task\Task;
18			use Symfony\Component\Console\Input\InputInterface;
19			use Symfony\Component\Console\Output\OutputInterface;
20			use Symfony\Component\Process\Process;
21
22	1		const FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'];
23
24			function spinner($message = '')
25			{
26	4		$frame = FRAMES[(int)(microtime(true) * 10) % count(FRAMES)];
27	4		return " $frame $message\r";
28			}
29
30			class Master
31			{
32			private $input;
33			private $output;
34			private $server;
35			private $messenger;
36			private $client;
37			private $config;
38
39	12		public function __construct(
40			InputInterface $input,
41			OutputInterface $output,
42			Server $server,
43			Messenger $messenger,
44			Client $client,
45			Configuration $config
46			)
47			{
48	12		$this->input = $input;
49	12		$this->output = $output;
50	12		$this->server = $server;
51	12		$this->messenger = $messenger;
52	12		$this->client = $client;
53	12		$this->config = $config;
54	12		}
55
56			/**
57			* @param Task[] $tasks
58			* @param Host[] $hosts
59			* @param Planner\|null $plan
60			* @return int
61			*/
62	12		public function run(array $tasks, array $hosts, $plan = null): int
63			{
64	12		$plan \|\| $this->server->start();
65	12		$plan \|\| $this->connect($hosts);
66
67	12		$globalLimit = (int)$this->input->getOption('limit') ?: count($hosts);
68
69	12		foreach ($tasks as $task) {
70	12		$plan \|\| $this->messenger->startTask($task);
71
72	12		$plannedHosts = $hosts;
73
74	12		$limit = min($globalLimit, $task->getLimit() ?? $globalLimit);
75
76	12		if ($task->isOnce()) {
77	3		$plannedHosts = [];
78	3		foreach ($hosts as $currentHost) {
79	3		if (Selector::apply($task->getSelector(), $currentHost)) {
80	3		$plannedHosts[] = $currentHost;
81	3		break;
82			}
83			}
84			}
85
86	12		if ($task->isLocal()) {
87			$plannedHosts = [new Localhost('localhost')];
88			}
89
90	12		if ($limit === 1 \|\| count($plannedHosts) === 1) {
91	9		foreach ($plannedHosts as $currentHost) {
92	9		if (!Selector::apply($task->getSelector(), $currentHost)) {
93			if ($plan) {
94			$plan->commit([], $task);
95			}
96			continue;
97			}
98
99	9		if ($plan) {
100			$plan->commit([$currentHost], $task);
101			continue;
102			}
103
104	9		$exitCode = $this->runTask($task, [$currentHost]);
105	9		if ($exitCode !== 0) {
106	2		return $exitCode;
107			}
108			}
109			} else {
110	4		foreach (array_chunk($hosts, $limit) as $chunk) {
111	4		$selector = $task->getSelector();
112	4		$selectedHosts = [];
113	4		foreach ($chunk as $currentHost) {
114	4		if ($selector === null \|\| Selector::apply($selector, $currentHost)) {
115	4		$selectedHosts[] = $currentHost;
116			}
117			}
118
119	4		if ($plan) {
120			$plan->commit($selectedHosts, $task);
121			continue;
122			}
123
124	4		$exitCode = $this->runTask($task, $selectedHosts);
125	4		if ($exitCode !== 0) {
126			return $exitCode;
127			}
128			}
129			}
130
131	12		if (!$plan) {
132	12		$this->messenger->endTask($task);
133			}
134			}
135
136	12		return 0;
137			}
138
139			/**
140			* @param Host[] $hosts
141			*/
142	12		private function connect(array $hosts)
143			{
144		View Code Duplication	$callback = function (string $output) {
145			$output = preg_replace('/\n$/', '', $output);
146			if (strlen($output) !== 0) {
147			$this->output->writeln($output);
148			}
149	12		};
150
151			// Connect to each host sequentially, to prevent getting locked.
152	12		foreach ($hosts as $host) {
153	12		if ($host instanceof Localhost) {
154	12		continue;
155			}
156			$process = $this->createConnectProcess($host);
157			$process->start();
158
159			while ($process->isRunning()) {
160			$this->gatherOutput([$process], $callback);
161			if ($this->output->isDecorated()) {
162			$this->output->write(spinner(str_pad("connect {$host->getTag()}", intval(getenv('COLUMNS')) - 1)));
163			}
164			usleep(1000);
165			}
166			}
167
168			// Clear spinner.
169	12		$this->output->write(str_repeat(' ', intval(getenv('COLUMNS')) - 1) . "\r");
170	12		}
171
172			/**
173			* @param Task $task
174			* @param Host[] $hosts
175			* @return int
176			*/
177	12		private function runTask(Task $task, array $hosts): int
178			{
179	12		if (getenv('DEPLOYER_LOCAL_WORKER') === 'true') {
180			// This allows to code coverage all recipe,
181			// as well as speedup tests by not spawning
182			// lots of processes. Also there is a few tests
183			// what runs with workers for tests subprocess
184			// communications.
185	8		foreach ($hosts as $host) {
186	8		$worker = new Worker(Deployer::get());
187	8		$exitCode = $worker->execute($task, $host);
188	8		if ($exitCode !== 0) {
189	2		return $exitCode;
190			}
191			}
192	8		return 0;
193			}
194
195		View Code Duplication	$callback = function (string $output) {
196	3		$output = preg_replace('/\n$/', '', $output);
197	3		if (strlen($output) !== 0) {
198	3		$this->output->writeln($output);
199			}
200	4		};
201
202			/** @var Process[] $processes */
203	4		$processes = [];
204
205			$this->server->addTimer(0, function () use(&$processes, $hosts, $task) {
206	4		foreach ($hosts as $host) {
207	4		$processes[] = $this->createProcess($host, $task);
208			}
209
210	4		foreach ($processes as $process) {
211	4		$process->start();
212			}
213	4		});
214
215			$this->server->addPeriodicTimer(0.03, function ($timer) use (&$processes, $callback) {
216	4		$this->gatherOutput($processes, $callback);
217	4		$this->output->write(spinner());
218	4		if ($this->allFinished($processes)) {
219	4		$this->server->stop();
220	4		$this->server->cancelTimer($timer);
221			}
222	4		});
223
224	4		$this->server->run();
225
226	4		$this->output->write(" \r"); // clear spinner
227	4		$this->gatherOutput($processes, $callback);
228
229	4		return $this->cumulativeExitCode($processes);
230			}
231
232	4		protected function createProcess(Host $host, Task $task): Process
233			{
234	4		$dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
235	4		$decorated = $this->output->isDecorated() ? '--decorated' : '';
236	4		$verbosity = self::stringifyVerbosity($this->output->getVerbosity());
237	4		$command = "$dep worker $task {$host->getAlias()} {$this->server->getPort()} {$this->input} $decorated $verbosity";
238
239	4		if ($this->output->isDebug()) {
240	1		$this->output->writeln("[{$host->getTag()}] $command");
241			}
242
243	4		return Process::fromShellCommandline($command);
244			}
245
246			protected function createConnectProcess(Host $host): Process
247			{
248			$dep = PHP_BINARY . ' ' . DEPLOYER_BIN;
249			$decorated = $this->output->isDecorated() ? '--decorated' : '';
250			$verbosity = self::stringifyVerbosity($this->output->getVerbosity());
251			$command = "$dep connect {$host->getAlias()} $decorated $verbosity";
252
253			if ($this->output->isDebug()) {
254			$this->output->writeln("[{$host->getTag()}] $command");
255			}
256
257			return Process::fromShellCommandline($command);
258			}
259
260			/**
261			* @param Process[] $processes
262			* @return bool
263			*/
264	4		protected function allFinished(array $processes): bool
265			{
266	4		foreach ($processes as $process) {
267	4		if (!$process->isTerminated()) {
268	4		return false;
269			}
270			}
271	4		return true;
272			}
273
274			/**
275			* @param Process[] $processes
276			* @param callable $callback
277			*/
278	4		protected function gatherOutput(array $processes, callable $callback)
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			* @return int
296			*/
297	4		protected function cumulativeExitCode(array $processes): int
298			{
299	4		foreach ($processes as $process) {
300	4		if ($process->getExitCode() > 0) {
301			return $process->getExitCode();
302			}
303			}
304	4		return 0;
305			}
306
307	4		private static function stringifyVerbosity(int $verbosity): string
308			{
309	4		switch ($verbosity) {
310			case OutputInterface::VERBOSITY_QUIET:
311			return '-q';
312			case OutputInterface::VERBOSITY_NORMAL:
313	3		return '';
314			case OutputInterface::VERBOSITY_VERBOSE:
315			return '-v';
316			case OutputInterface::VERBOSITY_VERY_VERBOSE:
317			return '-vv';
318			case OutputInterface::VERBOSITY_DEBUG:
319	1		return '-vvv';
320			default:
321			throw new Exception('Unknown verbosity level: ' . $verbosity);
322			}
323			}
324			}
325

deployphp / deployer

GitHub Access Token became invalid

Push — master ( e2875b...35ced9 )

Master::allFinished() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like