Lines 97.97% 97 / 99
Methods 90.00% 18 / 20
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 4 / 4 100.00% 1 / 1 1
 register 100.00% 33 / 33 100.00% 1 / 1 9
 startDrainingQueuesOf 100.00% 2 / 2 100.00% 1 / 1 2
 discardPendingWorkOf 100.00% 2 / 2 100.00% 1 / 1 2
 dispatchStateChange 100.00% 1 / 1 100.00% 1 / 1 1
 dispatchStateChanges 100.00% 6 / 6 100.00% 1 / 1 4
 dispatchEvent 100.00% 1 / 1 100.00% 1 / 1 1
 dispatchTopic 100.00% 1 / 1 100.00% 1 / 1 1
 dispatchMqttMessage 0.00% 0 / 1 0.00% 0 / 1 2
 dispatchTrigger 100.00% 1 / 1 100.00% 1 / 1 1
 dispatchConnection 100.00% 3 / 3 100.00% 1 / 1 3
 dispatchExposedCommand 100.00% 12 / 12 100.00% 1 / 1 4
 cancel 100.00% 8 / 8 100.00% 1 / 1 2
 cancelAll 50.00% 1 / 2 0.00% 0 / 1 2.50
 cancelSubscriptionsOf 100.00% 2 / 2 100.00% 1 / 1 2
 countFor 100.00% 1 / 1 100.00% 1 / 1 1
 refusedSubscription 100.00% 4 / 4 100.00% 1 / 1 1
 deliver 100.00% 4 / 4 100.00% 1 / 1 4
 listStateChangeQueuesMatching 100.00% 10 / 10 100.00% 1 / 1 4
 listQueuesOf 100.00% 1 / 1 100.00% 1 / 1 1
