Lines 100.00% 24 / 24
Functions and Methods 100.00% 7 / 7
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
SubscriptionRegistry 100.00% 24 / 24 100.00% 7 / 7 16 100.00% 1 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 add 100.00% 4 / 4 100.00% 1 / 1 2
 remove 100.00% 7 / 7 100.00% 1 / 1 5
 removeWorkerSubscriptions 100.00% 3 / 3 100.00% 1 / 1 3
 findRoutes 100.00% 7 / 7 100.00% 1 / 1 3
 listSubscriptions 100.00% 1 / 1 100.00% 1 / 1 1
 getRoutingStats 100.00% 1 / 1 100.00% 1 / 1 1
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Stewart\Runtime\Dispatch\SelectorIndex;
8use Stewart\Runtime\Model\Collection\SubscriptionIdCollection;
9use Stewart\Runtime\Model\RoutingStats;
10use Stewart\Runtime\Model\SubscriptionId;
11use Stewart\Runtime\Model\SubscriptionKind;
12use Stewart\Runtime\Model\WorkerId;
13
14final class SubscriptionRegistry
15{
16    /** @var array<string, BrokerSubscription> */
17    private array $subscriptions = [];
18
19    private readonly SelectorIndex $index;
20
21    /** @param iterable<BrokerSubscriptionListener> $brokerSubscriptionListeners */
22    public function __construct(private readonly iterable $brokerSubscriptionListeners = [])
23    {
24        $this->index = new SelectorIndex();
25    }
26
27    public function add(BrokerSubscription $subscription): void
28    {
29        $this->subscriptions[$subscription->subscriptionId->value] = $subscription;
30        $this->index->add($subscription->subscriptionId, $subscription->kind, $subscription->selector);
31
32        foreach ($this->brokerSubscriptionListeners as $listener) {
33            $listener->subscriptionAdded($subscription);
34        }
35    }
36
37    public function remove(SubscriptionId $subscriptionId, ?WorkerId $ownedBy = null): void
38    {
39        $subscription = $this->subscriptions[$subscriptionId->value] ?? null;
40
41        if ($subscription === null || ($ownedBy !== null && !$subscription->workerId->equals($ownedBy))) {
42            return;
43        }
44
45        unset($this->subscriptions[$subscriptionId->value]);
46        $this->index->remove($subscriptionId);
47
48        foreach ($this->brokerSubscriptionListeners as $listener) {
49            $listener->subscriptionRemoved($subscription);
50        }
51    }
52
53    public function removeWorkerSubscriptions(WorkerId $workerId): void
54    {
55        foreach ($this->subscriptions as $subscription) {
56            if ($subscription->workerId->equals($workerId)) {
57                $this->remove($subscription->subscriptionId);
58            }
59        }
60    }
61
62    public function findRoutes(SubscriptionKind $kind, string $key): SubscriptionRoutes
63    {
64        $idsByWorker = [];
65
66        foreach ($this->index->findMatching($kind, $key) as $subscriptionId) {
67            $subscription = $this->subscriptions[$subscriptionId->value] ?? null;
68
69            if ($subscription !== null) {
70                $idsByWorker[$subscription->workerId->value][] = $subscriptionId;
71            }
72        }
73
74        ksort($idsByWorker);
75
76        return new SubscriptionRoutes(array_map(SubscriptionIdCollection::fromList(...), $idsByWorker));
77    }
78
79    /** @return list<BrokerSubscription> */
80    public function listSubscriptions(): array
81    {
82        return array_values($this->subscriptions);
83    }
84
85    public function getRoutingStats(): RoutingStats
86    {
87        return $this->index->getRoutingStats();
88    }
89}