Lines 100.00% 27 / 27
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% 26 / 26 100.00% 1 / 1 7
19final readonly class ThrottleOperator 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('throttle');
34    }
35
36    public function attach(SubscriptionScope $scope, Closure $downstream): Subscription
37    {
38        $timers = $this->timers;
39        $windowLength = $this->window;
40        $edge = $this->edge;
41        $partition = $this->partition;
42        $stage = $scope->claimStage();
43        /** @var array<string, ThrottleWindow<T>> $windows */
44        $windows = [];
45
46        $scope->onTeardown(static function () use (&$windows): void {
47            foreach ($windows as $window) {
48                $window->cancelWindowTimer();
49            }
50        });
51
52        return $this->source->attach($scope, static function (mixed $event) use (&$windows, $partition, $timers, $windowLength, $scope, $stage, $edge, $downstream): void {
53            $key = $partition($event);
54            /** @var ThrottleWindow<T> $window */
55            $window = $windows[$key] ??= new ThrottleWindow($timers, $windowLength, $scope, $stage, $downstream, static function () use (&$windows, $key): void {
56                unset($windows[$key]);
57            });
58
59            if ($window->isOpen()) {
60                if ($edge->emitsTrailing()) {
61                    $window->holdEvent($event);
62                }
63
64                return;
65            }
66
67            $leads = $edge->emitsLeading();
68
69            if (!$leads && $edge->emitsTrailing()) {
70                $window->holdEvent($event);
71            }
72
73            $window->openWindow();
74
75            if ($leads) {
76                $downstream($event);
77            }
78        });
79    }
80}