Lines 83.54% 66 / 79
Functions and Methods 46.15% 6 / 13
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
ReconnectingMqttLink 83.54% 66 / 79 46.15% 6 / 13 40.46 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 startInBackground 75.00% 3 / 4 0.00% 0 / 1 2.06
 publishMessage 66.66% 6 / 9 0.00% 0 / 1 5.93
 subscribeFilter 100.00% 3 / 3 100.00% 1 / 1 2
 unsubscribeFilter 0.00% 0 / 3 0.00% 0 / 1 6
 close 100.00% 2 / 2 100.00% 1 / 1 1
 keepConnected 100.00% 11 / 11 100.00% 1 / 1 4
 attemptConnection 100.00% 13 / 13 100.00% 1 / 1 5
 runSession 93.75% 15 / 16 0.00% 0 / 1 3.00
 publishAtLeastOnce 80.00% 4 / 5 0.00% 0 / 1 3.07
 sendAtLeastOnce 60.00% 3 / 5 0.00% 0 / 1 2.26
 subscribeOn 60.00% 3 / 5 0.00% 0 / 1 3.58
 waitBeforeReconnecting 100.00% 2 / 2 100.00% 1 / 1 2
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Mqtt;
6
7use Amp\CancelledException;
8use Amp\DeferredCancellation;
9use Psr\Log\LoggerInterface;
10use Psr\Log\LogLevel;
11use Stewart\Contracts\Mqtt\MqttMessage;
12use Stewart\Contracts\Mqtt\MqttQos;
13use Stewart\Contracts\Time\Clock;
14use Stewart\Mqtt\Connection\MqttConnection;
15use Stewart\Mqtt\Connection\MqttConnector;
16use Stewart\Mqtt\Exception\MqttClientException;
17use Stewart\Runtime\Broker\Mqtt\MqttLink;
18use Stewart\Runtime\Broker\Mqtt\MqttMessageRouter;
19use Stewart\Runtime\Config\MqttConfig;
20use Stewart\Support\Time\Deadlines;
21use Throwable;
22
23use function Amp\async;
24
25final class ReconnectingMqttLink implements MqttLink
26{
27    /** @var array<string, true> */
28    private array $activeFilters = [];
29
30    private ?MqttConnection $connection = null;
31
32    private bool $started = false;
33
34    private readonly DeferredCancellation $stop;
35
36    public function __construct(
37        private readonly MqttConfig $config,
38        private readonly MqttConnector $connector,
39        private readonly OutboundMqttQueue $outbound,
40        private readonly Deadlines $deadlines,
41        private readonly Clock $clock,
42        private readonly LoggerInterface $logger,
43    ) {
44        $this->stop = new DeferredCancellation();
45    }
46
47    public function startInBackground(MqttMessageRouter $router): void
48    {
49        if ($this->started) {
50            return;
51        }
52
53        $this->started = true;
54        async(fn() => $this->keepConnected($router));
55    }
56
57    public function publishMessage(MqttMessage $message): void
58    {
59        if ($message->qos === MqttQos::AtLeastOnce && $this->outbound->canHoldMessages()) {
60            $this->publishAtLeastOnce($message);
61
62            return;
63        }
64
65        if ($this->connection === null) {
66            $this->logger->warning('MQTT message dropped while the server is unreachable', ['topic' => $message->topic]);
67
68            return;
69        }
70
71        try {
72            $this->connection->publishMessage($message)->ignore();
73        } catch (MqttClientException $e) {
74            $this->logger->warning('MQTT message dropped', ['topic' => $message->topic, 'exception' => $e]);
75        }
76    }
77
78    public function subscribeFilter(string $topicFilter): void
79    {
80        $this->activeFilters[$topicFilter] = true;
81
82        if ($this->connection !== null) {
83            $this->subscribeOn($this->connection, $topicFilter);
84        }
85    }
86
87    public function unsubscribeFilter(string $topicFilter): void
88    {
89        unset($this->activeFilters[$topicFilter]);
90
91        try {
92            $this->connection?->unsubscribeFilter($topicFilter)->ignore();
93        } catch (MqttClientException) {
94            // A lost connection drops every subscription with it.
95        }
96    }
97
98    public function close(): void
99    {
100        $this->stop->cancel();
101        $this->connection?->disconnect();
102    }
103
104    private function keepConnected(MqttMessageRouter $router): void
105    {
106        $failures = 0;
107
108        while (!$this->stop->getCancellation()->isRequested()) {
109            $connection = $this->attemptConnection($router, $failures + 1);
110
111            if ($connection === null) {
112                $this->waitBeforeReconnecting(++$failures);
113
114                continue;
115            }
116
117            $sessionStartedAt = $this->clock->getMonotonicTime();
118            $this->runSession($connection);
119
120            // A server that drops every session at once, over a poison message or a duplicate client_id, still backs off.
121            $sessionLength = $this->clock->getMonotonicTime()->elapsedSince($sessionStartedAt);
122            $failures = $this->config->keepalive->isLongerThan($sessionLength) ? $failures + 1 : 0;
123            $this->waitBeforeReconnecting(max($failures, 1));
124        }
125    }
126
127    private function attemptConnection(MqttMessageRouter $router, int $attempt): ?MqttConnection
128    {
129        try {
130            $connection = $this->connector->openConnection($this->config, $router->routeMqttMessage(...), $this->stop->getCancellation());
131        } catch (Throwable $e) {
132            if (!$this->stop->getCancellation()->isRequested()) {
133                $this->logger->log($attempt === 1 ? LogLevel::ERROR : LogLevel::WARNING, 'Could not reach the MQTT server', [
134                    'server' => (string) $this->config->serverUrl,
135                    'attempt' => $attempt,
136                    'exception' => $e,
137                ]);
138            }
139
140            return null;
141        }
142
143        if ($this->stop->getCancellation()->isRequested()) {
144            $connection->disconnect();
145
146            return null;
147        }
148
149        return $connection;
150    }
151
152    private function runSession(MqttConnection $connection): void
153    {
154        $this->connection = $connection;
155        $this->logger->info('Connected to the MQTT server', ['server' => (string) $this->config->serverUrl, 'filters' => \count($this->activeFilters)]);
156
157        foreach (array_keys($this->activeFilters) as $topicFilter) {
158            $this->subscribeOn($connection, (string) $topicFilter);
159        }
160
161        $this->outbound->replayPendingMessages(
162            function (int $sequence, MqttMessage $message) use ($connection): void {
163                $this->sendAtLeastOnce($connection, $sequence, $message);
164            },
165            function (MqttMessage $message, int $sendAttempts): void {
166                $this->logger->error('MQTT message dropped; the server never acknowledged it', ['topic' => $message->topic, 'attempts' => $sendAttempts]);
167            },
168        );
169
170        $lostBecause = $connection->awaitClosed();
171        $this->connection = null;
172
173        if (!$this->stop->getCancellation()->isRequested()) {
174            $this->logger->warning('Lost the MQTT connection', ['server' => (string) $this->config->serverUrl, 'exception' => $lostBecause]);
175        }
176    }
177
178    private function publishAtLeastOnce(MqttMessage $message): void
179    {
180        if ($this->outbound->isFull()) {
181            $this->logger->warning('MQTT outbound buffer is full; dropping the oldest message', ['buffered' => $this->outbound->countPending()]);
182        }
183
184        $sequence = $this->outbound->enqueueMessage($message);
185
186        if ($this->connection !== null) {
187            $this->sendAtLeastOnce($this->connection, $sequence, $message);
188        }
189    }
190
191    private function sendAtLeastOnce(MqttConnection $connection, int $sequence, MqttMessage $message): void
192    {
193        try {
194            $acknowledged = $connection->publishMessage($message);
195        } catch (MqttClientException) {
196            // Still buffered; the next connection sends it again.
197            return;
198        }
199
200        $this->outbound->recordSendAttempt($sequence);
201        $acknowledged->map(fn() => $this->outbound->acknowledgeMessage($sequence))->ignore();
202    }
203
204    private function subscribeOn(MqttConnection $connection, string $topicFilter): void
205    {
206        try {
207            $connection->subscribeFilter($topicFilter)->map(function (?MqttQos $grantedQos) use ($topicFilter): void {
208                if ($grantedQos === null) {
209                    $this->logger->error('The MQTT server refused a subscription', ['filter' => $topicFilter]);
210                }
211            })->ignore();
212        } catch (MqttClientException) {
213            // Resubscribed on the next connection.
214        }
215    }
216
217    private function waitBeforeReconnecting(int $attempt): void
218    {
219        try {
220            $this->deadlines->delay($this->config->reconnectBackoff->delayFor($attempt), $this->stop->getCancellation());
221        } catch (CancelledException) {
222            // Closing; the loop sees the stop and ends.
223        }
224    }
225}