Lines 95.45% 42 / 44
Functions and Methods 83.33% 5 / 6
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
PacketReader 95.45% 42 / 44 83.33% 5 / 6 21 0.00% 0 / 1
 readPacket 100.00% 5 / 5 100.00% 1 / 1 3
 readRemainingLength 100.00% 9 / 9 100.00% 1 / 1 3
 decodePacket 89.47% 17 / 19 0.00% 0 / 1 10.12
 decodePublish 100.00% 4 / 4 100.00% 1 / 1 2
 decodeConnAck 100.00% 2 / 2 100.00% 1 / 1 1
 decodeSubAck 100.00% 5 / 5 100.00% 1 / 1 2
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Mqtt\Packet;
6
7use Amp\ByteStream\BufferedReader;
8use Amp\Cancellation;
9use Stewart\Contracts\Mqtt\MqttMessage;
10use Stewart\Contracts\Mqtt\MqttQos;
11use Stewart\Mqtt\Exception\MqttClientException;
12use Throwable;
13
14final readonly class PacketReader
15{
16    private const int MAX_LENGTH_DIGITS = 4;
17
18    private const int LENGTH_DIGIT_BASE = 128;
19
20    private const int MAX_INBOUND_PACKET_BYTES = 16 * 1024 * 1024;
21
22    private const int TYPE_SHIFT = 4;
23
24    private const int FLAGS_MASK = 0x0F;
25
26    private const int PUBLISH_QOS_SHIFT = 1;
27
28    private const int PUBLISH_QOS_MASK = 0x03;
29
30    private const int PUBLISH_RETAIN_FLAG = 0x01;
31
32    private const int SUBSCRIPTION_FAILED = 0x80;
33
34    private const int SESSION_PRESENT_FLAG = 0x01;
35
36    /** @throws MqttClientException|Throwable */
37    public function readPacket(BufferedReader $reader, ?Cancellation $cancellation = null): InboundPacket
38    {
39        $fixedHeader = \ord($reader->readLength(1, $cancellation));
40        $remainingLength = $this->readRemainingLength($reader, $cancellation);
41
42        if ($remainingLength > self::MAX_INBOUND_PACKET_BYTES) {
43            throw MqttClientException::packetMalformed($remainingLength . '-byte');
44        }
45
46        return $this->decodePacket($fixedHeader, $remainingLength > 0 ? $reader->readLength($remainingLength, $cancellation) : '');
47    }
48
49    /** @throws MqttClientException|Throwable */
50    private function readRemainingLength(BufferedReader $reader, ?Cancellation $cancellation): int
51    {
52        $length = 0;
53        $multiplier = 1;
54
55        for ($digits = 0; $digits < self::MAX_LENGTH_DIGITS; ++$digits) {
56            $digit = \ord($reader->readLength(1, $cancellation));
57            $length += ($digit % self::LENGTH_DIGIT_BASE) * $multiplier;
58
59            if ($digit < self::LENGTH_DIGIT_BASE) {
60                return $length;
61            }
62
63            $multiplier *= self::LENGTH_DIGIT_BASE;
64        }
65
66        throw MqttClientException::packetMalformed('remaining length');
67    }
68
69    /** @throws MqttClientException */
70    private function decodePacket(int $fixedHeader, string $body): InboundPacket
71    {
72        $type = PacketType::tryFrom($fixedHeader >> self::TYPE_SHIFT)
73            ?? throw MqttClientException::packetUnexpected('type ' . ($fixedHeader >> self::TYPE_SHIFT));
74        $flags = $fixedHeader & self::FLAGS_MASK;
75        $bodyReader = new PacketBodyReader($type, $body);
76
77        if (!$type->isSentByServer()) {
78            throw MqttClientException::packetUnexpected($type->name);
79        }
80
81        if ($type === PacketType::Publish) {
82            return $this->decodePublish($flags, $bodyReader);
83        }
84
85        if ($flags !== 0) {
86            throw $bodyReader->createMalformedException();
87        }
88
89        $packet = match ($type) {
90            PacketType::ConnAck => $this->decodeConnAck($bodyReader),
91            PacketType::PubAck => new PubAckPacket($bodyReader->readUint16()),
92            PacketType::SubAck => $this->decodeSubAck($bodyReader),
93            PacketType::UnsubAck => new UnsubAckPacket($bodyReader->readUint16()),
94            PacketType::PingResp => new PingRespPacket(),
95            default => throw MqttClientException::packetUnexpected($type->name),
96        };
97
98        $bodyReader->assertFullyRead();
99
100        return $packet;
101    }
102
103    /** @throws MqttClientException */
104    private function decodePublish(int $flags, PacketBodyReader $body): PublishPacket
105    {
106        $qos = MqttQos::tryFrom(($flags >> self::PUBLISH_QOS_SHIFT) & self::PUBLISH_QOS_MASK) ?? throw $body->createMalformedException();
107        $topic = $body->readString();
108        $packetId = $qos === MqttQos::AtMostOnce ? null : $body->readUint16();
109
110        return new PublishPacket(new MqttMessage($topic, $body->readRemainder(), $qos, ($flags & self::PUBLISH_RETAIN_FLAG) !== 0), $packetId);
111    }
112
113    /** @throws MqttClientException */
114    private function decodeConnAck(PacketBodyReader $body): ConnAckPacket
115    {
116        $sessionPresent = ($body->readUint8() & self::SESSION_PRESENT_FLAG) !== 0;
117
118        return new ConnAckPacket($sessionPresent, ConnectReturnCode::tryFrom($body->readUint8()) ?? throw $body->createMalformedException());
119    }
120
121    /** @throws MqttClientException */
122    private function decodeSubAck(PacketBodyReader $body): SubAckPacket
123    {
124        $packetId = $body->readUint16();
125        $code = $body->readUint8();
126
127        if ($code === self::SUBSCRIPTION_FAILED) {
128            return new SubAckPacket($packetId, null);
129        }
130
131        return new SubAckPacket($packetId, MqttQos::tryFrom($code) ?? throw $body->createMalformedException());
132    }
133}