Lines 100.00% 20 / 20
Functions and Methods 100.00% 2 / 2
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
WhenChangedToOperator 100.00% 20 / 20 100.00% 2 / 2 8 100.00% 1 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 attach 100.00% 19 / 19 100.00% 1 / 1 7
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Support\Stream;
6
7use Closure;
8use Stewart\Contracts\Exception\TimeException;
9use Stewart\Contracts\State\StateChange;
10use Stewart\Contracts\Subscription;
11use Stewart\Contracts\Time\Duration;
12use Stewart\Contracts\Time\Timers;
13
14/**
15 * @internal
16 * @implements StreamSource<StateChange>
17 */
18final readonly class WhenChangedToOperator implements StreamSource
19{
20    /**
21     * @param StreamSource<StateChange> $source
22     * @throws TimeException
23     */
24    public function __construct(
25        private Timers $timers,
26        private StreamSource $source,
27        private string $state,
28        private ?Duration $window,
29        private PreviousStateRule $previousStateRule,
30    ) {
31        $window?->requireAtLeastOneMillisecond('whenChangedTo');
32    }
33
34    public function attach(SubscriptionScope $scope, Closure $downstream): Subscription
35    {
36        $state = $this->state;
37        $window = $this->window;
38        $previousStateRule = $this->previousStateRule;
39
40        if ($window === null) {
41            return $this->source->attach($scope, static function (StateChange $change) use ($state, $previousStateRule, $downstream): void {
42                if ($change->changedTo($state) && $previousStateRule->permits($change)) {
43                    $downstream($change);
44                }
45            });
46        }
47
48        /** @var PendingEmissions<StateChange> $pendings */
49        $pendings = new PendingEmissions($this->timers, $window, $scope, $scope->claimStage(), $downstream);
50
51        $scope->onTeardown($pendings->disarmAll(...));
52
53        return $this->source->attach($scope, static function (StateChange $change) use ($state, $previousStateRule, $pendings): void {
54            $entityId = $change->entityId->value;
55
56            if ($change->changedTo($state) && $previousStateRule->permits($change)) {
57                $pendings->hold($entityId, $change);
58                $pendings->arm($entityId);
59
60                return;
61            }
62
63            if ($change->to?->state !== $state) {
64                $pendings->disarm($entityId);
65            }
66        });
67    }
68}