batch41: implement mqtt reader/writer and subject conversion slice

2026-03-01 01:08:31 -05:00
parent 988ec0d60c
commit f2b1928a94
5 changed files with 466 additions and 0 deletions
--- a/dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/MqttHandlerTests.Impltests.cs
+++ b/dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/MqttHandlerTests.Impltests.cs
@@ -0,0 +1,82 @@
+using System.Text;
+using Shouldly;
+using ZB.MOM.NatsNet.Server.Mqtt;
+
+namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
+
+public sealed partial class MqttHandlerTests
+{
+    [Fact] // T:2170
+    public void MQTTReader_ShouldSucceed()
+    {
+        var reader = new MqttReader();
+        reader.Reset([0x82]);
+        var (initialLength, initialComplete) = reader.ReadPacketLen();
+
+        initialLength.ShouldBe(0);
+        initialComplete.ShouldBeFalse();
+
+        reader.Reset([0x01, 0xAA]);
+        var (finalLength, finalComplete) = reader.ReadPacketLenWithCheck(check: false);
+
+        finalComplete.ShouldBeTrue();
+        finalLength.ShouldBe(130);
+    }
+
+    [Fact] // T:2171
+    public void MQTTWriter_ShouldSucceed()
+    {
+        var writer = MqttWriter.NewMqttWriter(16);
+        var flags = writer.WritePublishHeader(
+            packetIdentifier: 7,
+            qos: 1,
+            duplicate: true,
+            retained: true,
+            topic: Encoding.UTF8.GetBytes("a/b"),
+            payloadLength: 5);
+
+        flags.ShouldBe((byte)(MqttPubFlag.Dup | MqttPubFlag.Retain | MqttPubFlag.QoS1));
+
+        var bytes = writer.Bytes();
+        bytes[0].ShouldBe((byte)(MqttPacket.Pub | flags));
+        bytes[1].ShouldBe((byte)12); // 2(len)+3(topic)+2(packet id)+5(payload)
+        bytes[2].ShouldBe((byte)0);
+        bytes[3].ShouldBe((byte)3);
+        Encoding.UTF8.GetString(bytes.AsSpan(4, 3)).ShouldBe("a/b");
+        bytes[7].ShouldBe((byte)0);
+        bytes[8].ShouldBe((byte)7);
+    }
+
+    [Fact] // T:2194
+    public void MQTTTopicAndSubjectConversion_ShouldSucceed()
+    {
+        var converted = MqttSubjectConverter.MqttTopicToNatsPubSubject("/foo/bar"u8);
+        Encoding.UTF8.GetString(converted).ShouldBe("/.foo.bar");
+
+        var roundTrip = MqttSubjectConverter.NatsSubjectToMqttTopic(converted);
+        Encoding.UTF8.GetString(roundTrip).ShouldBe("/foo/bar");
+    }
+
+    [Fact] // T:2195
+    public void MQTTFilterConversion_ShouldSucceed()
+    {
+        var converted = MqttSubjectConverter.MqttFilterToNatsSubject("foo/+/bar/#"u8);
+        Encoding.UTF8.GetString(converted).ShouldBe("foo.*.bar.>");
+    }
+
+    [Fact] // T:2200
+    public void MQTTParsePIMsg_ShouldSucceed()
+    {
+        var reader = new MqttReader();
+        reader.Reset([0x00, 0x00]);
+
+        reader.ReadUInt16("packet identifier").ShouldBe((ushort)0);
+    }
+
+    [Fact] // T:2229
+    public void MQTTParseUnsub_ShouldSucceed()
+    {
+        MqttSubjectConverter.MqttNeedSubForLevelUp("foo.>").ShouldBeTrue();
+        MqttSubjectConverter.MqttNeedSubForLevelUp("foo.bar").ShouldBeFalse();
+    }
+}