Lines 96.15% 50 / 52
Methods 84.61% 11 / 13
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 4 / 4 100.00% 1 / 1 1
 deliverVia 100.00% 1 / 1 100.00% 1 / 1 1
 emit 100.00% 5 / 5 100.00% 1 / 1 2
 onTeardown 100.00% 4 / 4 100.00% 1 / 1 2
 attachedTo 100.00% 3 / 3 100.00% 1 / 1 2
 claimStage 0.00% 0 / 1 0.00% 0 / 1 2
 trackOutstanding 66.66% 2 / 3 0.00% 0 / 1 2.15
 untrackOutstanding 100.00% 2 / 2 100.00% 1 / 1 1
 closeWhenDownstreamSettles 100.00% 2 / 2 100.00% 1 / 1 1
 isActive 100.00% 1 / 1 100.00% 1 / 1 1
 close 100.00% 14 / 14 100.00% 1 / 1 5
 runEmission 100.00% 6 / 6 100.00% 1 / 1 2
 closeIfSettled 100.00% 6 / 6 100.00% 1 / 1 6
13final class SubscriptionScope
14{
15    /** @var Closure(Closure(): void): void */
16    private Closure $sink;
17
18    /** @var list<Closure(): void> */
19    private array $teardowns = [];
20
21    private ?Subscription $subscription = null;
22
23    private bool $closed = false;
24
25    private int $stages = 0;
26
27    /** @var SplObjectStorage<object, int> */
28    private SplObjectStorage $outstanding;
29
30    private ?int $closeBelowStage = null;
31
32    private int $runningEmissions = 0;
33
34    public function __construct()
35    {
36        $this->sink = static function (Closure $emission): void {
37            $emission();
38        };
39        $this->outstanding = new SplObjectStorage();
40    }
41
42    /** @param Closure(Closure(): void): void $sink */
43    public function deliverVia(Closure $sink): void
44    {
45        $this->sink = $sink;
46    }
47
48    /** @param Closure(): void $emission */
49    public function emit(Closure $emission): void
50    {
51        if ($this->closed) {
52            return;
53        }
54
55        ($this->sink)(function () use ($emission): void {
56            $this->runEmission($emission);
57        });
58    }
59
60    /** @param Closure(): void $teardown */
61    public function onTeardown(Closure $teardown): void
62    {
63        if ($this->closed) {
64            $teardown();
65
66            return;
67        }
68
69        $this->teardowns[] = $teardown;
70    }
71
72    public function attachedTo(Subscription $subscription): void
73    {
74        $this->subscription = $subscription;
75
76        if ($this->closed) {
77            $subscription->unsubscribe();
78        }
79    }
80
81    public function claimStage(): int
82    {
83        return $this->stages++;
84    }
85
86    public function trackOutstanding(object $pending, int $stage): void
87    {
88        if ($this->closed) {
89            return;
90        }
91
92        $this->outstanding[$pending] = $stage;
93    }
94
95    public function untrackOutstanding(object $pending): void
96    {
97        unset($this->outstanding[$pending]);
98
99        $this->closeIfSettled();
100    }
101
102    public function closeWhenDownstreamSettles(int $stage): void
103    {
104        $this->closeBelowStage = min($this->closeBelowStage ?? $stage, $stage);
105
106        $this->closeIfSettled();
107    }
108
109    public function isActive(): bool
110    {
111        return !$this->closed;
112    }
113
114    public function close(): void
115    {
116        if ($this->closed) {
117            return;
118        }
119
120        $this->closed = true;
121        $this->outstanding = new SplObjectStorage();
122
123        $teardowns = $this->teardowns;
124        $this->teardowns = [];
125        $failure = null;
126
127        foreach ($teardowns as $teardown) {
128            try {
129                $teardown();
130            } catch (Throwable $e) {
131                $failure ??= $e;
132            }
133        }
134
135        $this->subscription?->unsubscribe();
136
137        if ($failure !== null) {
138            throw $failure;
139        }
140    }
141
142    /** @param Closure(): void $emission */
143    private function runEmission(Closure $emission): void
144    {
145        if ($this->closed) {
146            return;
147        }
148
149        ++$this->runningEmissions;
150
151        try {
152            $emission();
153        } finally {
154            --$this->runningEmissions;
155            $this->closeIfSettled();
156        }
157    }
158
159    private function closeIfSettled(): void
160    {
161        if ($this->closed || $this->closeBelowStage === null || $this->runningEmissions > 0) {
162            return;
163        }
164
165        foreach ($this->outstanding as $pending) {
166            if ($this->outstanding[$pending] < $this->closeBelowStage) {
167                return;
168            }
169        }
170
171        $this->close();
172    }
173}