Lines
92.30%
12 / 13
Methods
75.00%
3 / 4
Classes
0.00%
0 / 1
| Name | Lines | Methods | CRAP | ||||
|---|---|---|---|---|---|---|---|
| __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 | ||
| 19 | final 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 | } |