Lines 94.11% 16 / 17
Functions and Methods 66.66% 2 / 3
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
WorkerStartup 94.11% 16 / 17 66.66% 2 / 3 7.01 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 startWorkers 100.00% 11 / 11 100.00% 1 / 1 4
 spawnEveryWorker 80.00% 4 / 5 0.00% 0 / 1 2.03
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Psr\Log\LoggerInterface;
8use Stewart\Runtime\Broker\Collection\WorkerSlotCollection;
9use Stewart\Runtime\Broker\Exposure\ExposureReconciler;
10use Stewart\Runtime\Exception\BrokerError;
11use Stewart\Runtime\Exception\BrokerException;
12use Throwable;
13
14use function Amp\async;
15use function Amp\Future\awaitAll;
16
17final readonly class WorkerStartup
18{
19    public function __construct(
20        private WorkerSlotCollection $workerSlots,
21        private WorkerPool $pool,
22        private WorkerSlotRegistry $slots,
23        private WorkerWatchdog $watchdog,
24        private BrokerWorkerEvents $workerEvents,
25        private ExposureReconciler $reconciler,
26        private HaSession $session,
27        private BrokerRun $run,
28        private LoggerInterface $logger,
29    ) {}
30
31    /** @throws Throwable */
32    public function startWorkers(): void
33    {
34        $this->spawnEveryWorker();
35
36        // The run can end while a spawn suspends; run() then reports that reason instead.
37        if (!$this->run->isRunning()) {
38            return;
39        }
40
41        if (!$this->workerSlots->isEmpty() && $this->slots->countLiveWorkers() === 0) {
42            throw BrokerException::createForReason(BrokerError::NoWorkersStarted, workers: $this->workerSlots->count());
43        }
44
45        $this->watchdog->startProbing();
46        $this->reconciler->scheduleOnceWorkersSettle();
47
48        $this->logger->info('Stewart is running', [
49            'workers' => $this->slots->countLiveWorkers(),
50            'entities' => $this->session->countEntities(),
51        ]);
52    }
53
54    /** @throws Throwable */
55    private function spawnEveryWorker(): void
56    {
57        [$failures] = awaitAll($this->workerSlots->mapToList(
58            fn(WorkerSlot $slot) => async(fn(): bool => $this->pool->startWorker($slot, $this->workerEvents)),
59        ));
60
61        foreach ($failures as $failure) {
62            throw $failure;
63        }
64    }
65}