Lines 96.00% 24 / 25
Functions and Methods 75.00% 3 / 4
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
SubscribeHandler 96.00% 24 / 25 75.00% 3 / 4 7 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 handledMessageClass 0.00% 0 / 1 0.00% 0 / 1 2
 handle 100.00% 22 / 22 100.00% 1 / 1 4
 refuseSubscription 100.00% 1 / 1 100.00% 1 / 1 1
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker\Message;
6
7use Psr\Log\LoggerInterface;
8use Stewart\Runtime\Broker\BrokerSubscription;
9use Stewart\Runtime\Broker\SubscriptionRegistry;
10use Stewart\Runtime\Broker\WorkerHandle;
11use Stewart\Runtime\Ipc\Message\Subscribe;
12use Stewart\Runtime\Ipc\Message\SubscriptionAck;
13use Stewart\Runtime\Ipc\Message\WorkerMessage;
14
15/** @implements WorkerMessageHandler<Subscribe> */
16final readonly class SubscribeHandler implements WorkerMessageHandler
17{
18    public function __construct(
19        private SubscriptionRegistry $registry,
20        private LoggerInterface $logger,
21    ) {}
22
23    public function handledMessageClass(): string
24    {
25        return Subscribe::class;
26    }
27
28    /** @param Subscribe $message */
29    public function handle(WorkerHandle $handle, WorkerMessage $message): void
30    {
31        if (!$message->kind->isBrokerRouted()) {
32            $this->refuseSubscription($handle, $message, \sprintf('%s subscriptions are worker-local and must not be announced.', $message->kind->value));
33
34            return;
35        }
36
37        if ($message->kind->needsTriggerSpec()) {
38            $this->refuseSubscription($handle, $message, \sprintf('%s subscriptions are announced with subscribe_trigger.', $message->kind->value));
39
40            return;
41        }
42
43        if (!$message->kind->acceptsSelector($message->selector)) {
44            $this->refuseSubscription($handle, $message, \sprintf('%s subscriptions do not accept %s selectors.', $message->kind->value, $message->selector->getKind()->value));
45
46            return;
47        }
48
49        $this->registry->add(new BrokerSubscription(
50            subscriptionId: $message->subscriptionId,
51            workerId: $handle->id,
52            scope: $message->scope,
53            kind: $message->kind,
54            selector: $message->selector,
55        ));
56
57        $handle->send(new SubscriptionAck($message->subscriptionId, true, null));
58
59        $this->logger->debug('Subscription registered', [
60            'app' => $message->scope->wireValue(),
61            'kind' => $message->kind->value,
62            'selector' => $message->selector->toCanonicalKey(),
63        ]);
64    }
65
66    private function refuseSubscription(WorkerHandle $handle, Subscribe $message, string $reason): void
67    {
68        $handle->send(new SubscriptionAck($message->subscriptionId, false, $reason));
69    }
70}