Lines 100.00% 19 / 19
Methods 100.00% 10 / 10
Classes 100.00% 1 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 subscribe 100.00% 1 / 1 100.00% 1 / 1 1
 waitForNext 100.00% 8 / 8 100.00% 1 / 1 1
 filter 100.00% 1 / 1 100.00% 1 / 1 1
 map 100.00% 1 / 1 100.00% 1 / 1 1
 debounce 100.00% 1 / 1 100.00% 1 / 1 1
 throttle 100.00% 1 / 1 100.00% 1 / 1 1
 take 100.00% 1 / 1 100.00% 1 / 1 1
 takeUntil 100.00% 1 / 1 100.00% 1 / 1 1
 attachWithin 100.00% 3 / 3 100.00% 1 / 1 1
 extendWith n/a 0 / 0 n/a 0 / 0 0
 createPartitionKeyReader n/a 0 / 0 n/a 0 / 0 0
22abstract readonly class ComposedStream implements EventStream
23{
24    /** @param StreamSource<T> $source */
25    public function __construct(
26        protected StreamSource $source,
27        protected Timers $timers,
28    ) {}
29
30    final public function subscribe(Closure $handler): Subscription
31    {
32        return $this->attachWithin(new SubscriptionScope(), $handler);
33    }
34
35    /** @return T|null */
36    final public function waitForNext(?Duration $timeout = null): mixed
37    {
38        return $this->timers->waitFor(function (PendingWait $wait): Closure {
39            $scope = new SubscriptionScope();
40            // Registered before attaching, so a stream that completes at once fails the wait instead of hanging it.
41            $scope->onTeardown(static function () use ($wait): void {
42                $wait->fail(WaitException::createForReason(WaitError::Abandoned));
43            });
44            $this->attachWithin($scope, $wait->resolve(...));
45
46            return $scope->close(...);
47        }, $timeout);
48    }
49
50    public function filter(Closure $predicate): static
51    {
52        return $this->extendWith(new FilterOperator($this->source, $predicate));
53    }
54
55    /**
56     * @template U
57     * @param Closure(T): U $mapper
58     * @return OperatorStream<U>
59     */
60    public function map(Closure $mapper): OperatorStream
61    {
62        return new OperatorStream(new MapOperator($this->source, $mapper), $this->timers);
63    }
64
65    public function debounce(Duration $window, Edge $edge = Edge::Trailing): static
66    {
67        return $this->extendWith(new DebounceOperator($this->timers, $this->source, $window, $edge, $this->createPartitionKeyReader()));
68    }
69
70    public function throttle(Duration $window, Edge $edge = Edge::Leading): static
71    {
72        return $this->extendWith(new ThrottleOperator($this->timers, $this->source, $window, $edge, $this->createPartitionKeyReader()));
73    }
74
75    public function take(int $count): static
76    {
77        return $this->extendWith(new TakeOperator($this->source, $count));
78    }
79
80    public function takeUntil(EventStream $notifier): static
81    {
82        return $this->extendWith(new TakeUntilOperator($this->source, $notifier));
83    }
84
85    /** @param Closure(T): void $handler */
86    private function attachWithin(SubscriptionScope $scope, Closure $handler): Subscription
87    {
88        $subscription = $this->source->attach($scope, $handler);
89        $scope->attachedTo($subscription);
90
91        return $subscription;
92    }
93
94    /**
95     * @param StreamSource<T> $stage
96     */
97    abstract protected function extendWith(StreamSource $stage): static;
98
99    /** @return Closure(T): string */
100    abstract protected function createPartitionKeyReader(): Closure;
101}