Lines 100.00% 48 / 48
Methods 100.00% 2 / 2
Classes 100.00% 1 / 1
Name Lines Methods CRAP
 __construct 100.00% 5 / 5 100.00% 1 / 1 4
 attach 100.00% 43 / 43 100.00% 1 / 1 12
20final readonly class ThresholdCrossingOperator implements StreamSource
21{
22    /**
23     * @param StreamSource<StateChange> $source
24     * @throws StateException|TimeException
25     */
26    public function __construct(
27        private Timers $timers,
28        private StreamSource $source,
29        private ThresholdDirection $direction,
30        private float $threshold,
31        private ?Duration $window,
32        private ?string $attribute,
33        private float $hysteresis,
34    ) {
35        $window?->requireAtLeastOneMillisecond($direction->describeUsage());
36
37        if (!is_finite($threshold)) {
38            throw StateException::thresholdNotFinite($direction->describeUsage(), $threshold);
39        }
40
41        if (is_nan($hysteresis) || $hysteresis < 0.0) {
42            throw StateException::thresholdHysteresisNegative($direction->describeUsage(), $hysteresis);
43        }
44    }
45
46    public function attach(SubscriptionScope $scope, Closure $downstream): Subscription
47    {
48        $direction = $this->direction;
49        $threshold = $this->threshold;
50        $hysteresis = $this->hysteresis;
51        $attribute = $this->attribute;
52        $latches = new CrossingLatches();
53
54        $readValue = static fn(?EntityState $state): ?float => $attribute === null
55            ? $state?->getStateAsFloat()
56            : $state?->getFloatAttribute($attribute);
57
58        $deliver = static function (StateChange $change) use ($latches, $downstream): void {
59            $latches->recordFired($change->entityId->value);
60            $downstream($change);
61        };
62
63        /** @var PendingEmissions<StateChange>|null $pendings */
64        $pendings = null;
65
66        if ($this->window !== null) {
67            $pendings = new PendingEmissions($this->timers, $this->window, $scope, $scope->claimStage(), $deliver);
68            $scope->onTeardown($pendings->disarmAll(...));
69        }
70
71        return $this->source->attach($scope, static function (StateChange $change) use ($direction, $threshold, $hysteresis, $latches, $readValue, $deliver, $pendings): void {
72            $key = $change->entityId->value;
73            $value = $readValue($change->to);
74
75            if ($value === null) {
76                $latches->forget($key);
77                $pendings?->disarm($key);
78
79                return;
80            }
81
82            if (!$latches->isKnown($key)) {
83                $previous = $readValue($change->from);
84                $primingValue = $previous ?? $value;
85
86                $direction->isBeyond($primingValue, $threshold) ? $latches->recordFired($key) : $latches->recordArmed($key);
87
88                if ($previous === null) {
89                    return;
90                }
91            }
92
93            if ($latches->isFired($key)) {
94                if ($direction->hasReturned($value, $threshold, $hysteresis)) {
95                    $latches->recordArmed($key);
96                }
97
98                return;
99            }
100
101            if (!$direction->isBeyond($value, $threshold)) {
102                $pendings?->disarm($key);
103
104                return;
105            }
106
107            if ($pendings === null) {
108                $deliver($change);
109
110                return;
111            }
112
113            if (!$pendings->isArmed($key)) {
114                $pendings->hold($key, $change);
115                $pendings->arm($key);
116            }
117        });
118    }
119}