Lines 40.00% 6 / 15
Functions and Methods 28.57% 2 / 7
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
DispatchStreams 40.00% 6 / 15 28.57% 2 / 7 21.82 0.00% 0 / 1
 __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
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Worker\Context;
6
7use Stewart\Contracts\Connection\ConnectionEvent;
8use Stewart\Contracts\Event\HaEvent;
9use Stewart\Contracts\EventStream;
10use Stewart\Contracts\Mqtt\MqttMessage;
11use Stewart\Contracts\Registry\EntityFilter;
12use Stewart\Contracts\Selector\Selector;
13use Stewart\Contracts\State\StateChange;
14use Stewart\Contracts\StateChangeStream;
15use Stewart\Contracts\Time\Timers;
16use Stewart\Contracts\Topic\TopicEvent;
17use Stewart\Contracts\Trigger\TriggerEvent;
18use Stewart\Runtime\Dispatch\DispatchSource;
19use Stewart\Runtime\Dispatch\LocalDispatcher;
20use Stewart\Runtime\Model\ResourceScope;
21use Stewart\Runtime\Model\SubscriptionKind;
22use Stewart\Runtime\Registry\RegistryCache;
23use Stewart\Runtime\State\CachedStateReader;
24use Stewart\Runtime\State\StateCache;
25use Stewart\Support\Stream\OperatorStream;
26use Stewart\Support\Stream\StateChanges;
27use Stewart\Support\Time\Deadlines;
28use Stewart\Support\Wire\EventTypeSelector;
29use Stewart\Support\Wire\TriggerSpec;
30
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}