Lines 40.00% 6 / 15
Methods 28.57% 2 / 7
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 watchStateChanges 75.00% 3 / 4 0.00% 0 / 1 2.06
 watchEvents 0.00% 0 / 2 0.00% 0 / 1 2
 watchTopic 0.00% 0 / 2 0.00% 0 / 1 2
 watchMqttMessages 0.00% 0 / 2 0.00% 0 / 1 2
 watchTrigger 100.00% 2 / 2 100.00% 1 / 1 1
 watchConnection 0.00% 0 / 2 0.00% 0 / 1 2
31final readonly class DispatchStreams
32{
33    public function __construct(
34        private LocalDispatcher $dispatcher,
35        private Timers&Deadlines $timers,
36        private StateCache $states,
37        private RegistryCache $registry,
38    ) {}
39
40    public function watchStateChanges(ResourceScope $scope, Selector|EntityFilter $match): StateChangeStream
41    {
42        /** @var DispatchSource<StateChange> $source */
43        $source = $match instanceof EntityFilter
44            ? new DispatchSource($this->dispatcher, $scope, SubscriptionKind::StateChange, Selector::any(), entityFilter: $match)
45            : new DispatchSource($this->dispatcher, $scope, SubscriptionKind::StateChange, $match);
46
47        return new StateChanges($source, $this->timers, new CachedStateReader($this->states, $this->registry, $match));
48    }
49
50    /** @return EventStream<HaEvent> */
51    public function watchEvents(ResourceScope $scope, EventTypeSelector $eventType): EventStream
52    {
53        /** @var DispatchSource<HaEvent> $source */
54        $source = new DispatchSource($this->dispatcher, $scope, SubscriptionKind::Event, $eventType->selector);
55
56        return new OperatorStream($source, $this->timers);
57    }
58
59    /** @return EventStream<TopicEvent> */
60    public function watchTopic(ResourceScope $scope, Selector $selector): EventStream
61    {
62        /** @var DispatchSource<TopicEvent> $source */
63        $source = new DispatchSource($this->dispatcher, $scope, SubscriptionKind::Topic, $selector);
64
65        return new OperatorStream($source, $this->timers);
66    }
67
68    /** @return EventStream<MqttMessage> */
69    public function watchMqttMessages(ResourceScope $scope, Selector $topicFilter): EventStream
70    {
71        /** @var DispatchSource<MqttMessage> $source */
72        $source = new DispatchSource($this->dispatcher, $scope, SubscriptionKind::Mqtt, $topicFilter);
73
74        return new OperatorStream($source, $this->timers);
75    }
76
77    /** @return EventStream<TriggerEvent> */
78    public function watchTrigger(ResourceScope $scope, TriggerSpec $trigger): EventStream
79    {
80        /** @var DispatchSource<TriggerEvent> $source */
81        $source = new DispatchSource($this->dispatcher, $scope, SubscriptionKind::Trigger, Selector::exact($trigger->getSharingKey()), $trigger);
82
83        return new OperatorStream($source, $this->timers);
84    }
85
86    /** @return EventStream<ConnectionEvent> */
87    public function watchConnection(ResourceScope $scope): EventStream
88    {
89        /** @var DispatchSource<ConnectionEvent> $source */
90        $source = new DispatchSource($this->dispatcher, $scope, SubscriptionKind::Connection, Selector::any());
91
92        return new OperatorStream($source, $this->timers);
93    }
94}