Lines 80.00% 12 / 15
Functions and Methods 70.00% 7 / 10
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
ComposedStateChangeStream 80.00% 12 / 15 70.00% 7 / 10 16.80 0.00% 0 / 1
 __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
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Support\Stream;
6
7use Closure;
8use Stewart\Contracts\EventStream;
9use Stewart\Contracts\Exception\StateError;
10use Stewart\Contracts\Exception\StateException;
11use Stewart\Contracts\State\StateChange;
12use Stewart\Contracts\StateChangeStream;
13use Stewart\Contracts\Time\Duration;
14use Stewart\Contracts\Time\Timers;
15use Stewart\Support\State\CurrentStateReader;
16
17/**
18 * @internal
19 * @extends ComposedStream<StateChange>
20 */
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}