Lines 100.00% 13 / 13
Methods 100.00% 5 / 5
Classes 100.00% 1 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 forApp 100.00% 2 / 2 100.00% 1 / 1 1
 publish 100.00% 6 / 6 100.00% 1 / 1 2
 watchMessages 100.00% 2 / 2 100.00% 1 / 1 1
 assertBrokerMqttEnabled 100.00% 2 / 2 100.00% 1 / 1 2
22final readonly class WorkerMqtt implements Mqtt
23{
24    public function __construct(
25        private Transport $transport,
26        private DispatchStreams $streams,
27        private AppActivityCounters $activityCounters,
28        private WorkerSettings $workerSettings,
29        private ResourceScope $resourceScope,
30    ) {}
31
32    /** @throws MqttException */
33    public function forApp(AppId $appId): self
34    {
35        $this->assertBrokerMqttEnabled();
36
37        return new self($this->transport, $this->streams, $this->activityCounters, $this->workerSettings, ResourceScope::forApp($appId));
38    }
39
40    public function publish(string $topic, string|array $payload, MqttQos $qos = MqttQos::AtMostOnce, bool $retain = false): void
41    {
42        $this->assertBrokerMqttEnabled();
43        $message = MqttMessage::createForPublish($topic, $payload, $qos, $retain);
44
45        try {
46            $this->transport->send(new MqttPublish($message, $this->resourceScope));
47        } catch (TransportException $e) {
48            throw MqttException::createForReason(MqttError::PublishFailed, $e, topic: $topic);
49        }
50
51        $this->activityCounters->findOrCreateActivityForScope($this->resourceScope)->recordPublish();
52    }
53
54    public function watchMessages(string $topicFilter): EventStream
55    {
56        $this->assertBrokerMqttEnabled();
57
58        return $this->streams->watchMqttMessages($this->resourceScope, Selector::mqttFilter($topicFilter));
59    }
60
61    /** @throws MqttException */
62    private function assertBrokerMqttEnabled(): void
63    {
64        if (!$this->workerSettings->mqttEnabled) {
65            throw MqttException::createForReason(MqttError::NotConfigured);
66        }
67    }
68}