Lines 93.06% 94 / 101
Methods 76.47% 13 / 17
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 push 100.00% 1 / 1 100.00% 1 / 1 1
 pushBurst 100.00% 3 / 3 100.00% 1 / 1 2
 emit 83.33% 5 / 6 0.00% 0 / 1 3.04
 startDrainingIfPending 100.00% 2 / 2 100.00% 1 / 1 3
 discardPendingWork 87.50% 7 / 8 0.00% 0 / 1 3.02
 cancel 72.72% 8 / 11 0.00% 0 / 1 4.32
 pushWithin 81.81% 9 / 11 0.00% 0 / 1 4.10
 dropOnOverflow 100.00% 6 / 6 100.00% 1 / 1 3
 findOldestSupersededIndex 100.00% 12 / 12 100.00% 1 / 1 6
 dropSupersededStateChange 100.00% 12 / 12 100.00% 1 / 1 4
 dropOldestEvent 100.00% 5 / 5 100.00% 1 / 1 3
 recordOverflowDrop 100.00% 4 / 4 100.00% 1 / 1 1
 enqueue 100.00% 3 / 3 100.00% 1 / 1 2
 startDraining 100.00% 4 / 4 100.00% 1 / 1 2
 drain 100.00% 6 / 6 100.00% 1 / 1 5
 runQueuedWork 100.00% 6 / 6 100.00% 1 / 1 3
