Lines 100.00% 21 / 21
Methods 100.00% 12 / 12
Classes 100.00% 1 / 1
Name Lines Methods CRAP
 extendWith 100.00% 1 / 1 100.00% 1 / 1 1
 createPartitionKeyReader 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] __construct 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] subscribe 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] waitForNext 100.00% 8 / 8 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] filter 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] map 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] debounce 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] throttle 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] take 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] takeUntil 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStream] attachWithin 100.00% 3 / 3 100.00% 1 / 1 1
14final readonly class OperatorStream extends ComposedStream
15{
16    protected function extendWith(StreamSource $stage): static
17    {
18        return new static($stage, $this->timers);
19    }
20
21    protected function createPartitionKeyReader(): Closure
22    {
23        return static fn(mixed $event): string => '';
24    }
25}

Inherited from Stewart\Support\Stream\ComposedStream

25    public function __construct(
26        protected StreamSource $source,
27        protected Timers $timers,
28    ) {}
30    final public function subscribe(Closure $handler): Subscription
31    {
32        return $this->attachWithin(new SubscriptionScope(), $handler);
33    }
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    }
50    public function filter(Closure $predicate): static
51    {
52        return $this->extendWith(new FilterOperator($this->source, $predicate));
53    }
60    public function map(Closure $mapper): OperatorStream
61    {
62        return new OperatorStream(new MapOperator($this->source, $mapper), $this->timers);
63    }
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    }
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    }
75    public function take(int $count): static
76    {
77        return $this->extendWith(new TakeOperator($this->source, $count));
78    }
80    public function takeUntil(EventStream $notifier): static
81    {
82        return $this->extendWith(new TakeUntilOperator($this->source, $notifier));
83    }
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    }