Lines 100.00% 15 / 15
Functions and Methods 100.00% 8 / 8
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
PendingEmissions 100.00% 15 / 15 100.00% 8 / 8 9 100.00% 1 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 isArmed 100.00% 1 / 1 100.00% 1 / 1 1
 hold 100.00% 1 / 1 100.00% 1 / 1 1
 arm 100.00% 4 / 4 100.00% 1 / 1 1
 disarm 100.00% 2 / 2 100.00% 1 / 1 1
 disarmAll 100.00% 3 / 3 100.00% 1 / 1 2
 pendingFor 100.00% 1 / 1 100.00% 1 / 1 1
 settle 100.00% 2 / 2 100.00% 1 / 1 1
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Support\Stream;
6
7use Closure;
8use Stewart\Contracts\Time\Duration;
9use Stewart\Contracts\Time\Timers;
10
11/**
12 * @internal
13 * @template T
14 */
15final class PendingEmissions
16{
17    /** @var array<string, PendingEmission<T>> */
18    private array $pending = [];
19
20    /** @param Closure(T): void $downstream */
21    public function __construct(
22        private readonly Timers $timers,
23        private readonly Duration $window,
24        private readonly SubscriptionScope $scope,
25        private readonly int $stage,
26        private readonly Closure $downstream,
27    ) {}
28
29    public function isArmed(string $key): bool
30    {
31        return ($this->pending[$key] ?? null)?->isArmed() ?? false;
32    }
33
34    /** @param T $event */
35    public function hold(string $key, mixed $event): void
36    {
37        $this->pendingFor($key)->hold($event);
38    }
39
40    public function arm(string $key): void
41    {
42        $pending = $this->pendingFor($key);
43
44        $pending->arm(function () use ($key, $pending): void {
45            $this->settle($key, $pending);
46        });
47    }
48
49    public function disarm(string $key): void
50    {
51        ($this->pending[$key] ?? null)?->disarm();
52
53        unset($this->pending[$key]);
54    }
55
56    public function disarmAll(): void
57    {
58        foreach ($this->pending as $pending) {
59            $pending->disarm();
60        }
61
62        $this->pending = [];
63    }
64
65    /** @return PendingEmission<T> */
66    private function pendingFor(string $key): PendingEmission
67    {
68        /** @var PendingEmission<T> */
69        return $this->pending[$key] ??= new PendingEmission($this->timers, $this->window, $this->scope, $this->stage);
70    }
71
72    /** @param PendingEmission<T> $pending */
73    private function settle(string $key, PendingEmission $pending): void
74    {
75        unset($this->pending[$key]);
76
77        $pending->flushInto($this->downstream);
78    }
79}