Lines 94.87% 37 / 39
Functions and Methods 84.61% 11 / 13
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
BrokerRun 94.87% 37 / 39 84.61% 11 / 13 18.04 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 start 100.00% 3 / 3 100.00% 1 / 1 1
 isRunning 100.00% 1 / 1 100.00% 1 / 1 1
 isStopping 100.00% 1 / 1 100.00% 1 / 1 1
 awaitStopped 100.00% 1 / 1 100.00% 1 / 1 1
 stop 100.00% 1 / 1 100.00% 1 / 1 1
 stopToRestart 100.00% 1 / 1 100.00% 1 / 1 1
 stopWithError 100.00% 1 / 1 100.00% 1 / 1 1
 killWorkersNow 75.00% 3 / 4 0.00% 0 / 1 2.06
 stopRun 100.00% 10 / 10 100.00% 1 / 1 3
 runStopSteps 100.00% 7 / 7 100.00% 1 / 1 1
 moveTo 66.66% 2 / 3 0.00% 0 / 1 2.15
 attemptStopStep 100.00% 5 / 5 100.00% 1 / 1 2
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Amp\DeferredFuture;
8use Closure;
9use LogicException;
10use Psr\Log\LoggerInterface;
11use Stewart\Runtime\Broker\Mqtt\MqttLink;
12use Stewart\Runtime\Config\StewartConfig;
13use Stewart\Runtime\Lifecycle\BrokerRunPhase;
14use Stewart\Runtime\Lifecycle\BrokerStopOutcome;
15use Throwable;
16
17final class BrokerRun
18{
19    private BrokerRunPhase $phase = BrokerRunPhase::Idle;
20
21    /** @var DeferredFuture<BrokerStopOutcome> */
22    private DeferredFuture $runFinished;
23
24    public function __construct(
25        private readonly WorkerPool $pool,
26        private readonly WorkerWatchdog $watchdog,
27        private readonly HaSession $session,
28        private readonly MqttLink $mqtt,
29        private readonly LoggerInterface $logger,
30        private readonly StewartConfig $config,
31        private readonly DaemonStartTime $startTime,
32    ) {
33        $this->runFinished = new DeferredFuture();
34    }
35
36    public function start(): void
37    {
38        $this->moveTo(BrokerRunPhase::Running);
39        $this->runFinished = new DeferredFuture();
40        $this->startTime->recordStart();
41    }
42
43    public function isRunning(): bool
44    {
45        return $this->phase === BrokerRunPhase::Running;
46    }
47
48    public function isStopping(): bool
49    {
50        return $this->phase === BrokerRunPhase::Stopping;
51    }
52
53    public function awaitStopped(): BrokerStopOutcome
54    {
55        return $this->runFinished->getFuture()->await();
56    }
57
58    public function stop(string $reason): void
59    {
60        $this->stopRun($reason, BrokerStopOutcome::Stopped, null);
61    }
62
63    public function stopToRestart(string $reason): void
64    {
65        $this->stopRun($reason, BrokerStopOutcome::RestartRequested, null);
66    }
67
68    public function stopWithError(string $reason, Throwable $error): void
69    {
70        $this->stopRun($reason, BrokerStopOutcome::Stopped, $error);
71    }
72
73    public function killWorkersNow(): void
74    {
75        if (!$this->isStopping()) {
76            return;
77        }
78
79        $this->logger->warning('Killing workers without waiting for their grace');
80        $this->pool->terminateWorkers();
81    }
82
83    private function stopRun(string $reason, BrokerStopOutcome $outcome, ?Throwable $error): void
84    {
85        if (!$this->isRunning()) {
86            return;
87        }
88
89        $this->moveTo(BrokerRunPhase::Stopping);
90
91        $this->logger->info('Shutting down', ['reason' => $reason, 'grace' => (string) $this->config->shutdownGrace]);
92
93        $stepFailure = $this->runStopSteps($reason);
94        $failure = $error ?? $stepFailure;
95        $this->moveTo(BrokerRunPhase::Stopped);
96
97        if ($failure === null) {
98            $this->runFinished->complete($outcome);
99        } else {
100            $this->runFinished->error($failure);
101        }
102    }
103
104    private function runStopSteps(string $reason): ?Throwable
105    {
106        $workersFailure = $this->attemptStopStep('Could not stop the workers while shutting down', function () use ($reason): void {
107            $this->watchdog->stopProbing();
108            $this->pool->shutdown($reason, $this->config->shutdownGrace);
109        });
110        $sessionFailure = $this->attemptStopStep('Could not close the Home Assistant session while shutting down', $this->session->close(...));
111        $mqttFailure = $this->attemptStopStep('Could not close the MQTT connection while shutting down', $this->mqtt->close(...));
112
113        return $workersFailure ?? $sessionFailure ?? $mqttFailure;
114    }
115
116    private function moveTo(BrokerRunPhase $next): void
117    {
118        if (!$this->phase->canEnter($next)) {
119            throw new LogicException(\sprintf('The broker run cannot move from %s to %s.', $this->phase->name, $next->name));
120        }
121
122        $this->phase = $next;
123    }
124
125    /** @param Closure(): void $step */
126    private function attemptStopStep(string $failureMessage, Closure $step): ?Throwable
127    {
128        try {
129            $step();
130        } catch (Throwable $e) {
131            $this->logger->error($failureMessage, ['exception' => $e]);
132
133            return $e;
134        }
135
136        return null;
137    }
138}