Lines 91.17% 31 / 34
Methods 85.00% 17 / 20
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 extendWith 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStateChangeStream] __construct 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStateChangeStream] distinctUntilChanged 100.00% 2 / 2 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStateChangeStream] startWithCurrentState 0.00% 0 / 1 0.00% 0 / 1 2
 [Stewart\Support\Stream\ComposedStateChangeStream] whenChangedTo 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStateChangeStream] whenStableFor 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStateChangeStream] whenAbove 0.00% 0 / 1 0.00% 0 / 1 2
 [Stewart\Support\Stream\ComposedStateChangeStream] whenBelow 0.00% 0 / 1 0.00% 0 / 1 2
 [Stewart\Support\Stream\ComposedStateChangeStream] acrossEntities 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStateChangeStream] createPartitionKeyReader 100.00% 1 / 1 100.00% 1 / 1 1
 [Stewart\Support\Stream\ComposedStateChangeStream] 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
8final readonly class StateChanges extends ComposedStateChangeStream
9{
10    protected function extendWith(StreamSource $stage): static
11    {
12        return new static($stage, $this->timers, $this->currentStates);
13    }
14}

Inherited from Stewart\Support\Stream\ComposedStateChangeStream

24    public function __construct(
25        StreamSource $source,
26        Timers $timers,
27        protected CurrentStateReader $currentStates,
28    ) {
29        parent::__construct($source, $timers);
30    }
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    }
39    public function startWithCurrentState(): static
40    {
41        return $this->extendWith(new StartWithCurrentStateOperator($this->source, $this->currentStates));
42    }
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    }
49    public function whenStableFor(Duration $window): static
50    {
51        return $this->extendWith(new WhenStableForOperator($this->timers, $this->source, $window));
52    }
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    }
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    }
65    public function acrossEntities(): EventStream
66    {
67        return new OperatorStream($this->source, $this->timers);
68    }
70    protected function createPartitionKeyReader(): Closure
71    {
72        return static fn(StateChange $change): string => $change->entityId->value;
73    }
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    }

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    }