15final class SubscriptionQueue
16{
17    /** @var SplQueue<QueuedWork> */
18    private SplQueue $pending;
19
20    private int $pendingEvents = 0;
21
22    private bool $draining = false;
23
24    private bool $cancelled = false;
25
26    private int $dropped = 0;
27
28    public function __construct(
29        public readonly RegisteredSubscription $subscription,
30        private readonly int $limit,
31        private readonly DispatchListener $listener,
32        private readonly ScopeState $scopeState,
33    ) {
34        $this->pending = new SplQueue();
35    }
36
37    /** @param ?Closure(QueueDiscardReason): void $onDiscarded */
38    public function push(object $event, ?Closure $onDiscarded = null): void
39    {
40        $this->pushWithin($event, $this->limit, $onDiscarded);
41    }
42
43    /** @param list<StateChange> $changes */
44    public function pushBurst(array $changes): void
45    {
46        $capacity = max($this->limit, \count($changes));
47
48        foreach ($changes as $change) {
49            $this->pushWithin($change, $capacity);
50        }
51    }
52
53    /** @param Closure(): void $emission */
54    public function emit(Closure $emission): void
55    {
56        if ($this->cancelled) {
57            return;
58        }
59
60        if ($this->scopeState->isPaused()) {
61            $this->listener->eventSuppressed($this->subscription);
62
63            return;
64        }
65
66        $this->enqueue(QueuedWork::forStreamEmission($emission));
67    }
68
69    public function startDrainingIfPending(): void
70    {
71        if (!$this->cancelled && !$this->pending->isEmpty()) {
72            $this->startDraining();
73        }
74    }
75
76    public function discardPendingWork(): void
77    {
78        if ($this->cancelled) {
79            return;
80        }
81
82        $discarded = $this->pending;
83        $this->pending = new SplQueue();
84        $this->pendingEvents = 0;
85
86        foreach ($discarded as $work) {
87            $this->listener->eventSuppressed($this->subscription);
88            $work->discard(QueueDiscardReason::Paused);
89        }
90    }
91
92    public function cancel(): void
93    {
94        if ($this->cancelled) {
95            return;
96        }
97
98        $this->cancelled = true;
99        $discarded = $this->pending;
100        $this->pending = new SplQueue();
101        $this->pendingEvents = 0;
102
103        foreach ($discarded as $work) {
104            $work->discard(QueueDiscardReason::Cancelled);
105        }
106
107        try {
108            $this->subscription->subscriptionScope->close();
109        } catch (Throwable $e) {
110            $this->listener->handlerFailed($this->subscription, $e);
111        }
112    }
113
114    /** @param ?Closure(QueueDiscardReason): void $onDiscarded */
115    private function pushWithin(object $event, int $capacity, ?Closure $onDiscarded = null): void
116    {
117        if ($this->cancelled) {
118            $onDiscarded?->__invoke(QueueDiscardReason::Cancelled);
119
120            return;
121        }
122
123        if ($this->scopeState->isPaused()) {
124            $this->listener->eventSuppressed($this->subscription);
125            $onDiscarded?->__invoke(QueueDiscardReason::Paused);
126
127            return;
128        }
129
130        if ($this->pendingEvents >= $capacity) {
131            $event = $this->dropOnOverflow($event);
132        }
133
134        ++$this->pendingEvents;
135
136        $this->enqueue(QueuedWork::forDroppableEvent($event, $this->subscription->handler, $onDiscarded));
137    }
138
139    private function dropOnOverflow(object $incoming): object
140    {
141        if ($incoming instanceof StateChange) {
142            $superseded = $this->findOldestSupersededIndex($incoming);
143
144            if ($superseded !== null) {
145                return $this->dropSupersededStateChange($superseded, $incoming);
146            }
147        }
148
149        $this->dropOldestEvent();
150
151        return $incoming;
152    }
153
154    private function findOldestSupersededIndex(StateChange $incoming): ?int
155    {
156        $firstIndexByEntity = [];
157        $supersededIndexes = [];
158
159        foreach ($this->pending as $index => $work) {
160            if (!$work->event instanceof StateChange) {
161                continue;
162            }
163
164            $entityId = $work->event->entityId->value;
165
166            if (isset($firstIndexByEntity[$entityId])) {
167                $supersededIndexes[] = $firstIndexByEntity[$entityId];
168
169                continue;
170            }
171
172            $firstIndexByEntity[$entityId] = $index;
173        }
174
175        if (isset($firstIndexByEntity[$incoming->entityId->value])) {
176            $supersededIndexes[] = $firstIndexByEntity[$incoming->entityId->value];
177        }
178
179        return $supersededIndexes === [] ? null : min($supersededIndexes);
180    }
181
182    private function dropSupersededStateChange(int $index, StateChange $incoming): StateChange
183    {
184        $superseded = $this->pending->offsetGet($index);
185        \assert($superseded->event instanceof StateChange);
186
187        $this->pending->offsetUnset($index);
188        $this->recordOverflowDrop($superseded);
189
190        $entityId = $superseded->event->entityId->value;
191        $from = $superseded->event->from;
192
193        for ($next = $index; $next < $this->pending->count(); ++$next) {
194            $work = $this->pending->offsetGet($next);
195
196            if ($work->event instanceof StateChange && $work->event->entityId->value === $entityId) {
197                $this->pending->offsetSet($next, $work->withEvent($work->event->withFrom($from)));
198
199                return $incoming;
200            }
201        }
202
203        return $incoming->withFrom($from);
204    }
205
206    private function dropOldestEvent(): void
207    {
208        foreach ($this->pending as $index => $work) {
209            if (!$work->isDroppable()) {
210                continue;
211            }
212
213            $this->pending->offsetUnset($index);
214            $this->recordOverflowDrop($work);
215
216            return;
217        }
218    }
219
220    private function recordOverflowDrop(QueuedWork $work): void
221    {
222        --$this->pendingEvents;
223        ++$this->dropped;
224
225        $this->listener->eventDropped($this->subscription, $this->dropped);
226        $work->discard(QueueDiscardReason::Overflow);
227    }
228
229    private function enqueue(QueuedWork $work): void
230    {
231        $this->pending->enqueue($work);
232
233        if ($this->scopeState->isLive()) {
234            $this->startDraining();
235        }
236    }
237
238    private function startDraining(): void
239    {
240        if ($this->draining) {
241            return;
242        }
243
244        $this->draining = true;
245
246        async($this->drain(...))->ignore();
247    }
248
249    private function drain(): void
250    {
251        try {
252            while (!$this->cancelled && !$this->scopeState->isPaused() && !$this->pending->isEmpty()) {
253                $work = $this->pending->dequeue();
254
255                if ($work->isDroppable()) {
256                    --$this->pendingEvents;
257                }
258
259                $this->runQueuedWork($work);
260            }
261        } finally {
262            $this->draining = false;
263        }
264    }
265
266    private function runQueuedWork(QueuedWork $work): void
267    {
268        try {
269            $work();
270        } catch (Throwable $e) {
271            $this->listener->handlerFailed($this->subscription, $e);
272
273            return;
274        }
275
276        if ($work->isDroppable()) {
277            $this->listener->eventDelivered($this->subscription);
278        }
279    }
280}