Lines 50.98% 26 / 51
Functions and Methods 71.42% 5 / 7
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
EventRouter 50.98% 26 / 51 71.42% 5 / 7 21.78 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 broadcastStateChange 100.00% 3 / 3 100.00% 1 / 1 2
 routeEvent 100.00% 11 / 11 100.00% 1 / 1 2
 routePublish 0.00% 0 / 16 0.00% 0 / 1 2
 routeMqttMessage 100.00% 9 / 9 100.00% 1 / 1 1
 routeTrigger 0.00% 0 / 9 0.00% 0 / 1 2
 deliverToMatched 100.00% 2 / 2 100.00% 1 / 1 2
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Closure;
8use Psr\Log\LoggerInterface;
9use Stewart\Contracts\Event\HaEvent;
10use Stewart\Contracts\Mqtt\MqttMessage;
11use Stewart\Contracts\State\StateChange;
12use Stewart\Contracts\Topic\TopicEvent;
13use Stewart\Contracts\Trigger\TriggerEvent;
14use Stewart\Runtime\Broker\Mqtt\MqttMessageRouter;
15use Stewart\Runtime\Ipc\Message\BrokerMessage;
16use Stewart\Runtime\Ipc\Message\EventFired;
17use Stewart\Runtime\Ipc\Message\MqttMessageDelivery;
18use Stewart\Runtime\Ipc\Message\Publish;
19use Stewart\Runtime\Ipc\Message\TopicMessage;
20use Stewart\Runtime\Ipc\Message\TriggerFired;
21use Stewart\Runtime\Ipc\Wire\EncodedStateChange;
22use Stewart\Runtime\Model\Collection\SubscriptionIdCollection;
23use Stewart\Runtime\Model\SubscriptionKind;
24use Stewart\Support\Wire\TriggerSpec;
25
26final readonly class EventRouter implements MqttMessageRouter
27{
28    public function __construct(
29        private WorkerSlotRegistry $slots,
30        private SubscriptionRegistry $registry,
31        private LoggerInterface $logger,
32    ) {}
33
34    public function broadcastStateChange(StateChange $change): void
35    {
36        $encoded = new EncodedStateChange($change);
37
38        foreach ($this->slots->listLiveHandles() as $handle) {
39            $handle->sendStateChange($encoded);
40        }
41    }
42
43    public function routeEvent(HaEvent $event): void
44    {
45        $routes = $this->registry->findRoutes(SubscriptionKind::Event, $event->type);
46
47        $this->deliverToMatched(
48            $routes,
49            static fn(SubscriptionIdCollection $deliverTo): BrokerMessage => new EventFired($event, $deliverTo->listValues()),
50        );
51
52        if ($routes->isEmpty()) {
53            return;
54        }
55
56        $this->logger->debug('Home Assistant event routed', [
57            'event_type' => $event->type,
58            'workers' => $routes->listWorkerIds()->toInts(),
59        ]);
60    }
61
62    public function routePublish(Publish $message): void
63    {
64        $event = new TopicEvent(
65            topic: $message->topic,
66            payload: $message->payload,
67            publisher: $message->publisherScope->appId,
68            publishedAt: $message->publishedAt,
69        );
70
71        $routes = $this->registry->findRoutes(SubscriptionKind::Topic, $message->topic);
72
73        $this->deliverToMatched(
74            $routes,
75            static fn(SubscriptionIdCollection $deliverTo): BrokerMessage => new TopicMessage($event, $deliverTo->listValues()),
76        );
77
78        $this->logger->debug('Topic published', [
79            'topic' => $message->topic,
80            'from' => $message->publisherScope->wireValue(),
81            'workers' => $routes->listWorkerIds()->toInts(),
82        ]);
83    }
84
85    public function routeMqttMessage(MqttMessage $message): void
86    {
87        $routes = $this->registry->findRoutes(SubscriptionKind::Mqtt, $message->topic);
88
89        $this->deliverToMatched(
90            $routes,
91            static fn(SubscriptionIdCollection $deliverTo): BrokerMessage => new MqttMessageDelivery($message, $deliverTo->listValues()),
92        );
93
94        $this->logger->debug('MQTT message routed', [
95            'topic' => $message->topic,
96            'workers' => $routes->listWorkerIds()->toInts(),
97        ]);
98    }
99
100    public function routeTrigger(TriggerSpec $spec, TriggerEvent $event): void
101    {
102        $routes = $this->registry->findRoutes(SubscriptionKind::Trigger, $spec->getSharingKey());
103
104        $this->deliverToMatched(
105            $routes,
106            static fn(SubscriptionIdCollection $deliverTo): BrokerMessage => new TriggerFired($event, $deliverTo->listValues()),
107        );
108
109        $this->logger->debug('Home Assistant trigger routed', [
110            'platform' => $event->getPlatform(),
111            'workers' => $routes->listWorkerIds()->toInts(),
112        ]);
113    }
114
115    /** @param Closure(SubscriptionIdCollection): BrokerMessage $message */
116    private function deliverToMatched(SubscriptionRoutes $routes, Closure $message): void
117    {
118        foreach ($routes->listWorkerIds() as $workerId) {
119            $this->slots->findHandleForWorker($workerId)?->send($message($routes->listSubscriptionIdsForWorker($workerId)));
120        }
121    }
122}