Lines 100.00% 13 / 13
Functions and Methods 100.00% 5 / 5
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
WorkerMqtt 100.00% 13 / 13 100.00% 5 / 5 7 100.00% 1 / 1
 __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
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Worker;
6
7use Stewart\Contracts\App\AppId;
8use Stewart\Contracts\EventStream;
9use Stewart\Contracts\Exception\MqttError;
10use Stewart\Contracts\Exception\MqttException;
11use Stewart\Contracts\Mqtt\Mqtt;
12use Stewart\Contracts\Mqtt\MqttMessage;
13use Stewart\Contracts\Mqtt\MqttQos;
14use Stewart\Contracts\Selector\Selector;
15use Stewart\Runtime\Exception\TransportException;
16use Stewart\Runtime\Ipc\Message\MqttPublish;
17use Stewart\Runtime\Ipc\Transport;
18use Stewart\Runtime\Ipc\WorkerSettings;
19use Stewart\Runtime\Model\ResourceScope;
20use Stewart\Runtime\Worker\Context\DispatchStreams;
21
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}