Lines 66.66% 10 / 15
Methods 44.44% 4 / 9
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 subscribe 100.00% 4 / 4 100.00% 1 / 1 1
 filter 100.00% 1 / 1 100.00% 1 / 1 1
 map 0.00% 0 / 1 0.00% 0 / 1 2
 debounce 0.00% 0 / 1 0.00% 0 / 1 2
 throttle 0.00% 0 / 1 0.00% 0 / 1 2
 take 0.00% 0 / 1 0.00% 0 / 1 2
 takeUntil 0.00% 0 / 1 0.00% 0 / 1 2
 waitForNext 100.00% 4 / 4 100.00% 1 / 1 2
18final readonly class SettlingCommandStream implements EventStream
19{
20    /** @param EventStream<T> $commands */
21    public function __construct(
22        private EventStream $commands,
23        private ExposedCommandSettlements $settlements,
24    ) {}
25
26    public function subscribe(Closure $handler): Subscription
27    {
28        $settlements = $this->settlements;
29
30        return $this->commands->subscribe(static function (ExposedCommand $command) use ($settlements, $handler): void {
31            $settlements->settleThroughHandler($command, $handler);
32        });
33    }
34
35    public function filter(Closure $predicate): static
36    {
37        return new self($this->commands->filter($predicate), $this->settlements);
38    }
39
40    public function map(Closure $mapper): EventStream
41    {
42        return $this->commands->map($mapper);
43    }
44
45    public function debounce(Duration $window, Edge $edge = Edge::Trailing): static
46    {
47        return new self($this->commands->debounce($window, $edge), $this->settlements);
48    }
49
50    public function throttle(Duration $window, Edge $edge = Edge::Leading): static
51    {
52        return new self($this->commands->throttle($window, $edge), $this->settlements);
53    }
54
55    public function take(int $count): static
56    {
57        return new self($this->commands->take($count), $this->settlements);
58    }
59
60    public function takeUntil(EventStream $notifier): static
61    {
62        return new self($this->commands->takeUntil($notifier), $this->settlements);
63    }
64
65    // A command handed to a waiting app counts as handled, like one whose handler returned.
66    public function waitForNext(?Duration $timeout = null): mixed
67    {
68        $command = $this->commands->waitForNext($timeout);
69
70        if ($command !== null) {
71            $this->settlements->acceptCommand($command);
72        }
73
74        return $command;
75    }
76}