Lines 100.00% 39 / 39
Functions and Methods 100.00% 8 / 8
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
PendingEmission 100.00% 39 / 39 100.00% 8 / 8 12 100.00% 1 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 hold 100.00% 3 / 3 100.00% 1 / 1 1
 isArmed 100.00% 1 / 1 100.00% 1 / 1 1
 arm 100.00% 9 / 9 100.00% 1 / 1 1
 disarm 100.00% 7 / 7 100.00% 1 / 1 1
 flushInto 100.00% 8 / 8 100.00% 1 / 1 2
 settleIfCurrent 100.00% 6 / 6 100.00% 1 / 1 2
 syncOutstanding 100.00% 4 / 4 100.00% 1 / 1 3
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Support\Stream;
6
7use Closure;
8use Stewart\Contracts\Time\Duration;
9use Stewart\Contracts\Time\TimerHandle;
10use Stewart\Contracts\Time\Timers;
11
12/**
13 * @internal
14 * @template T
15 */
16final class PendingEmission
17{
18    private ?TimerHandle $timer = null;
19
20    private int $generation = 0;
21
22    private bool $armed = false;
23
24    private bool $holding = false;
25
26    /** @var T|null */
27    private mixed $event = null;
28
29    public function __construct(
30        private readonly Timers $timers,
31        private readonly Duration $window,
32        private readonly SubscriptionScope $scope,
33        private readonly int $stage,
34    ) {}
35
36    /** @param T $event */
37    public function hold(mixed $event): void
38    {
39        $this->event = $event;
40        $this->holding = true;
41
42        $this->syncOutstanding();
43    }
44
45    public function isArmed(): bool
46    {
47        return $this->armed;
48    }
49
50    /** @param Closure(): void $onSettle */
51    public function arm(Closure $onSettle): void
52    {
53        $this->timer?->cancel();
54        $generation = ++$this->generation;
55        $this->armed = true;
56
57        $this->syncOutstanding();
58
59        // The timer only queues the decision, so events queued before it still count as inside the window.
60        $this->timer = $this->timers->startTimer($this->window, function () use ($generation, $onSettle): void {
61            $this->scope->emit(function () use ($generation, $onSettle): void {
62                $this->settleIfCurrent($generation, $onSettle);
63            });
64        });
65    }
66
67    public function disarm(): void
68    {
69        $this->timer?->cancel();
70        $this->timer = null;
71        ++$this->generation;
72        $this->armed = false;
73        $this->event = null;
74        $this->holding = false;
75
76        $this->syncOutstanding();
77    }
78
79    /** @param Closure(T): void $downstream */
80    public function flushInto(Closure $downstream): bool
81    {
82        if (!$this->holding) {
83            return false;
84        }
85
86        /** @var T $held */
87        $held = $this->event;
88        $this->event = null;
89        $this->holding = false;
90
91        $this->syncOutstanding();
92
93        $downstream($held);
94
95        return true;
96    }
97
98    /** @param Closure(): void $onSettle */
99    private function settleIfCurrent(int $generation, Closure $onSettle): void
100    {
101        if ($generation !== $this->generation) {
102            return;
103        }
104
105        $this->timer = null;
106        $this->armed = false;
107
108        $this->syncOutstanding();
109
110        $onSettle();
111    }
112
113    private function syncOutstanding(): void
114    {
115        if ($this->armed && $this->holding) {
116            $this->scope->trackOutstanding($this, $this->stage);
117
118            return;
119        }
120
121        $this->scope->untrackOutstanding($this);
122    }
123}