Lines 95.00% 19 / 20
Functions and Methods 83.33% 5 / 6
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
PushSource 95.00% 19 / 20 83.33% 5 / 6 8 0.00% 0 / 1
 __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
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Testing\Stream;
6
7use Closure;
8use Stewart\Contracts\Subscription;
9use Stewart\Support\Stream\StreamSource;
10use Stewart\Support\Stream\SubscriptionScope;
11use Stewart\Testing\Async\HandlerFibers;
12
13/**
14 * @template T
15 * @implements StreamSource<T>
16 */
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}