Lines 92.30% 12 / 13
Functions and Methods 75.00% 3 / 4
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
ExposedCommandStreams 92.30% 12 / 13 75.00% 3 / 4 8.03 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 watchCommands 100.00% 2 / 2 100.00% 1 / 1 1
 deliverCommand 100.00% 7 / 7 100.00% 1 / 1 2
 describeDiscard 66.66% 2 / 3 0.00% 0 / 1 4.59
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Worker\Exposure;
6
7use Stewart\Contracts\EventStream;
8use Stewart\Contracts\Exposure\Command\ExposedCommand;
9use Stewart\Contracts\Exposure\ExposedEntityKey;
10use Stewart\Contracts\Selector\Selector;
11use Stewart\Contracts\Time\Timers;
12use Stewart\Runtime\Dispatch\DispatchSource;
13use Stewart\Runtime\Dispatch\LocalDispatcher;
14use Stewart\Runtime\Dispatch\QueueDiscardReason;
15use Stewart\Runtime\Model\ResourceScope;
16use Stewart\Runtime\Model\SubscriptionKind;
17use Stewart\Support\Stream\OperatorStream;
18
19final readonly class ExposedCommandStreams
20{
21    public function __construct(
22        private LocalDispatcher $dispatcher,
23        private Timers $timers,
24        private ExposedCommandSettlements $settlements,
25    ) {}
26
27    /** @return EventStream<ExposedCommand> */
28    public function watchCommands(ResourceScope $scope, ExposedEntityKey $key): EventStream
29    {
30        /** @var DispatchSource<ExposedCommand> $source */
31        $source = new DispatchSource($this->dispatcher, $scope, SubscriptionKind::ExposedCommand, Selector::exact($key->value));
32
33        return new SettlingCommandStream(new OperatorStream($source, $this->timers), $this->settlements);
34    }
35
36    public function deliverCommand(string $commandId, ResourceScope $scope, ExposedEntityKey $key, ExposedCommand $command, ExposedHandle $handle): void
37    {
38        $this->settlements->openCommand($commandId, $command, $handle);
39
40        $settlements = $this->settlements;
41        $refuseDiscarded = static function (QueueDiscardReason $reason) use ($settlements, $command, $scope): void {
42            $settlements->rejectCommand($command, self::describeDiscard($reason, $scope));
43        };
44
45        if (!$this->dispatcher->dispatchExposedCommand($scope, $key, $command, $refuseDiscarded)) {
46            $this->settlements->acceptCommand($command);
47        }
48    }
49
50    private static function describeDiscard(QueueDiscardReason $reason, ResourceScope $scope): string
51    {
52        return match ($reason) {
53            QueueDiscardReason::Paused => \sprintf('App %s was paused before it handled the command.', $scope),
54            QueueDiscardReason::Overflow => \sprintf('App %s dropped the command because it is falling behind.', $scope),
55            QueueDiscardReason::Cancelled => \sprintf('App %s stopped watching commands for this entity.', $scope),
56        };
57    }
58}