Lines 100.00% 9 / 9
Methods 100.00% 2 / 2
Classes 100.00% 1 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 attach 100.00% 8 / 8 100.00% 1 / 1 2
18final readonly class WhenStableForOperator 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 Duration $window,
28    ) {
29        $window->requireAtLeastOneMillisecond('whenStableFor');
30    }
31
32    public function attach(SubscriptionScope $scope, Closure $downstream): Subscription
33    {
34        /** @var PendingEmissions<StateChange> $pendings */
35        $pendings = new PendingEmissions($this->timers, $this->window, $scope, $scope->claimStage(), $downstream);
36
37        $scope->onTeardown($pendings->disarmAll(...));
38
39        return $this->source->attach($scope, static function (StateChange $change) use ($pendings): void {
40            if (!$change->hasStateChanged()) {
41                return;
42            }
43
44            $pendings->hold($change->entityId->value, $change);
45            $pendings->arm($change->entityId->value);
46        });
47    }
48}