Lines 90.90% 30 / 33
Methods 84.21% 16 / 19
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 distinctUntilChanged 100.00% 2 / 2 100.00% 1 / 1 1
 startWithCurrentState 0.00% 0 / 1 0.00% 0 / 1 2
 whenChangedTo 100.00% 1 / 1 100.00% 1 / 1 1
 whenStableFor 100.00% 1 / 1 100.00% 1 / 1 1
 whenAbove 0.00% 0 / 1 0.00% 0 / 1 2
 whenBelow 0.00% 0 / 1 0.00% 0 / 1 2
 acrossEntities 100.00% 1 / 1 100.00% 1 / 1 1
 createPartitionKeyReader 100.00% 1 / 1 100.00% 1 / 1 1
 choosePreviousStateRule 100.00% 5 / 5 100.00% 1 / 1 6
 [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
 [Stewart\Support\Stream\ComposedStream] extendWith n/a 0 / 0 n/a 0 / 0 0
21abstract readonly class ComposedStateChangeStream extends ComposedStream implements StateChangeStream
22{
23    /** @param StreamSource<StateChange> $source */
24    public function __construct(
25        StreamSource $source,
26        Timers $timers,
27        protected CurrentStateReader $currentStates,
28    ) {
29        parent::__construct($source, $timers);
30    }
31
32    public function distinctUntilChanged(): static
33    {
34        $key = static fn(StateChange $change): ?string => $change->to?->state;
35
36        return $this->extendWith(new DistinctUntilChangedOperator($this->source, $key, $this->createPartitionKeyReader()));
37    }
38
39    public function startWithCurrentState(): static
40    {
41        return $this->extendWith(new StartWithCurrentStateOperator($this->source, $this->currentStates));
42    }
43
44    public function whenChangedTo(string $state, ?Duration $for = null, string|array|null $from = null, bool $fromAnyState = false): static
45    {
46        return $this->extendWith(new WhenChangedToOperator($this->timers, $this->source, $state, $for, self::choosePreviousStateRule($from, $fromAnyState)));
47    }
48
49    public function whenStableFor(Duration $window): static
50    {
51        return $this->extendWith(new WhenStableForOperator($this->timers, $this->source, $window));
52    }
53
54    public function whenAbove(float $threshold, ?Duration $for = null, ?string $attribute = null, float $hysteresis = 0.0): static
55    {
56        return $this->extendWith(new ThresholdCrossingOperator($this->timers, $this->source, ThresholdDirection::Above, $threshold, $for, $attribute, $hysteresis));
57    }
58
59    public function whenBelow(float $threshold, ?Duration $for = null, ?string $attribute = null, float $hysteresis = 0.0): static
60    {
61        return $this->extendWith(new ThresholdCrossingOperator($this->timers, $this->source, ThresholdDirection::Below, $threshold, $for, $attribute, $hysteresis));
62    }
63
64    /** @return EventStream<StateChange> */
65    public function acrossEntities(): EventStream
66    {
67        return new OperatorStream($this->source, $this->timers);
68    }
69
70    protected function createPartitionKeyReader(): Closure
71    {
72        return static fn(StateChange $change): string => $change->entityId->value;
73    }
74
75    /**
76     * @param string|non-empty-list<string>|null $from
77     * @throws StateException
78     */
79    private static function choosePreviousStateRule(string|array|null $from, bool $fromAnyState): PreviousStateRule
80    {
81        if ($from !== null && $fromAnyState) {
82            throw StateException::createForReason(StateError::TransitionRuleConflict);
83        }
84
85        return match (true) {
86            $from !== null => PreviousStateRule::allowingOnly(...(array) $from),
87            $fromAnyState => PreviousStateRule::allowingAny(),
88            default => PreviousStateRule::excludingUnavailable(),
89        };
90    }
91}

Inherited from Stewart\Support\Stream\ComposedStream

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    }
97    abstract protected function extendWith(StreamSource $stage): static;