Lines 74.35% 58 / 78
Functions and Methods 50.00% 8 / 16
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
MqttConnection 74.35% 58 / 78 50.00% 8 / 16 62.34 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 startReading 100.00% 2 / 2 100.00% 1 / 1 1
 publishMessage 71.42% 5 / 7 0.00% 0 / 1 2.09
 subscribeFilter 100.00% 4 / 4 100.00% 1 / 1 1
 unsubscribeFilter 0.00% 0 / 4 0.00% 0 / 1 2
 disconnect 60.00% 3 / 5 0.00% 0 / 1 3.58
 awaitClosed 100.00% 1 / 1 100.00% 1 / 1 1
 readUntilClosed 100.00% 4 / 4 100.00% 1 / 1 4
 handlePacket 60.00% 3 / 5 0.00% 0 / 1 8.30
 deliverPublish 100.00% 3 / 3 100.00% 1 / 1 2
 settleSubscription 100.00% 3 / 3 100.00% 1 / 1 1
 scheduleKeepalive 84.61% 11 / 13 0.00% 0 / 1 4.06
 writePacket 28.57% 2 / 7 0.00% 0 / 1 6.28
 settleAck 33.33% 1 / 3 0.00% 0 / 1 1.30
 allocatePacketId 80.00% 4 / 5 0.00% 0 / 1 4.13
 finishClosing 100.00% 11 / 11 100.00% 1 / 1 3
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Mqtt\Connection;
6
7use Amp\ByteStream\BufferedReader;
8use Amp\DeferredFuture;
9use Amp\Future;
10use Amp\Socket\Socket;
11use Closure;
12use Stewart\Contracts\Mqtt\MqttMessage;
13use Stewart\Contracts\Mqtt\MqttQos;
14use Stewart\Contracts\Time\Duration;
15use Stewart\Contracts\Time\TimerHandle;
16use Stewart\Contracts\Time\Timers;
17use Stewart\Mqtt\Exception\MqttClientError;
18use Stewart\Mqtt\Exception\MqttClientException;
19use Stewart\Mqtt\Packet\InboundPacket;
20use Stewart\Mqtt\Packet\PacketEncoder;
21use Stewart\Mqtt\Packet\PacketReader;
22use Stewart\Mqtt\Packet\PingRespPacket;
23use Stewart\Mqtt\Packet\PubAckPacket;
24use Stewart\Mqtt\Packet\PublishPacket;
25use Stewart\Mqtt\Packet\SubAckPacket;
26use Stewart\Mqtt\Packet\UnsubAckPacket;
27use Throwable;
28
29use function Amp\async;
30
31final class MqttConnection
32{
33    private const int HIGHEST_PACKET_ID = 65535;
34
35    private const MqttQos SUBSCRIPTION_QOS = MqttQos::AtLeastOnce;
36
37    /** @var array<int, DeferredFuture<null>> */
38    private array $pendingAcks = [];
39
40    /** @var array<int, DeferredFuture<MqttQos|null>> */
41    private array $pendingSubAcks = [];
42
43    private int $lastPacketId = 0;
44
45    private bool $awaitingPingResponse = false;
46
47    private ?TimerHandle $keepaliveTimer = null;
48
49    /** @var DeferredFuture<MqttClientException|null> */
50    private readonly DeferredFuture $closed;
51
52    /** @param Closure(MqttMessage): void $onMessage */
53    public function __construct(
54        private readonly Socket $socket,
55        private readonly BufferedReader $bufferedReader,
56        private readonly PacketReader $packetReader,
57        private readonly PacketEncoder $encoder,
58        private readonly Timers $timers,
59        private readonly Duration $keepalive,
60        private readonly Closure $onMessage,
61    ) {
62        $this->closed = new DeferredFuture();
63    }
64
65    public function startReading(): void
66    {
67        async($this->readUntilClosed(...));
68        $this->scheduleKeepalive();
69    }
70
71    /**
72     * @return Future<null>
73     * @throws MqttClientException
74     */
75    public function publishMessage(MqttMessage $message): Future
76    {
77        if ($message->qos === MqttQos::AtMostOnce) {
78            $this->writePacket($this->encoder->encodePublish($message));
79
80            return Future::complete();
81        }
82
83        $packetId = $this->allocatePacketId();
84        $this->pendingAcks[$packetId] = new DeferredFuture();
85        $this->writePacket($this->encoder->encodePublish($message, $packetId));
86
87        return $this->pendingAcks[$packetId]->getFuture();
88    }
89
90    /**
91     * @return Future<MqttQos|null>
92     * @throws MqttClientException
93     */
94    public function subscribeFilter(string $topicFilter): Future
95    {
96        $packetId = $this->allocatePacketId();
97        $this->pendingSubAcks[$packetId] = new DeferredFuture();
98        $this->writePacket($this->encoder->encodeSubscribe($packetId, $topicFilter, self::SUBSCRIPTION_QOS));
99
100        return $this->pendingSubAcks[$packetId]->getFuture();
101    }
102
103    /**
104     * @return Future<null>
105     * @throws MqttClientException
106     */
107    public function unsubscribeFilter(string $topicFilter): Future
108    {
109        $packetId = $this->allocatePacketId();
110        $this->pendingAcks[$packetId] = new DeferredFuture();
111        $this->writePacket($this->encoder->encodeUnsubscribe($packetId, $topicFilter));
112
113        return $this->pendingAcks[$packetId]->getFuture();
114    }
115
116    public function disconnect(): void
117    {
118        if ($this->closed->isComplete()) {
119            return;
120        }
121
122        try {
123            $this->socket->write($this->encoder->encodeDisconnect());
124        } catch (Throwable) {
125            // The socket is closing either way; the server then treats the client as gone.
126        }
127
128        $this->finishClosing(null);
129    }
130
131    public function awaitClosed(): ?MqttClientException
132    {
133        return $this->closed->getFuture()->await();
134    }
135
136    private function readUntilClosed(): void
137    {
138        try {
139            while (!$this->closed->isComplete()) {
140                $this->handlePacket($this->packetReader->readPacket($this->bufferedReader));
141            }
142        } catch (Throwable $e) {
143            $this->finishClosing($e instanceof MqttClientException ? $e : MqttClientException::createForReason(MqttClientError::ConnectionLost, $e));
144        }
145    }
146
147    /** @throws MqttClientException */
148    private function handlePacket(InboundPacket $packet): void
149    {
150        match (true) {
151            $packet instanceof PublishPacket => $this->deliverPublish($packet),
152            $packet instanceof PubAckPacket, $packet instanceof UnsubAckPacket => $this->settleAck($packet),
153            $packet instanceof SubAckPacket => $this->settleSubscription($packet),
154            $packet instanceof PingRespPacket => $this->awaitingPingResponse = false,
155            default => throw MqttClientException::packetUnexpected($packet->getPacketType()->name),
156        };
157    }
158
159    /** @throws MqttClientException */
160    private function deliverPublish(PublishPacket $packet): void
161    {
162        if ($packet->packetId !== null) {
163            $this->writePacket($this->encoder->encodePubAck($packet->packetId));
164        }
165
166        ($this->onMessage)($packet->message);
167    }
168
169    /** @throws MqttClientException */
170    private function settleSubscription(SubAckPacket $packet): void
171    {
172        $pending = $this->pendingSubAcks[$packet->packetId] ?? throw MqttClientException::packetUnexpected($packet->getPacketType()->name);
173        unset($this->pendingSubAcks[$packet->packetId]);
174        $pending->complete($packet->grantedQos);
175    }
176
177    private function scheduleKeepalive(): void
178    {
179        $this->keepaliveTimer = $this->timers->startTimer($this->keepalive, function (): void {
180            if ($this->closed->isComplete()) {
181                return;
182            }
183
184            if ($this->awaitingPingResponse) {
185                $this->finishClosing(MqttClientException::createForReason(MqttClientError::KeepaliveTimedOut, keepalive: (string) $this->keepalive));
186
187                return;
188            }
189
190            $this->awaitingPingResponse = true;
191            async(function (): void {
192                try {
193                    $this->writePacket($this->encoder->encodePingRequest());
194                } catch (MqttClientException) {
195                }
196            });
197            $this->scheduleKeepalive();
198        });
199    }
200
201    /** @throws MqttClientException */
202    private function writePacket(string $bytes): void
203    {
204        if ($this->closed->isComplete()) {
205            throw MqttClientException::createForReason(MqttClientError::ConnectionClosed);
206        }
207
208        try {
209            $this->socket->write($bytes);
210        } catch (Throwable $e) {
211            $lost = MqttClientException::createForReason(MqttClientError::ConnectionLost, $e);
212            $this->finishClosing($lost);
213
214            throw $lost;
215        }
216    }
217
218    /** @throws MqttClientException */
219    private function settleAck(PubAckPacket|UnsubAckPacket $ack): void
220    {
221        $pending = $this->pendingAcks[$ack->packetId] ?? throw MqttClientException::packetUnexpected($ack->getPacketType()->name);
222        unset($this->pendingAcks[$ack->packetId]);
223        $pending->complete();
224    }
225
226    /** @throws MqttClientException */
227    private function allocatePacketId(): int
228    {
229        for ($attempt = 0; $attempt < self::HIGHEST_PACKET_ID; ++$attempt) {
230            $this->lastPacketId = $this->lastPacketId % self::HIGHEST_PACKET_ID + 1;
231
232            if (!isset($this->pendingAcks[$this->lastPacketId]) && !isset($this->pendingSubAcks[$this->lastPacketId])) {
233                return $this->lastPacketId;
234            }
235        }
236
237        throw MqttClientException::createForReason(MqttClientError::PacketIdsExhausted);
238    }
239
240    private function finishClosing(?MqttClientException $reason): void
241    {
242        if ($this->closed->isComplete()) {
243            return;
244        }
245
246        $this->keepaliveTimer?->cancel();
247        $this->socket->close();
248        $failure = $reason ?? MqttClientException::createForReason(MqttClientError::ConnectionClosed);
249
250        foreach ([...$this->pendingAcks, ...$this->pendingSubAcks] as $pending) {
251            $pending->error($failure);
252            $pending->getFuture()->ignore();
253        }
254
255        $this->pendingAcks = [];
256        $this->pendingSubAcks = [];
257        $this->closed->complete($reason);
258    }
259}