Lines 95.00% 19 / 20
Methods 83.33% 5 / 6
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 attach 100.00% 8 / 8 100.00% 1 / 1 1
 push 100.00% 4 / 4 100.00% 1 / 1 2
 cancelAttachments 100.00% 5 / 5 100.00% 1 / 1 2
 countSubscribers 100.00% 1 / 1 100.00% 1 / 1 1
 countAttachments 0.00% 0 / 1 0.00% 0 / 1 2
17final class PushSource implements StreamSource
18{
19    /** @var array<string, Closure(T): void> */
20    private array $targets = [];
21
22    /** @var array<string, SubscriptionScope> */
23    private array $scopes = [];
24
25    private int $counter = 0;
26
27    public function __construct(private readonly HandlerFibers $fibers = new HandlerFibers()) {}
28
29    public function attach(SubscriptionScope $scope, Closure $downstream): Subscription
30    {
31        $id = 'push:' . $this->counter++;
32        $this->targets[$id] = $downstream;
33        $this->scopes[$id] = $scope;
34        $scope->deliverVia($this->fibers->runSettled(...));
35
36        return new PushSubscription($id, function (string $cancelled) use ($scope): void {
37            unset($this->targets[$cancelled], $this->scopes[$cancelled]);
38            $scope->close();
39        });
40    }
41
42    /** @param T $event */
43    public function push(mixed $event): void
44    {
45        foreach ($this->targets as $target) {
46            $this->fibers->runSettled(static function () use ($target, $event): void {
47                $target($event);
48            });
49        }
50    }
51
52    public function cancelAttachments(): void
53    {
54        $scopes = $this->scopes;
55        $this->targets = [];
56        $this->scopes = [];
57
58        foreach ($scopes as $scope) {
59            $scope->close();
60        }
61    }
62
63    public function countSubscribers(): int
64    {
65        return \count($this->targets);
66    }
67
68    public function countAttachments(): int
69    {
70        return $this->counter;
71    }
72}