Lines 100.00% 14 / 14
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% 13 / 13 100.00% 1 / 1 5
19final readonly class DebounceOperator implements StreamSource
20{
21    /**
22     * @param StreamSource<T> $source
23     * @param Closure(T): string $partition
24     * @throws TimeException
25     */
26    public function __construct(
27        private Timers $timers,
28        private StreamSource $source,
29        private Duration $window,
30        private Edge $edge,
31        private Closure $partition,
32    ) {
33        $window->requireAtLeastOneMillisecond('debounce');
34    }
35
36    public function attach(SubscriptionScope $scope, Closure $downstream): Subscription
37    {
38        $edge = $this->edge;
39        $partition = $this->partition;
40        /** @var PendingEmissions<T> $pendings */
41        $pendings = new PendingEmissions($this->timers, $this->window, $scope, $scope->claimStage(), $downstream);
42
43        $scope->onTeardown($pendings->disarmAll(...));
44
45        return $this->source->attach($scope, static function (mixed $event) use ($pendings, $partition, $edge, $downstream): void {
46            $key = $partition($event);
47            $leads = !$pendings->isArmed($key) && $edge->emitsLeading();
48
49            if (!$leads && $edge->emitsTrailing()) {
50                $pendings->hold($key, $event);
51            }
52
53            $pendings->arm($key);
54
55            if ($leads) {
56                $downstream($event);
57            }
58        });
59    }
60}