33final class LocalDispatcher
34{
35    /** @var array<string, SubscriptionQueue> */
36    private array $queues = [];
37
38    private int $counter = 0;
39
40    private readonly string $subscriptionIdPrefix;
41
42    private readonly int $subscriptionQueueLimit;
43
44    private readonly SelectorIndex $stateIndex;
45
46    private readonly EntityFilterIndex $stateFilterIndex;
47
48    public function __construct(
49        WorkerId $workerId,
50        WorkerSettings $workerSettings,
51        private readonly DispatchListener $listener,
52        private readonly SubscriptionListener $subscriptions,
53        private readonly ScopeLifecycle $scopes,
54        RegistryCache $registry,
55    ) {
56        $this->subscriptionIdPrefix = $workerId->toProcessIdPrefix();
57        $this->subscriptionQueueLimit = $workerSettings->subscriptionBuffer;
58        $this->stateIndex = new SelectorIndex();
59        $this->stateFilterIndex = new EntityFilterIndex($registry);
60    }
61
62    public function register(
63        ResourceScope $scope,
64        SubscriptionKind $kind,
65        Selector $selector,
66        SubscriptionScope $subscriptionScope,
67        Closure $handler,
68        ?TriggerSpec $trigger = null,
69        ?EntityFilter $entityFilter = null,
70    ): Subscription {
71        if ($kind->needsTriggerSpec() !== ($trigger !== null)) {
72            throw new LogicException(\sprintf('A %s subscription %s a trigger spec.', $kind->value, $trigger === null ? 'needs' : 'takes no'));
73        }
74
75        if ($entityFilter !== null && $kind !== SubscriptionKind::StateChange) {
76            throw new LogicException(\sprintf('A %s subscription takes no entity filter.', $kind->value));
77        }
78
79        $id = SubscriptionId::fromString($this->subscriptionIdPrefix . ':' . $this->counter++);
80
81        $scopeState = $this->scopes->findOrCreateStateOf($scope);
82
83        if ($scopeState->isClosed()) {
84            return $this->refusedSubscription($id, $subscriptionScope);
85        }
86
87        $subscription = new RegisteredSubscription(
88            $id,
89            $scope,
90            $kind,
91            $selector,
92            $subscriptionScope,
93            $handler,
94            $trigger,
95            $entityFilter,
96        );
97
98        $queue = new SubscriptionQueue($subscription, $this->subscriptionQueueLimit, $this->listener, $scopeState);
99
100        $this->queues[$subscription->id->value] = $queue;
101        $subscriptionScope->deliverVia($queue->emit(...));
102
103        if ($entityFilter !== null) {
104            $this->stateFilterIndex->add($subscription->id, $entityFilter);
105        } elseif ($kind === SubscriptionKind::StateChange) {
106            $this->stateIndex->add($subscription->id, $kind, $selector);
107        }
108
109        try {
110            $this->subscriptions->subscriptionRegistered($subscription);
111        } catch (Throwable $e) {
112            unset($this->queues[$subscription->id->value]);
113            $this->stateIndex->remove($subscription->id);
114            $this->stateFilterIndex->remove($subscription->id);
115            $queue->cancel();
116
117            throw $e;
118        }
119
120        return new LocalSubscription($subscription->id, $this->cancel(...));
121    }
122
123    public function startDrainingQueuesOf(ResourceScope $scope): void
124    {
125        foreach ($this->listQueuesOf($scope) as $queue) {
126            $queue->startDrainingIfPending();
127        }
128    }
129
130    public function discardPendingWorkOf(ResourceScope $scope): void
131    {
132        foreach ($this->listQueuesOf($scope) as $queue) {
133            $queue->discardPendingWork();
134        }
135    }
136
137    public function dispatchStateChange(StateChange $change): void
138    {
139        $this->dispatchStateChanges(StateChangeCollection::fromList([$change]));
140    }
141
142    public function dispatchStateChanges(StateChangeCollection $changes): void
143    {
144        $changesByQueue = [];
145
146        foreach ($changes as $change) {
147            foreach ($this->listStateChangeQueuesMatching($change) as $subscriptionId => $queue) {
148                $changesByQueue[$subscriptionId][] = $change;
149            }
150        }
151
152        foreach ($changesByQueue as $subscriptionId => $queueChanges) {
153            $this->queues[$subscriptionId]->pushBurst($queueChanges);
154        }
155    }
156
157    public function dispatchEvent(HaEvent $event, SubscriptionIdCollection $deliverTo): void
158    {
159        $this->deliver(SubscriptionKind::Event, $event, $deliverTo);
160    }
161
162    public function dispatchTopic(TopicEvent $event, SubscriptionIdCollection $deliverTo): void
163    {
164        $this->deliver(SubscriptionKind::Topic, $event, $deliverTo);
165    }
166
167    public function dispatchMqttMessage(MqttMessage $message, SubscriptionIdCollection $deliverTo): void
168    {
169        $this->deliver(SubscriptionKind::Mqtt, $message, $deliverTo);
170    }
171
172    public function dispatchTrigger(TriggerEvent $event, SubscriptionIdCollection $deliverTo): void
173    {
174        $this->deliver(SubscriptionKind::Trigger, $event, $deliverTo);
175    }
176
177    public function dispatchConnection(ConnectionEvent $event): void
178    {
179        foreach ($this->queues as $queue) {
180            if ($queue->subscription->kind === SubscriptionKind::Connection) {
181                $queue->push($event);
182            }
183        }
184    }
185
186    /** @param Closure(QueueDiscardReason): void $onDiscarded */
187    public function dispatchExposedCommand(ResourceScope $scope, ExposedEntityKey $key, ExposedCommand $command, Closure $onDiscarded): bool
188    {
189        $queues = array_filter(
190            $this->listQueuesOf($scope),
191            static fn(SubscriptionQueue $queue): bool => $queue->subscription->kind === SubscriptionKind::ExposedCommand && $queue->subscription->selector->matches($key->value),
192        );
193        $undiscarded = \count($queues);
194
195        // Another subscription may still handle the command, so only the last discard reports it.
196        $discardOne = static function (QueueDiscardReason $reason) use (&$undiscarded, $onDiscarded): void {
197            if (--$undiscarded === 0) {
198                $onDiscarded($reason);
199            }
200        };
201
202        foreach ($queues as $queue) {
203            $queue->push($command, $discardOne);
204        }
205
206        return $queues !== [];
207    }
208
209    public function cancel(SubscriptionId $subscriptionId): void
210    {
211        $queue = $this->queues[$subscriptionId->value] ?? null;
212
213        if ($queue === null) {
214            return;
215        }
216
217        unset($this->queues[$subscriptionId->value]);
218        $this->stateIndex->remove($subscriptionId);
219        $this->stateFilterIndex->remove($subscriptionId);
220        $queue->cancel();
221
222        $this->subscriptions->subscriptionCancelled($queue->subscription);
223    }
224
225    public function cancelAll(): void
226    {
227        foreach ($this->queues as $queue) {
228            $this->cancel($queue->subscription->id);
229        }
230    }
231
232    public function cancelSubscriptionsOf(ResourceScope $scope): void
233    {
234        foreach ($this->listQueuesOf($scope) as $queue) {
235            $this->cancel($queue->subscription->id);
236        }
237    }
238
239    public function countFor(ResourceScope $scope): int
240    {
241        return \count($this->listQueuesOf($scope));
242    }
243
244    private function refusedSubscription(SubscriptionId $id, SubscriptionScope $subscriptionScope): Subscription
245    {
246        $subscriptionScope->close();
247
248        $subscription = new LocalSubscription($id, static function (): void {});
249        $subscription->unsubscribe();
250
251        return $subscription;
252    }
253
254    private function deliver(SubscriptionKind $kind, object $event, SubscriptionIdCollection $deliverTo): void
255    {
256        foreach ($deliverTo as $id) {
257            $queue = $this->queues[$id->value] ?? null;
258
259            if ($queue !== null && $queue->subscription->kind === $kind) {
260                $queue->push($event);
261            }
262        }
263    }
264
265    /** @return array<string, SubscriptionQueue> */
266    private function listStateChangeQueuesMatching(StateChange $change): array
267    {
268        $matching = [];
269        $deliverTo = [
270            ...$this->stateIndex->findMatching(SubscriptionKind::StateChange, $change->entityId->value),
271            ...$this->stateFilterIndex->findMatching($change->entityId),
272        ];
273
274        foreach ($deliverTo as $id) {
275            $queue = $this->queues[$id->value] ?? null;
276
277            if ($queue !== null && $queue->subscription->kind === SubscriptionKind::StateChange) {
278                $matching[$id->value] = $queue;
279            }
280        }
281
282        return $matching;
283    }
284
285    /** @return array<string, SubscriptionQueue> */
286    private function listQueuesOf(ResourceScope $scope): array
287    {
288        return array_filter($this->queues, static fn(SubscriptionQueue $queue): bool => $queue->subscription->scope->equals($scope));
289    }
290}