Lines 100.00% 9 / 9
Functions and Methods 100.00% 2 / 2
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
WhenStableForOperator 100.00% 9 / 9 100.00% 2 / 2 3 100.00% 1 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 attach 100.00% 8 / 8 100.00% 1 / 1 2
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 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}