Lines 91.66% 11 / 12
Functions and Methods 80.00% 4 / 5
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
BrokerWorkerEvents 91.66% 11 / 12 80.00% 4 / 5 6.02 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 workerSpawned 100.00% 6 / 6 100.00% 1 / 1 2
 workerMessage 100.00% 1 / 1 100.00% 1 / 1 1
 workerGone 100.00% 3 / 3 100.00% 1 / 1 1
 workerQuarantined 0.00% 0 / 1 0.00% 0 / 1 2
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Stewart\Runtime\Broker\Exposure\ExposureCommandRouter;
8use Stewart\Runtime\Broker\Exposure\ExposureLink;
9use Stewart\Runtime\Broker\Exposure\ExposureReconciler;
10use Stewart\Runtime\Broker\Message\WorkerMessageDispatcher;
11use Stewart\Runtime\Ipc\Message\StateSnapshot;
12use Stewart\Runtime\Ipc\Message\WorkerMessage;
13
14final readonly class BrokerWorkerEvents implements WorkerPoolListener
15{
16    public function __construct(
17        private HaSession $session,
18        private ExposureLink $exposures,
19        private BootstrapMessageFactory $bootstrapMessageFactory,
20        private ConnectionTracker $connection,
21        private WorkerMessageDispatcher $messages,
22        private SubscriptionRegistry $registry,
23        private ExposureReconciler $reconciler,
24        private ExposureCommandRouter $commandRouter,
25    ) {}
26
27    public function workerSpawned(WorkerHandle $handle): void
28    {
29        $snapshot = $this->session->snapshotStateCache();
30
31        $handle->send($this->bootstrapMessageFactory->createBootstrapMessage($handle, $this->session));
32        $handle->send($this->session->snapshotRegistry()->toRegistrySnapshot());
33        $handle->send(new StateSnapshot($snapshot->states, $snapshot->revision));
34
35        if ($this->connection->lastLoss !== null) {
36            $handle->send($this->connection->lastLoss);
37        }
38    }
39
40    public function workerMessage(WorkerHandle $handle, WorkerMessage $message): void
41    {
42        $this->messages->dispatch($handle, $message);
43    }
44
45    public function workerGone(WorkerHandle $handle, string $reason): void
46    {
47        $this->registry->removeWorkerSubscriptions($handle->id);
48        $this->exposures->orphanExposuresOfWorker($handle->id);
49        $this->commandRouter->failCommandsOf($handle->id);
50    }
51
52    public function workerQuarantined(WorkerSlot $slot): void
53    {
54        $this->reconciler->recordWorkerQuarantined($slot->workerId);
55    }
56}