diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs index acc763ce..f74ba975 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs @@ -4,454 +4,406 @@ using System; using System.Buffers; -using System.Collections.Generic; -using System.Linq; -using System.Text; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Formatter; using MQTTnet.Packets; using MQTTnet.Protocol; -namespace MQTTnet.Tests.Formatter +namespace MQTTnet.Tests.Formatter; + +[TestClass] +public sealed class MqttPacketSerialization_V5_Tests { - [TestClass] - public sealed class MqttPacketSerialization_V5_Tests + [TestMethod] + public void Empty_Auth_Packet_Is_Success() { - [TestMethod] - public void Serialize_Full_MqttAuthPacket_V500() + var buffer = MqttPacketSerializationHelper.EncodePacket(new MqttAuthPacket(), MqttProtocolVersion.V500); + var packet = MqttPacketSerializationHelper.DecodePacket(buffer, MqttProtocolVersion.V500); + + Assert.IsNotNull(packet); + Assert.IsInstanceOfType(packet); + } + + [TestMethod] + public void Serialize_Full_MqttAuthPacket_V500() + { + var authPacket = new MqttAuthPacket { - var authPacket = new MqttAuthPacket - { - AuthenticationData = Encoding.UTF8.GetBytes("AuthenticationData"), - AuthenticationMethod = "AuthenticationMethod", - ReasonCode = MqttAuthenticateReasonCode.ContinueAuthentication, - ReasonString = "ReasonString", - UserProperties = new List + AuthenticationData = "AuthenticationData"u8.ToArray(), + AuthenticationMethod = "AuthenticationMethod", + ReasonCode = MqttAuthenticateReasonCode.ContinueAuthentication, + ReasonString = "ReasonString", + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(authPacket, MqttProtocolVersion.V500); + + CollectionAssert.AreEqual(authPacket.AuthenticationData, deserialized.AuthenticationData); + Assert.AreEqual(authPacket.AuthenticationMethod, deserialized.AuthenticationMethod); + Assert.AreEqual(authPacket.ReasonCode, deserialized.ReasonCode); + Assert.AreEqual(authPacket.ReasonString, deserialized.ReasonString); + CollectionAssert.AreEqual(authPacket.UserProperties, deserialized.UserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttConnAckPacket_V500() + { + var connAckPacket = new MqttConnAckPacket + { + AuthenticationData = "AuthenticationData"u8.ToArray(), + AuthenticationMethod = "AuthenticationMethod", + ReasonCode = MqttConnectReasonCode.ServerUnavailable, + ReasonString = "ReasonString", + ReceiveMaximum = 123, + ResponseInformation = "ResponseInformation", + RetainAvailable = true, + ReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized, + ServerReference = "ServerReference", + AssignedClientIdentifier = "AssignedClientIdentifier", + IsSessionPresent = true, + MaximumPacketSize = 456, + MaximumQoS = MqttQualityOfServiceLevel.ExactlyOnce, + ServerKeepAlive = 789, + SessionExpiryInterval = 852, + SharedSubscriptionAvailable = true, + SubscriptionIdentifiersAvailable = true, + TopicAliasMaximum = 963, + WildcardSubscriptionAvailable = true, + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(connAckPacket, MqttProtocolVersion.V500); + + CollectionAssert.AreEqual(connAckPacket.AuthenticationData, deserialized.AuthenticationData); + Assert.AreEqual(connAckPacket.AuthenticationMethod, deserialized.AuthenticationMethod); + Assert.AreEqual(connAckPacket.ReasonCode, deserialized.ReasonCode); + Assert.AreEqual(connAckPacket.ReasonString, deserialized.ReasonString); + Assert.AreEqual(connAckPacket.ReceiveMaximum, deserialized.ReceiveMaximum); + Assert.AreEqual(connAckPacket.ResponseInformation, deserialized.ResponseInformation); + Assert.AreEqual(connAckPacket.RetainAvailable, deserialized.RetainAvailable); + // Return Code only used in MQTTv3 + Assert.AreEqual(connAckPacket.ServerReference, deserialized.ServerReference); + Assert.AreEqual(connAckPacket.AssignedClientIdentifier, deserialized.AssignedClientIdentifier); + Assert.AreEqual(connAckPacket.IsSessionPresent, deserialized.IsSessionPresent); + Assert.AreEqual(connAckPacket.MaximumPacketSize, deserialized.MaximumPacketSize); + Assert.AreEqual(connAckPacket.MaximumQoS, deserialized.MaximumQoS); + Assert.AreEqual(connAckPacket.ServerKeepAlive, deserialized.ServerKeepAlive); + Assert.AreEqual(connAckPacket.SessionExpiryInterval, deserialized.SessionExpiryInterval); + Assert.AreEqual(connAckPacket.SharedSubscriptionAvailable, deserialized.SharedSubscriptionAvailable); + Assert.AreEqual(connAckPacket.SubscriptionIdentifiersAvailable, deserialized.SubscriptionIdentifiersAvailable); + Assert.AreEqual(connAckPacket.TopicAliasMaximum, deserialized.TopicAliasMaximum); + Assert.AreEqual(connAckPacket.WildcardSubscriptionAvailable, deserialized.WildcardSubscriptionAvailable); + CollectionAssert.AreEqual(connAckPacket.UserProperties, deserialized.UserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttConnectPacket_V500() + { + var connectPacket = new MqttConnectPacket + { + Username = "Username", + Password = "Password"u8.ToArray(), + ClientId = "ClientId", + AuthenticationData = "AuthenticationData"u8.ToArray(), + AuthenticationMethod = "AuthenticationMethod", + CleanSession = true, + ReceiveMaximum = 123, + WillFlag = true, + WillTopic = "WillTopic", + WillMessage = "WillMessage"u8.ToArray(), + WillRetain = true, + KeepAlivePeriod = 456, + MaximumPacketSize = 789, + RequestProblemInformation = true, + RequestResponseInformation = true, + SessionExpiryInterval = 27, + TopicAliasMaximum = 67, + WillContentType = "WillContentType", + WillCorrelationData = "WillCorrelationData"u8.ToArray(), + WillDelayInterval = 782, + WillQoS = MqttQualityOfServiceLevel.ExactlyOnce, + WillResponseTopic = "WillResponseTopic", + WillMessageExpiryInterval = 542, + WillPayloadFormatIndicator = MqttPayloadFormatIndicator.CharacterData, + UserProperties = [new MqttUserProperty("Foo", "Bar")], + WillUserProperties = [new MqttUserProperty("WillFoo", "WillBar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(connectPacket, MqttProtocolVersion.V500); + + Assert.AreEqual(connectPacket.Username, deserialized.Username); + CollectionAssert.AreEqual(connectPacket.Password, deserialized.Password); + Assert.AreEqual(connectPacket.ClientId, deserialized.ClientId); + CollectionAssert.AreEqual(connectPacket.AuthenticationData, deserialized.AuthenticationData); + Assert.AreEqual(connectPacket.AuthenticationMethod, deserialized.AuthenticationMethod); + Assert.AreEqual(connectPacket.CleanSession, deserialized.CleanSession); + Assert.AreEqual(connectPacket.ReceiveMaximum, deserialized.ReceiveMaximum); + Assert.AreEqual(connectPacket.WillFlag, deserialized.WillFlag); + Assert.AreEqual(connectPacket.WillTopic, deserialized.WillTopic); + CollectionAssert.AreEqual(connectPacket.WillMessage, deserialized.WillMessage); + Assert.AreEqual(connectPacket.WillRetain, deserialized.WillRetain); + Assert.AreEqual(connectPacket.KeepAlivePeriod, deserialized.KeepAlivePeriod); + Assert.AreEqual(connectPacket.MaximumPacketSize, deserialized.MaximumPacketSize); + Assert.AreEqual(connectPacket.RequestProblemInformation, deserialized.RequestProblemInformation); + Assert.AreEqual(connectPacket.RequestResponseInformation, deserialized.RequestResponseInformation); + Assert.AreEqual(connectPacket.SessionExpiryInterval, deserialized.SessionExpiryInterval); + Assert.AreEqual(connectPacket.TopicAliasMaximum, deserialized.TopicAliasMaximum); + Assert.AreEqual(connectPacket.WillContentType, deserialized.WillContentType); + CollectionAssert.AreEqual(connectPacket.WillCorrelationData, deserialized.WillCorrelationData); + Assert.AreEqual(connectPacket.WillDelayInterval, deserialized.WillDelayInterval); + Assert.AreEqual(connectPacket.WillQoS, deserialized.WillQoS); + Assert.AreEqual(connectPacket.WillResponseTopic, deserialized.WillResponseTopic); + Assert.AreEqual(connectPacket.WillMessageExpiryInterval, deserialized.WillMessageExpiryInterval); + Assert.AreEqual(connectPacket.WillPayloadFormatIndicator, deserialized.WillPayloadFormatIndicator); + CollectionAssert.AreEqual(connectPacket.UserProperties, deserialized.UserProperties); + CollectionAssert.AreEqual(connectPacket.WillUserProperties, deserialized.WillUserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttDisconnectPacket_V500() + { + var disconnectPacket = new MqttDisconnectPacket + { + ReasonCode = MqttDisconnectReasonCode.QuotaExceeded, + ReasonString = "ReasonString", + ServerReference = "ServerReference", + SessionExpiryInterval = 234, + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(disconnectPacket, MqttProtocolVersion.V500); + + Assert.AreEqual(disconnectPacket.ReasonCode, deserialized.ReasonCode); + Assert.AreEqual(disconnectPacket.ReasonString, deserialized.ReasonString); + Assert.AreEqual(disconnectPacket.ServerReference, deserialized.ServerReference); + Assert.AreEqual(disconnectPacket.SessionExpiryInterval, deserialized.SessionExpiryInterval); + CollectionAssert.AreEqual(disconnectPacket.UserProperties, deserialized.UserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttPingReqPacket_V500() + { + var pingReqPacket = new MqttPingReqPacket(); + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pingReqPacket, MqttProtocolVersion.V500); + + Assert.IsNotNull(deserialized); + } + + [TestMethod] + public void Serialize_Full_MqttPingRespPacket_V500() + { + var pingRespPacket = new MqttPingRespPacket(); + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pingRespPacket, MqttProtocolVersion.V500); + + Assert.IsNotNull(deserialized); + } + + [TestMethod] + public void Serialize_Full_MqttPubAckPacket_V500() + { + var pubAckPacket = new MqttPubAckPacket + { + PacketIdentifier = 123, + ReasonCode = MqttPubAckReasonCode.NoMatchingSubscribers, + ReasonString = "ReasonString", + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pubAckPacket, MqttProtocolVersion.V500); + + Assert.AreEqual(pubAckPacket.PacketIdentifier, deserialized.PacketIdentifier); + Assert.AreEqual(pubAckPacket.ReasonCode, deserialized.ReasonCode); + Assert.AreEqual(pubAckPacket.ReasonString, deserialized.ReasonString); + CollectionAssert.AreEqual(pubAckPacket.UserProperties, deserialized.UserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttPubCompPacket_V500() + { + var pubCompPacket = new MqttPubCompPacket + { + PacketIdentifier = 123, + ReasonCode = MqttPubCompReasonCode.PacketIdentifierNotFound, + ReasonString = "ReasonString", + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pubCompPacket, MqttProtocolVersion.V500); + + Assert.AreEqual(pubCompPacket.PacketIdentifier, deserialized.PacketIdentifier); + Assert.AreEqual(pubCompPacket.ReasonCode, deserialized.ReasonCode); + Assert.AreEqual(pubCompPacket.ReasonString, deserialized.ReasonString); + CollectionAssert.AreEqual(pubCompPacket.UserProperties, deserialized.UserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttPublishPacket_V500() + { + var publishPacket = new MqttPublishPacket + { + PacketIdentifier = 123, + Dup = true, + Retain = true, + PayloadSegment = new ArraySegment("Payload"u8.ToArray()), + QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, + Topic = "Topic", + ResponseTopic = "/Response", + ContentType = "Content-Type", + CorrelationData = "CorrelationData"u8.ToArray(), + TopicAlias = 27, + SubscriptionIdentifiers = [123], + MessageExpiryInterval = 38, + PayloadFormatIndicator = MqttPayloadFormatIndicator.CharacterData, + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(publishPacket, MqttProtocolVersion.V500); + + Assert.AreEqual(publishPacket.PacketIdentifier, deserialized.PacketIdentifier); + Assert.AreEqual(publishPacket.Dup, deserialized.Dup); + Assert.AreEqual(publishPacket.Retain, deserialized.Retain); + CollectionAssert.AreEqual(publishPacket.Payload.ToArray(), deserialized.Payload.ToArray()); + Assert.AreEqual(publishPacket.QualityOfServiceLevel, deserialized.QualityOfServiceLevel); + Assert.AreEqual(publishPacket.Topic, deserialized.Topic); + Assert.AreEqual(publishPacket.ResponseTopic, deserialized.ResponseTopic); + Assert.AreEqual(publishPacket.ContentType, deserialized.ContentType); + CollectionAssert.AreEqual(publishPacket.CorrelationData, deserialized.CorrelationData); + Assert.AreEqual(publishPacket.TopicAlias, deserialized.TopicAlias); + CollectionAssert.AreEqual(publishPacket.SubscriptionIdentifiers, deserialized.SubscriptionIdentifiers); + Assert.AreEqual(publishPacket.MessageExpiryInterval, deserialized.MessageExpiryInterval); + Assert.AreEqual(publishPacket.PayloadFormatIndicator, deserialized.PayloadFormatIndicator); + CollectionAssert.AreEqual(publishPacket.UserProperties, deserialized.UserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttPubRecPacket_V500() + { + var pubRecPacket = new MqttPubRecPacket + { + PacketIdentifier = 123, + ReasonCode = MqttPubRecReasonCode.UnspecifiedError, + ReasonString = "ReasonString", + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pubRecPacket, MqttProtocolVersion.V500); + + Assert.AreEqual(pubRecPacket.PacketIdentifier, deserialized.PacketIdentifier); + Assert.AreEqual(pubRecPacket.ReasonCode, deserialized.ReasonCode); + Assert.AreEqual(pubRecPacket.ReasonString, deserialized.ReasonString); + CollectionAssert.AreEqual(pubRecPacket.UserProperties, deserialized.UserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttPubRelPacket_V500() + { + var pubRelPacket = new MqttPubRelPacket + { + PacketIdentifier = 123, + ReasonCode = MqttPubRelReasonCode.PacketIdentifierNotFound, + ReasonString = "ReasonString", + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pubRelPacket, MqttProtocolVersion.V500); + + Assert.AreEqual(pubRelPacket.PacketIdentifier, deserialized.PacketIdentifier); + Assert.AreEqual(pubRelPacket.ReasonCode, deserialized.ReasonCode); + Assert.AreEqual(pubRelPacket.ReasonString, deserialized.ReasonString); + CollectionAssert.AreEqual(pubRelPacket.UserProperties, deserialized.UserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttSubAckPacket_V500() + { + var subAckPacket = new MqttSubAckPacket + { + PacketIdentifier = 123, + ReasonString = "ReasonString", + ReasonCodes = [MqttSubscribeReasonCode.GrantedQoS1], + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; + + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(subAckPacket, MqttProtocolVersion.V500); + + Assert.AreEqual(subAckPacket.PacketIdentifier, deserialized.PacketIdentifier); + Assert.AreEqual(subAckPacket.ReasonString, deserialized.ReasonString); + Assert.AreEqual(subAckPacket.ReasonCodes.Count, deserialized.ReasonCodes.Count); + Assert.AreEqual(subAckPacket.ReasonCodes[0], deserialized.ReasonCodes[0]); + CollectionAssert.AreEqual(subAckPacket.UserProperties, deserialized.UserProperties); + } + + [TestMethod] + public void Serialize_Full_MqttSubscribePacket_V500() + { + var subscribePacket = new MqttSubscribePacket + { + PacketIdentifier = 123, + SubscriptionIdentifier = 456, + TopicFilters = + [ + new MqttTopicFilter { - new MqttUserProperty("Foo", "Bar") + Topic = "Topic", + NoLocal = true, + RetainHandling = MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly, + RetainAsPublished = true, + QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce } - }; + ], + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(authPacket, MqttProtocolVersion.V500); + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(subscribePacket, MqttProtocolVersion.V500); - CollectionAssert.AreEqual(authPacket.AuthenticationData, deserialized.AuthenticationData); - Assert.AreEqual(authPacket.AuthenticationMethod, deserialized.AuthenticationMethod); - Assert.AreEqual(authPacket.ReasonCode, deserialized.ReasonCode); - Assert.AreEqual(authPacket.ReasonString, deserialized.ReasonString); - CollectionAssert.AreEqual(authPacket.UserProperties, deserialized.UserProperties); - } + Assert.AreEqual(subscribePacket.PacketIdentifier, deserialized.PacketIdentifier); + Assert.AreEqual(subscribePacket.SubscriptionIdentifier, deserialized.SubscriptionIdentifier); + Assert.AreEqual(1, deserialized.TopicFilters.Count); + Assert.AreEqual(subscribePacket.TopicFilters[0].Topic, deserialized.TopicFilters[0].Topic); + Assert.AreEqual(subscribePacket.TopicFilters[0].NoLocal, deserialized.TopicFilters[0].NoLocal); + Assert.AreEqual(subscribePacket.TopicFilters[0].RetainHandling, deserialized.TopicFilters[0].RetainHandling); + Assert.AreEqual(subscribePacket.TopicFilters[0].RetainAsPublished, deserialized.TopicFilters[0].RetainAsPublished); + Assert.AreEqual(subscribePacket.TopicFilters[0].QualityOfServiceLevel, deserialized.TopicFilters[0].QualityOfServiceLevel); + CollectionAssert.AreEqual(subscribePacket.UserProperties, deserialized.UserProperties); + } - [TestMethod] - public void Serialize_Full_MqttConnAckPacket_V500() + [TestMethod] + public void Serialize_Full_MqttUnsubAckPacket_V500() + { + var unsubAckPacket = new MqttUnsubAckPacket { - var connAckPacket = new MqttConnAckPacket - { - AuthenticationData = Encoding.UTF8.GetBytes("AuthenticationData"), - AuthenticationMethod = "AuthenticationMethod", - ReasonCode = MqttConnectReasonCode.ServerUnavailable, - ReasonString = "ReasonString", - ReceiveMaximum = 123, - ResponseInformation = "ResponseInformation", - RetainAvailable = true, - ReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized, - ServerReference = "ServerReference", - AssignedClientIdentifier = "AssignedClientIdentifier", - IsSessionPresent = true, - MaximumPacketSize = 456, - MaximumQoS = MqttQualityOfServiceLevel.ExactlyOnce, - ServerKeepAlive = 789, - SessionExpiryInterval = 852, - SharedSubscriptionAvailable = true, - SubscriptionIdentifiersAvailable = true, - TopicAliasMaximum = 963, - WildcardSubscriptionAvailable = true, - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; + PacketIdentifier = 123, + ReasonCodes = [MqttUnsubscribeReasonCode.ImplementationSpecificError], + ReasonString = "ReasonString", + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(connAckPacket, MqttProtocolVersion.V500); + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(unsubAckPacket, MqttProtocolVersion.V500); - CollectionAssert.AreEqual(connAckPacket.AuthenticationData, deserialized.AuthenticationData); - Assert.AreEqual(connAckPacket.AuthenticationMethod, deserialized.AuthenticationMethod); - Assert.AreEqual(connAckPacket.ReasonCode, deserialized.ReasonCode); - Assert.AreEqual(connAckPacket.ReasonString, deserialized.ReasonString); - Assert.AreEqual(connAckPacket.ReceiveMaximum, deserialized.ReceiveMaximum); - Assert.AreEqual(connAckPacket.ResponseInformation, deserialized.ResponseInformation); - Assert.AreEqual(connAckPacket.RetainAvailable, deserialized.RetainAvailable); - // Return Code only used in MQTTv3 - Assert.AreEqual(connAckPacket.ServerReference, deserialized.ServerReference); - Assert.AreEqual(connAckPacket.AssignedClientIdentifier, deserialized.AssignedClientIdentifier); - Assert.AreEqual(connAckPacket.IsSessionPresent, deserialized.IsSessionPresent); - Assert.AreEqual(connAckPacket.MaximumPacketSize, deserialized.MaximumPacketSize); - Assert.AreEqual(connAckPacket.MaximumQoS, deserialized.MaximumQoS); - Assert.AreEqual(connAckPacket.ServerKeepAlive, deserialized.ServerKeepAlive); - Assert.AreEqual(connAckPacket.SessionExpiryInterval, deserialized.SessionExpiryInterval); - Assert.AreEqual(connAckPacket.SharedSubscriptionAvailable, deserialized.SharedSubscriptionAvailable); - Assert.AreEqual(connAckPacket.SubscriptionIdentifiersAvailable, deserialized.SubscriptionIdentifiersAvailable); - Assert.AreEqual(connAckPacket.TopicAliasMaximum, deserialized.TopicAliasMaximum); - Assert.AreEqual(connAckPacket.WildcardSubscriptionAvailable, deserialized.WildcardSubscriptionAvailable); - CollectionAssert.AreEqual(connAckPacket.UserProperties, deserialized.UserProperties); - } + Assert.AreEqual(unsubAckPacket.PacketIdentifier, deserialized.PacketIdentifier); + Assert.AreEqual(unsubAckPacket.ReasonString, deserialized.ReasonString); + Assert.AreEqual(unsubAckPacket.ReasonCodes.Count, deserialized.ReasonCodes.Count); + Assert.AreEqual(unsubAckPacket.ReasonCodes[0], deserialized.ReasonCodes[0]); + CollectionAssert.AreEqual(unsubAckPacket.UserProperties, deserialized.UserProperties); + } - [TestMethod] - public void Serialize_Full_MqttConnectPacket_V500() + [TestMethod] + public void Serialize_Full_MqttUnsubscribePacket_V500() + { + var unsubscribePacket = new MqttUnsubscribePacket { - var connectPacket = new MqttConnectPacket - { - Username = "Username", - Password = Encoding.UTF8.GetBytes("Password"), - ClientId = "ClientId", - AuthenticationData = Encoding.UTF8.GetBytes("AuthenticationData"), - AuthenticationMethod = "AuthenticationMethod", - CleanSession = true, - ReceiveMaximum = 123, - WillFlag = true, - WillTopic = "WillTopic", - WillMessage = Encoding.UTF8.GetBytes("WillMessage"), - WillRetain = true, - KeepAlivePeriod = 456, - MaximumPacketSize = 789, - RequestProblemInformation = true, - RequestResponseInformation = true, - SessionExpiryInterval = 27, - TopicAliasMaximum = 67, - WillContentType = "WillContentType", - WillCorrelationData = Encoding.UTF8.GetBytes("WillCorrelationData"), - WillDelayInterval = 782, - WillQoS = MqttQualityOfServiceLevel.ExactlyOnce, - WillResponseTopic = "WillResponseTopic", - WillMessageExpiryInterval = 542, - WillPayloadFormatIndicator = MqttPayloadFormatIndicator.CharacterData, - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - }, - WillUserProperties = new List - { - new MqttUserProperty("WillFoo", "WillBar") - } - }; + PacketIdentifier = 123, + TopicFilters = ["TopicFilter1"], + UserProperties = [new MqttUserProperty("Foo", "Bar")] + }; - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(connectPacket, MqttProtocolVersion.V500); + var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(unsubscribePacket, MqttProtocolVersion.V500); - Assert.AreEqual(connectPacket.Username, deserialized.Username); - CollectionAssert.AreEqual(connectPacket.Password, deserialized.Password); - Assert.AreEqual(connectPacket.ClientId, deserialized.ClientId); - CollectionAssert.AreEqual(connectPacket.AuthenticationData, deserialized.AuthenticationData); - Assert.AreEqual(connectPacket.AuthenticationMethod, deserialized.AuthenticationMethod); - Assert.AreEqual(connectPacket.CleanSession, deserialized.CleanSession); - Assert.AreEqual(connectPacket.ReceiveMaximum, deserialized.ReceiveMaximum); - Assert.AreEqual(connectPacket.WillFlag, deserialized.WillFlag); - Assert.AreEqual(connectPacket.WillTopic, deserialized.WillTopic); - CollectionAssert.AreEqual(connectPacket.WillMessage, deserialized.WillMessage); - Assert.AreEqual(connectPacket.WillRetain, deserialized.WillRetain); - Assert.AreEqual(connectPacket.KeepAlivePeriod, deserialized.KeepAlivePeriod); - Assert.AreEqual(connectPacket.MaximumPacketSize, deserialized.MaximumPacketSize); - Assert.AreEqual(connectPacket.RequestProblemInformation, deserialized.RequestProblemInformation); - Assert.AreEqual(connectPacket.RequestResponseInformation, deserialized.RequestResponseInformation); - Assert.AreEqual(connectPacket.SessionExpiryInterval, deserialized.SessionExpiryInterval); - Assert.AreEqual(connectPacket.TopicAliasMaximum, deserialized.TopicAliasMaximum); - Assert.AreEqual(connectPacket.WillContentType, deserialized.WillContentType); - CollectionAssert.AreEqual(connectPacket.WillCorrelationData, deserialized.WillCorrelationData); - Assert.AreEqual(connectPacket.WillDelayInterval, deserialized.WillDelayInterval); - Assert.AreEqual(connectPacket.WillQoS, deserialized.WillQoS); - Assert.AreEqual(connectPacket.WillResponseTopic, deserialized.WillResponseTopic); - Assert.AreEqual(connectPacket.WillMessageExpiryInterval, deserialized.WillMessageExpiryInterval); - Assert.AreEqual(connectPacket.WillPayloadFormatIndicator, deserialized.WillPayloadFormatIndicator); - CollectionAssert.AreEqual(connectPacket.UserProperties, deserialized.UserProperties); - CollectionAssert.AreEqual(connectPacket.WillUserProperties, deserialized.WillUserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttDisconnectPacket_V500() - { - var disconnectPacket = new MqttDisconnectPacket - { - ReasonCode = MqttDisconnectReasonCode.QuotaExceeded, - ReasonString = "ReasonString", - ServerReference = "ServerReference", - SessionExpiryInterval = 234, - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(disconnectPacket, MqttProtocolVersion.V500); - - Assert.AreEqual(disconnectPacket.ReasonCode, deserialized.ReasonCode); - Assert.AreEqual(disconnectPacket.ReasonString, deserialized.ReasonString); - Assert.AreEqual(disconnectPacket.ServerReference, deserialized.ServerReference); - Assert.AreEqual(disconnectPacket.SessionExpiryInterval, deserialized.SessionExpiryInterval); - CollectionAssert.AreEqual(disconnectPacket.UserProperties, deserialized.UserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttPingReqPacket_V500() - { - var pingReqPacket = new MqttPingReqPacket(); - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pingReqPacket, MqttProtocolVersion.V500); - - Assert.IsNotNull(deserialized); - } - - [TestMethod] - public void Serialize_Full_MqttPingRespPacket_V500() - { - var pingRespPacket = new MqttPingRespPacket(); - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pingRespPacket, MqttProtocolVersion.V500); - - Assert.IsNotNull(deserialized); - } - - [TestMethod] - public void Serialize_Full_MqttPubAckPacket_V500() - { - var pubAckPacket = new MqttPubAckPacket - { - PacketIdentifier = 123, - ReasonCode = MqttPubAckReasonCode.NoMatchingSubscribers, - ReasonString = "ReasonString", - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pubAckPacket, MqttProtocolVersion.V500); - - Assert.AreEqual(pubAckPacket.PacketIdentifier, deserialized.PacketIdentifier); - Assert.AreEqual(pubAckPacket.ReasonCode, deserialized.ReasonCode); - Assert.AreEqual(pubAckPacket.ReasonString, deserialized.ReasonString); - CollectionAssert.AreEqual(pubAckPacket.UserProperties, deserialized.UserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttPubCompPacket_V500() - { - var pubCompPacket = new MqttPubCompPacket - { - PacketIdentifier = 123, - ReasonCode = MqttPubCompReasonCode.PacketIdentifierNotFound, - ReasonString = "ReasonString", - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pubCompPacket, MqttProtocolVersion.V500); - - Assert.AreEqual(pubCompPacket.PacketIdentifier, deserialized.PacketIdentifier); - Assert.AreEqual(pubCompPacket.ReasonCode, deserialized.ReasonCode); - Assert.AreEqual(pubCompPacket.ReasonString, deserialized.ReasonString); - CollectionAssert.AreEqual(pubCompPacket.UserProperties, deserialized.UserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttPublishPacket_V500() - { - var publishPacket = new MqttPublishPacket - { - PacketIdentifier = 123, - Dup = true, - Retain = true, - PayloadSegment = new ArraySegment(Encoding.ASCII.GetBytes("Payload")), - QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, - Topic = "Topic", - ResponseTopic = "/Response", - ContentType = "Content-Type", - CorrelationData = Encoding.UTF8.GetBytes("CorrelationData"), - TopicAlias = 27, - SubscriptionIdentifiers = new List - { - 123 - }, - MessageExpiryInterval = 38, - PayloadFormatIndicator = MqttPayloadFormatIndicator.CharacterData, - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(publishPacket, MqttProtocolVersion.V500); - - Assert.AreEqual(publishPacket.PacketIdentifier, deserialized.PacketIdentifier); - Assert.AreEqual(publishPacket.Dup, deserialized.Dup); - Assert.AreEqual(publishPacket.Retain, deserialized.Retain); - CollectionAssert.AreEqual(publishPacket.Payload.ToArray(), deserialized.Payload.ToArray()); - Assert.AreEqual(publishPacket.QualityOfServiceLevel, deserialized.QualityOfServiceLevel); - Assert.AreEqual(publishPacket.Topic, deserialized.Topic); - Assert.AreEqual(publishPacket.ResponseTopic, deserialized.ResponseTopic); - Assert.AreEqual(publishPacket.ContentType, deserialized.ContentType); - CollectionAssert.AreEqual(publishPacket.CorrelationData, deserialized.CorrelationData); - Assert.AreEqual(publishPacket.TopicAlias, deserialized.TopicAlias); - CollectionAssert.AreEqual(publishPacket.SubscriptionIdentifiers, deserialized.SubscriptionIdentifiers); - Assert.AreEqual(publishPacket.MessageExpiryInterval, deserialized.MessageExpiryInterval); - Assert.AreEqual(publishPacket.PayloadFormatIndicator, deserialized.PayloadFormatIndicator); - CollectionAssert.AreEqual(publishPacket.UserProperties, deserialized.UserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttPubRecPacket_V500() - { - var pubRecPacket = new MqttPubRecPacket - { - PacketIdentifier = 123, - ReasonCode = MqttPubRecReasonCode.UnspecifiedError, - ReasonString = "ReasonString", - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pubRecPacket, MqttProtocolVersion.V500); - - Assert.AreEqual(pubRecPacket.PacketIdentifier, deserialized.PacketIdentifier); - Assert.AreEqual(pubRecPacket.ReasonCode, deserialized.ReasonCode); - Assert.AreEqual(pubRecPacket.ReasonString, deserialized.ReasonString); - CollectionAssert.AreEqual(pubRecPacket.UserProperties, deserialized.UserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttPubRelPacket_V500() - { - var pubRelPacket = new MqttPubRelPacket - { - PacketIdentifier = 123, - ReasonCode = MqttPubRelReasonCode.PacketIdentifierNotFound, - ReasonString = "ReasonString", - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(pubRelPacket, MqttProtocolVersion.V500); - - Assert.AreEqual(pubRelPacket.PacketIdentifier, deserialized.PacketIdentifier); - Assert.AreEqual(pubRelPacket.ReasonCode, deserialized.ReasonCode); - Assert.AreEqual(pubRelPacket.ReasonString, deserialized.ReasonString); - CollectionAssert.AreEqual(pubRelPacket.UserProperties, deserialized.UserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttSubAckPacket_V500() - { - var subAckPacket = new MqttSubAckPacket - { - PacketIdentifier = 123, - ReasonString = "ReasonString", - ReasonCodes = new List - { - MqttSubscribeReasonCode.GrantedQoS1 - }, - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(subAckPacket, MqttProtocolVersion.V500); - - Assert.AreEqual(subAckPacket.PacketIdentifier, deserialized.PacketIdentifier); - Assert.AreEqual(subAckPacket.ReasonString, deserialized.ReasonString); - Assert.AreEqual(subAckPacket.ReasonCodes.Count, deserialized.ReasonCodes.Count); - Assert.AreEqual(subAckPacket.ReasonCodes[0], deserialized.ReasonCodes[0]); - CollectionAssert.AreEqual(subAckPacket.UserProperties, deserialized.UserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttSubscribePacket_V500() - { - var subscribePacket = new MqttSubscribePacket - { - PacketIdentifier = 123, - SubscriptionIdentifier = 456, - TopicFilters = new List - { - new MqttTopicFilter - { - Topic = "Topic", - NoLocal = true, - RetainHandling = MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly, - RetainAsPublished = true, - QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce - } - }, - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(subscribePacket, MqttProtocolVersion.V500); - - Assert.AreEqual(subscribePacket.PacketIdentifier, deserialized.PacketIdentifier); - Assert.AreEqual(subscribePacket.SubscriptionIdentifier, deserialized.SubscriptionIdentifier); - Assert.AreEqual(1, deserialized.TopicFilters.Count); - Assert.AreEqual(subscribePacket.TopicFilters[0].Topic, deserialized.TopicFilters[0].Topic); - Assert.AreEqual(subscribePacket.TopicFilters[0].NoLocal, deserialized.TopicFilters[0].NoLocal); - Assert.AreEqual(subscribePacket.TopicFilters[0].RetainHandling, deserialized.TopicFilters[0].RetainHandling); - Assert.AreEqual(subscribePacket.TopicFilters[0].RetainAsPublished, deserialized.TopicFilters[0].RetainAsPublished); - Assert.AreEqual(subscribePacket.TopicFilters[0].QualityOfServiceLevel, deserialized.TopicFilters[0].QualityOfServiceLevel); - CollectionAssert.AreEqual(subscribePacket.UserProperties, deserialized.UserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttUnsubAckPacket_V500() - { - var unsubAckPacket = new MqttUnsubAckPacket - { - PacketIdentifier = 123, - ReasonCodes = new List - { - MqttUnsubscribeReasonCode.ImplementationSpecificError - }, - ReasonString = "ReasonString", - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(unsubAckPacket, MqttProtocolVersion.V500); - - Assert.AreEqual(unsubAckPacket.PacketIdentifier, deserialized.PacketIdentifier); - Assert.AreEqual(unsubAckPacket.ReasonString, deserialized.ReasonString); - Assert.AreEqual(unsubAckPacket.ReasonCodes.Count, deserialized.ReasonCodes.Count); - Assert.AreEqual(unsubAckPacket.ReasonCodes[0], deserialized.ReasonCodes[0]); - CollectionAssert.AreEqual(unsubAckPacket.UserProperties, deserialized.UserProperties); - } - - [TestMethod] - public void Serialize_Full_MqttUnsubscribePacket_V500() - { - var unsubscribePacket = new MqttUnsubscribePacket - { - PacketIdentifier = 123, - TopicFilters = new List - { - "TopicFilter1" - }, - UserProperties = new List - { - new MqttUserProperty("Foo", "Bar") - } - }; - - var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(unsubscribePacket, MqttProtocolVersion.V500); - - Assert.AreEqual(unsubscribePacket.PacketIdentifier, deserialized.PacketIdentifier); - Assert.AreEqual(unsubscribePacket.TopicFilters.Count, deserialized.TopicFilters.Count); - Assert.AreEqual(unsubscribePacket.TopicFilters[0], deserialized.TopicFilters[0]); - CollectionAssert.AreEqual(unsubscribePacket.UserProperties, deserialized.UserProperties); - } + Assert.AreEqual(unsubscribePacket.PacketIdentifier, deserialized.PacketIdentifier); + Assert.AreEqual(unsubscribePacket.TopicFilters.Count, deserialized.TopicFilters.Count); + Assert.AreEqual(unsubscribePacket.TopicFilters[0], deserialized.TopicFilters[0]); + CollectionAssert.AreEqual(unsubscribePacket.UserProperties, deserialized.UserProperties); } } \ No newline at end of file diff --git a/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs b/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs index 71f190ed..048550e8 100644 --- a/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs +++ b/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs @@ -6,60 +6,59 @@ using MQTTnet.Formatter; using MQTTnet.Packets; using MQTTnet.Tests.Mockups; -namespace MQTTnet.Tests +namespace MQTTnet.Tests; + +public sealed class MqttPacketSerializationHelper : IDisposable { - public sealed class MqttPacketSerializationHelper : IDisposable + readonly IMqttPacketFormatter _packetFormatter; + readonly MqttProtocolVersion _protocolVersion; + + public MqttPacketSerializationHelper(MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311, MqttBufferWriter bufferWriter = null) { - readonly IMqttPacketFormatter _packetFormatter; - readonly MqttProtocolVersion _protocolVersion; + _protocolVersion = protocolVersion; - public MqttPacketSerializationHelper(MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311, MqttBufferWriter bufferWriter = null) + if (bufferWriter == null) { - _protocolVersion = protocolVersion; - - if (bufferWriter == null) - { - bufferWriter = new MqttBufferWriter(4096, 65535); - } - - _packetFormatter = MqttPacketFormatterAdapter.GetMqttPacketFormatter(_protocolVersion, bufferWriter); + bufferWriter = new MqttBufferWriter(4096, 65535); } - public MqttPacket Decode(MqttPacketBuffer buffer) - { - using (var channel = new MemoryMqttChannel(buffer.ToArray())) - { - var formatterAdapter = new MqttPacketFormatterAdapter(_protocolVersion, new MqttBufferWriter(4096, 65535)); + _packetFormatter = MqttPacketFormatterAdapter.GetMqttPacketFormatter(_protocolVersion, bufferWriter); + } - var adapter = new MqttChannelAdapter(channel, formatterAdapter, MqttNetNullLogger.Instance); - return adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult(); - } - } + public MqttPacket Decode(MqttPacketBuffer buffer) + { + using var channel = new MemoryMqttChannel(buffer.ToArray()); + var formatterAdapter = new MqttPacketFormatterAdapter(_protocolVersion, new MqttBufferWriter(4096, 65535)); - public void Dispose() - { - } + var adapter = new MqttChannelAdapter(channel, formatterAdapter, MqttNetNullLogger.Instance); + return adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult(); + } - public MqttPacketBuffer Encode(MqttPacket packet) - { - return _packetFormatter.Encode(packet); - } + public static MqttPacket DecodePacket(byte[] buffer, MqttProtocolVersion protocolVersion) + { + using var helper = new MqttPacketSerializationHelper(protocolVersion); + return helper.Decode(new MqttPacketBuffer(buffer)); + } - public static TPacket EncodeAndDecodePacket(TPacket packet, MqttProtocolVersion protocolVersion) where TPacket : MqttPacket - { - using (var helper = new MqttPacketSerializationHelper(protocolVersion)) - { - var buffer = helper.Encode(packet); - return (TPacket)helper.Decode(buffer); - } - } + public void Dispose() + { + } - public static byte[] EncodePacket(MqttPacket packet) - { - using (var helper = new MqttPacketSerializationHelper()) - { - return helper.Encode(packet).ToArray(); - } - } + public MqttPacketBuffer Encode(MqttPacket packet) + { + return _packetFormatter.Encode(packet); + } + + public static TPacket EncodeAndDecodePacket(TPacket packet, MqttProtocolVersion protocolVersion) where TPacket : MqttPacket + { + using var helper = new MqttPacketSerializationHelper(protocolVersion); + var buffer = helper.Encode(packet); + return (TPacket)helper.Decode(buffer); + } + + public static byte[] EncodePacket(MqttPacket packet, MqttProtocolVersion protocolVersion) + { + using var helper = new MqttPacketSerializationHelper(protocolVersion); + return helper.Encode(packet).ToArray(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs b/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs index 24210b85..a6e8c555 100644 --- a/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs +++ b/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs @@ -10,7 +10,7 @@ namespace MQTTnet.Formatter public interface IMqttPacketFormatter { MqttPacket Decode(ReceivedMqttPacket receivedMqttPacket); - + MqttPacketBuffer Encode(MqttPacket mqttPacket); } } \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs index c9010cab..e27f129f 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs @@ -68,12 +68,12 @@ namespace MQTTnet.Formatter.V5 MqttPacket DecodeAuthPacket(ArraySegment body) { - ThrowIfBodyIsEmpty(body); - _bufferReader.SetBuffer(body.Array, body.Offset, body.Count); var packet = new MqttAuthPacket(); + // MQTT spec: The Reason Code and Property Length can be omitted if the Reason Code is 0x00 (Success) and there are no Properties. + // In this case the AUTH has a Remaining Length of 0. if (_bufferReader.EndOfStream) { packet.ReasonCode = MqttAuthenticateReasonCode.Success; diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs index df04f6f9..95699925 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs @@ -9,534 +9,538 @@ using MQTTnet.Exceptions; using MQTTnet.Packets; using MQTTnet.Protocol; -namespace MQTTnet.Formatter.V5 +namespace MQTTnet.Formatter.V5; + +public sealed class MqttV5PacketEncoder { - public sealed class MqttV5PacketEncoder + const int FixedHeaderSize = 1; + + readonly MqttBufferWriter _bufferWriter; + readonly MqttV5PropertiesWriter _propertiesWriter = new(new MqttBufferWriter(1024, 4096)); + + public MqttV5PacketEncoder(MqttBufferWriter bufferWriter) { - const int FixedHeaderSize = 1; + _bufferWriter = bufferWriter ?? throw new ArgumentNullException(nameof(bufferWriter)); + } - readonly MqttBufferWriter _bufferWriter; - readonly MqttV5PropertiesWriter _propertiesWriter = new MqttV5PropertiesWriter(new MqttBufferWriter(1024, 4096)); - - public MqttV5PacketEncoder(MqttBufferWriter bufferWriter) + public MqttPacketBuffer Encode(MqttPacket packet) + { + if (packet == null) { - _bufferWriter = bufferWriter ?? throw new ArgumentNullException(nameof(bufferWriter)); + throw new ArgumentNullException(nameof(packet)); } - public MqttPacketBuffer Encode(MqttPacket packet) + // Leave enough head space for max header size (fixed + 4 variable remaining length = 5 bytes) + const int ReservedHeaderSize = 5; + _bufferWriter.Reset(ReservedHeaderSize); + _bufferWriter.Seek(ReservedHeaderSize); + + var fixedHeader = EncodePacket(packet); + var remainingLength = (uint)_bufferWriter.Length - ReservedHeaderSize; + + ReadOnlySequence payload = default; + if (packet is MqttPublishPacket publishPacket) { - if (packet == null) - { - throw new ArgumentNullException(nameof(packet)); - } - - // Leave enough head space for max header size (fixed + 4 variable remaining length = 5 bytes) - const int ReservedHeaderSize = 5; - _bufferWriter.Reset(ReservedHeaderSize); - _bufferWriter.Seek(ReservedHeaderSize); - - var fixedHeader = EncodePacket(packet); - uint remainingLength = (uint)_bufferWriter.Length - ReservedHeaderSize; - - ReadOnlySequence payload = default; - if (packet is MqttPublishPacket publishPacket) - { - payload = publishPacket.Payload; - remainingLength += (uint)payload.Length; - } - else - { - publishPacket = null; - } - - var remainingLengthSize = MqttBufferWriter.GetVariableByteIntegerSize(remainingLength); - - var headerSize = FixedHeaderSize + remainingLengthSize; - var headerOffset = 5 - headerSize; - - // Position cursor on correct offset on beginning of array (has leading 0x0) - _bufferWriter.Seek(headerOffset); - _bufferWriter.WriteByte(fixedHeader); - _bufferWriter.WriteVariableByteInteger(remainingLength); - - var buffer = _bufferWriter.GetBuffer(); - var firstSegment = new ArraySegment(buffer, headerOffset, _bufferWriter.Length - headerOffset); - - return publishPacket == null - ? new MqttPacketBuffer(firstSegment) - : new MqttPacketBuffer(firstSegment, publishPacket.Payload); + payload = publishPacket.Payload; + remainingLength += (uint)payload.Length; + } + else + { + publishPacket = null; } - byte EncodeAuthPacket(MqttAuthPacket packet) + var remainingLengthSize = MqttBufferWriter.GetVariableByteIntegerSize(remainingLength); + + var headerSize = FixedHeaderSize + remainingLengthSize; + var headerOffset = 5 - headerSize; + + // Position cursor on correct offset on beginning of array (has leading 0x0) + _bufferWriter.Seek(headerOffset); + _bufferWriter.WriteByte(fixedHeader); + _bufferWriter.WriteVariableByteInteger(remainingLength); + + var buffer = _bufferWriter.GetBuffer(); + var firstSegment = new ArraySegment(buffer, headerOffset, _bufferWriter.Length - headerOffset); + + return publishPacket == null ? new MqttPacketBuffer(firstSegment) : new MqttPacketBuffer(firstSegment, publishPacket.Payload); + } + + byte EncodeAuthPacket(MqttAuthPacket packet) + { + _propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod); + _propertiesWriter.WriteAuthenticationData(packet.AuthenticationData); + _propertiesWriter.WriteReasonString(packet.ReasonString); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + // MQTT spec: The Reason Code and Property Length can be omitted if the Reason Code is 0x00 (Success) and there are no Properties. + // In this case the AUTH has a Remaining Length of 0. + if (packet.ReasonCode == MqttAuthenticateReasonCode.Success && _propertiesWriter.Length == 0) { - _bufferWriter.WriteByte((byte)packet.ReasonCode); - - _propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod); - _propertiesWriter.WriteAuthenticationData(packet.AuthenticationData); - _propertiesWriter.WriteReasonString(packet.ReasonString); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Auth); } - byte EncodeConnAckPacket(MqttConnAckPacket packet) + _bufferWriter.WriteByte((byte)packet.ReasonCode); + + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); + + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Auth); + } + + byte EncodeConnAckPacket(MqttConnAckPacket packet) + { + byte connectAcknowledgeFlags = 0x0; + if (packet.IsSessionPresent) { - byte connectAcknowledgeFlags = 0x0; - if (packet.IsSessionPresent) + connectAcknowledgeFlags |= 0x1; + } + + _bufferWriter.WriteByte(connectAcknowledgeFlags); + _bufferWriter.WriteByte((byte)packet.ReasonCode); + + _propertiesWriter.WriteSessionExpiryInterval(packet.SessionExpiryInterval); + _propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod); + _propertiesWriter.WriteAuthenticationData(packet.AuthenticationData); + _propertiesWriter.WriteRetainAvailable(packet.RetainAvailable); + _propertiesWriter.WriteReceiveMaximum(packet.ReceiveMaximum); + _propertiesWriter.WriteMaximumQoS(packet.MaximumQoS); + _propertiesWriter.WriteAssignedClientIdentifier(packet.AssignedClientIdentifier); + _propertiesWriter.WriteTopicAliasMaximum(packet.TopicAliasMaximum); + _propertiesWriter.WriteReasonString(packet.ReasonString); + _propertiesWriter.WriteMaximumPacketSize(packet.MaximumPacketSize); + _propertiesWriter.WriteWildcardSubscriptionAvailable(packet.WildcardSubscriptionAvailable); + _propertiesWriter.WriteSubscriptionIdentifiersAvailable(packet.SubscriptionIdentifiersAvailable); + _propertiesWriter.WriteSharedSubscriptionAvailable(packet.SharedSubscriptionAvailable); + _propertiesWriter.WriteServerKeepAlive(packet.ServerKeepAlive); + _propertiesWriter.WriteResponseInformation(packet.ResponseInformation); + _propertiesWriter.WriteServerReference(packet.ServerReference); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); + + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); + } + + byte EncodeConnectPacket(MqttConnectPacket packet) + { + if (string.IsNullOrEmpty(packet.ClientId) && !packet.CleanSession) + { + throw new MqttProtocolViolationException("CleanSession must be set if ClientId is empty [MQTT-3.1.3-7]."); + } + + _bufferWriter.WriteString("MQTT"); + _bufferWriter.WriteByte(5); // [3.1.2.2 Protocol Version] + + byte connectFlags = 0x0; + if (packet.CleanSession) + { + connectFlags |= 0x2; + } + + if (packet.WillFlag) + { + connectFlags |= 0x4; + connectFlags |= (byte)((byte)packet.WillQoS << 3); + + if (packet.WillRetain) { - connectAcknowledgeFlags |= 0x1; + connectFlags |= 0x20; } + } - _bufferWriter.WriteByte(connectAcknowledgeFlags); - _bufferWriter.WriteByte((byte)packet.ReasonCode); + if (packet.Password != null && packet.Username == null) + { + throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22]."); + } - _propertiesWriter.WriteSessionExpiryInterval(packet.SessionExpiryInterval); - _propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod); - _propertiesWriter.WriteAuthenticationData(packet.AuthenticationData); - _propertiesWriter.WriteRetainAvailable(packet.RetainAvailable); - _propertiesWriter.WriteReceiveMaximum(packet.ReceiveMaximum); - _propertiesWriter.WriteMaximumQoS(packet.MaximumQoS); - _propertiesWriter.WriteAssignedClientIdentifier(packet.AssignedClientIdentifier); - _propertiesWriter.WriteTopicAliasMaximum(packet.TopicAliasMaximum); - _propertiesWriter.WriteReasonString(packet.ReasonString); - _propertiesWriter.WriteMaximumPacketSize(packet.MaximumPacketSize); - _propertiesWriter.WriteWildcardSubscriptionAvailable(packet.WildcardSubscriptionAvailable); - _propertiesWriter.WriteSubscriptionIdentifiersAvailable(packet.SubscriptionIdentifiersAvailable); - _propertiesWriter.WriteSharedSubscriptionAvailable(packet.SharedSubscriptionAvailable); - _propertiesWriter.WriteServerKeepAlive(packet.ServerKeepAlive); - _propertiesWriter.WriteResponseInformation(packet.ResponseInformation); - _propertiesWriter.WriteServerReference(packet.ServerReference); - _propertiesWriter.WriteUserProperties(packet.UserProperties); + if (packet.Password != null) + { + connectFlags |= 0x40; + } + + if (packet.Username != null) + { + connectFlags |= 0x80; + } + + _bufferWriter.WriteByte(connectFlags); + _bufferWriter.WriteTwoByteInteger(packet.KeepAlivePeriod); + + _propertiesWriter.WriteSessionExpiryInterval(packet.SessionExpiryInterval); + _propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod); + _propertiesWriter.WriteAuthenticationData(packet.AuthenticationData); + _propertiesWriter.WriteRequestProblemInformation(packet.RequestProblemInformation); + _propertiesWriter.WriteRequestResponseInformation(packet.RequestResponseInformation); + _propertiesWriter.WriteReceiveMaximum(packet.ReceiveMaximum); + _propertiesWriter.WriteTopicAliasMaximum(packet.TopicAliasMaximum); + _propertiesWriter.WriteMaximumPacketSize(packet.MaximumPacketSize); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); + + _bufferWriter.WriteString(packet.ClientId); + + if (packet.WillFlag) + { + _propertiesWriter.WritePayloadFormatIndicator(packet.WillPayloadFormatIndicator); + _propertiesWriter.WriteMessageExpiryInterval(packet.WillMessageExpiryInterval); + _propertiesWriter.WriteResponseTopic(packet.WillResponseTopic); + _propertiesWriter.WriteCorrelationData(packet.WillCorrelationData); + _propertiesWriter.WriteContentType(packet.WillContentType); + _propertiesWriter.WriteUserProperties(packet.WillUserProperties); + _propertiesWriter.WriteWillDelayInterval(packet.WillDelayInterval); _propertiesWriter.WriteTo(_bufferWriter); _propertiesWriter.Reset(); - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); + _bufferWriter.WriteString(packet.WillTopic); + _bufferWriter.WriteBinary(packet.WillMessage); } - byte EncodeConnectPacket(MqttConnectPacket packet) + if (packet.Username != null) { - if (string.IsNullOrEmpty(packet.ClientId) && !packet.CleanSession) - { - throw new MqttProtocolViolationException("CleanSession must be set if ClientId is empty [MQTT-3.1.3-7]."); - } - - _bufferWriter.WriteString("MQTT"); - _bufferWriter.WriteByte(5); // [3.1.2.2 Protocol Version] - - byte connectFlags = 0x0; - if (packet.CleanSession) - { - connectFlags |= 0x2; - } - - if (packet.WillFlag) - { - connectFlags |= 0x4; - connectFlags |= (byte)((byte)packet.WillQoS << 3); - - if (packet.WillRetain) - { - connectFlags |= 0x20; - } - } - - if (packet.Password != null && packet.Username == null) - { - throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22]."); - } - - if (packet.Password != null) - { - connectFlags |= 0x40; - } - - if (packet.Username != null) - { - connectFlags |= 0x80; - } - - _bufferWriter.WriteByte(connectFlags); - _bufferWriter.WriteTwoByteInteger(packet.KeepAlivePeriod); - - _propertiesWriter.WriteSessionExpiryInterval(packet.SessionExpiryInterval); - _propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod); - _propertiesWriter.WriteAuthenticationData(packet.AuthenticationData); - _propertiesWriter.WriteRequestProblemInformation(packet.RequestProblemInformation); - _propertiesWriter.WriteRequestResponseInformation(packet.RequestResponseInformation); - _propertiesWriter.WriteReceiveMaximum(packet.ReceiveMaximum); - _propertiesWriter.WriteTopicAliasMaximum(packet.TopicAliasMaximum); - _propertiesWriter.WriteMaximumPacketSize(packet.MaximumPacketSize); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); - - _bufferWriter.WriteString(packet.ClientId); - - if (packet.WillFlag) - { - _propertiesWriter.WritePayloadFormatIndicator(packet.WillPayloadFormatIndicator); - _propertiesWriter.WriteMessageExpiryInterval(packet.WillMessageExpiryInterval); - _propertiesWriter.WriteResponseTopic(packet.WillResponseTopic); - _propertiesWriter.WriteCorrelationData(packet.WillCorrelationData); - _propertiesWriter.WriteContentType(packet.WillContentType); - _propertiesWriter.WriteUserProperties(packet.WillUserProperties); - _propertiesWriter.WriteWillDelayInterval(packet.WillDelayInterval); - - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); - - _bufferWriter.WriteString(packet.WillTopic); - _bufferWriter.WriteBinary(packet.WillMessage); - } - - if (packet.Username != null) - { - _bufferWriter.WriteString(packet.Username); - } - - if (packet.Password != null) - { - _bufferWriter.WriteBinary(packet.Password); - } - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Connect); + _bufferWriter.WriteString(packet.Username); } - byte EncodeDisconnectPacket(MqttDisconnectPacket packet) + if (packet.Password != null) + { + _bufferWriter.WriteBinary(packet.Password); + } + + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Connect); + } + + byte EncodeDisconnectPacket(MqttDisconnectPacket packet) + { + _bufferWriter.WriteByte((byte)packet.ReasonCode); + + _propertiesWriter.WriteServerReference(packet.ServerReference); + _propertiesWriter.WriteReasonString(packet.ReasonString); + _propertiesWriter.WriteSessionExpiryInterval(packet.SessionExpiryInterval); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); + + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Disconnect); + } + + byte EncodePacket(MqttPacket packet) + { + switch (packet) + { + case MqttConnectPacket connectPacket: + return EncodeConnectPacket(connectPacket); + case MqttConnAckPacket connAckPacket: + return EncodeConnAckPacket(connAckPacket); + case MqttDisconnectPacket disconnectPacket: + return EncodeDisconnectPacket(disconnectPacket); + case MqttPingReqPacket _: + return EncodePingReqPacket(); + case MqttPingRespPacket _: + return EncodePingRespPacket(); + case MqttPublishPacket publishPacket: + return EncodePublishPacket(publishPacket); + case MqttPubAckPacket pubAckPacket: + return EncodePubAckPacket(pubAckPacket); + case MqttPubRecPacket pubRecPacket: + return EncodePubRecPacket(pubRecPacket); + case MqttPubRelPacket pubRelPacket: + return EncodePubRelPacket(pubRelPacket); + case MqttPubCompPacket pubCompPacket: + return EncodePubCompPacket(pubCompPacket); + case MqttSubscribePacket subscribePacket: + return EncodeSubscribePacket(subscribePacket); + case MqttSubAckPacket subAckPacket: + return EncodeSubAckPacket(subAckPacket); + case MqttUnsubscribePacket unsubscribePacket: + return EncodeUnsubscribePacket(unsubscribePacket); + case MqttUnsubAckPacket unsubAckPacket: + return EncodeUnsubAckPacket(unsubAckPacket); + case MqttAuthPacket authPacket: + return EncodeAuthPacket(authPacket); + + default: + throw new MqttProtocolViolationException("Packet type invalid."); + } + } + + static byte EncodePingReqPacket() + { + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PingReq); + } + + static byte EncodePingRespPacket() + { + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PingResp); + } + + byte EncodePubAckPacket(MqttPubAckPacket packet) + { + if (packet.PacketIdentifier == 0) + { + throw new MqttProtocolViolationException("PubAck packet has no packet identifier."); + } + + _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); + + _propertiesWriter.WriteReasonString(packet.ReasonString); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubAckReasonCode.Success) { _bufferWriter.WriteByte((byte)packet.ReasonCode); - - _propertiesWriter.WriteServerReference(packet.ServerReference); - _propertiesWriter.WriteReasonString(packet.ReasonString); - _propertiesWriter.WriteSessionExpiryInterval(packet.SessionExpiryInterval); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - _propertiesWriter.WriteTo(_bufferWriter); _propertiesWriter.Reset(); - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Disconnect); } - byte EncodePacket(MqttPacket packet) + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PubAck); + } + + byte EncodePubCompPacket(MqttPubCompPacket packet) + { + ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); + + _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); + + _propertiesWriter.WriteReasonString(packet.ReasonString); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubCompReasonCode.Success) { - switch (packet) - { - case MqttConnectPacket connectPacket: - return EncodeConnectPacket(connectPacket); - case MqttConnAckPacket connAckPacket: - return EncodeConnAckPacket(connAckPacket); - case MqttDisconnectPacket disconnectPacket: - return EncodeDisconnectPacket(disconnectPacket); - case MqttPingReqPacket _: - return EncodePingReqPacket(); - case MqttPingRespPacket _: - return EncodePingRespPacket(); - case MqttPublishPacket publishPacket: - return EncodePublishPacket(publishPacket); - case MqttPubAckPacket pubAckPacket: - return EncodePubAckPacket(pubAckPacket); - case MqttPubRecPacket pubRecPacket: - return EncodePubRecPacket(pubRecPacket); - case MqttPubRelPacket pubRelPacket: - return EncodePubRelPacket(pubRelPacket); - case MqttPubCompPacket pubCompPacket: - return EncodePubCompPacket(pubCompPacket); - case MqttSubscribePacket subscribePacket: - return EncodeSubscribePacket(subscribePacket); - case MqttSubAckPacket subAckPacket: - return EncodeSubAckPacket(subAckPacket); - case MqttUnsubscribePacket unsubscribePacket: - return EncodeUnsubscribePacket(unsubscribePacket); - case MqttUnsubAckPacket unsubAckPacket: - return EncodeUnsubAckPacket(unsubAckPacket); - case MqttAuthPacket authPacket: - return EncodeAuthPacket(authPacket); - - default: - throw new MqttProtocolViolationException("Packet type invalid."); - } + _bufferWriter.WriteByte((byte)packet.ReasonCode); + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); } - static byte EncodePingReqPacket() + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PubComp); + } + + byte EncodePublishPacket(MqttPublishPacket packet) + { + if (packet.QualityOfServiceLevel == 0 && packet.Dup) { - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PingReq); + throw new MqttProtocolViolationException("Dup flag must be false for QoS 0 packets [MQTT-3.3.1-2]."); } - static byte EncodePingRespPacket() - { - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PingResp); - } + _bufferWriter.WriteString(packet.Topic); - byte EncodePubAckPacket(MqttPubAckPacket packet) + if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { if (packet.PacketIdentifier == 0) { - throw new MqttProtocolViolationException("PubAck packet has no packet identifier."); + throw new MqttProtocolViolationException("Publish packet has no packet identifier."); } _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); - - _propertiesWriter.WriteReasonString(packet.ReasonString); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - - if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubAckReasonCode.Success) + } + else + { + if (packet.PacketIdentifier > 0) { - _bufferWriter.WriteByte((byte)packet.ReasonCode); - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); + throw new MqttProtocolViolationException("Packet identifier must be 0 if QoS == 0 [MQTT-2.3.1-5]."); } - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PubAck); } - byte EncodePubCompPacket(MqttPubCompPacket packet) + _propertiesWriter.WriteContentType(packet.ContentType); + _propertiesWriter.WriteCorrelationData(packet.CorrelationData); + _propertiesWriter.WriteMessageExpiryInterval(packet.MessageExpiryInterval); + _propertiesWriter.WritePayloadFormatIndicator(packet.PayloadFormatIndicator); + _propertiesWriter.WriteResponseTopic(packet.ResponseTopic); + _propertiesWriter.WriteSubscriptionIdentifiers(packet.SubscriptionIdentifiers); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + _propertiesWriter.WriteTopicAlias(packet.TopicAlias); + + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); + + // The payload is the past part of the packet. But it is not added here in order to keep + // memory allocation low. + + byte fixedHeader = 0; + + if (packet.Retain) { - ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - - _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); - - _propertiesWriter.WriteReasonString(packet.ReasonString); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - - if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubCompReasonCode.Success) - { - _bufferWriter.WriteByte((byte)packet.ReasonCode); - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); - } - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PubComp); + fixedHeader |= 0x01; } - byte EncodePublishPacket(MqttPublishPacket packet) + fixedHeader |= (byte)((byte)packet.QualityOfServiceLevel << 1); + + if (packet.Dup) { - if (packet.QualityOfServiceLevel == 0 && packet.Dup) - { - throw new MqttProtocolViolationException("Dup flag must be false for QoS 0 packets [MQTT-3.3.1-2]."); - } + fixedHeader |= 0x08; + } - _bufferWriter.WriteString(packet.Topic); + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader); + } - if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) - { - if (packet.PacketIdentifier == 0) - { - throw new MqttProtocolViolationException("Publish packet has no packet identifier."); - } + byte EncodePubRecPacket(MqttPubRecPacket packet) + { + ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); - } - else - { - if (packet.PacketIdentifier > 0) - { - throw new MqttProtocolViolationException("Packet identifier must be 0 if QoS == 0 [MQTT-2.3.1-5]."); - } - } + _propertiesWriter.WriteReasonString(packet.ReasonString); + _propertiesWriter.WriteUserProperties(packet.UserProperties); - _propertiesWriter.WriteContentType(packet.ContentType); - _propertiesWriter.WriteCorrelationData(packet.CorrelationData); - _propertiesWriter.WriteMessageExpiryInterval(packet.MessageExpiryInterval); - _propertiesWriter.WritePayloadFormatIndicator(packet.PayloadFormatIndicator); - _propertiesWriter.WriteResponseTopic(packet.ResponseTopic); - _propertiesWriter.WriteSubscriptionIdentifiers(packet.SubscriptionIdentifiers); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - _propertiesWriter.WriteTopicAlias(packet.TopicAlias); + _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); + if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubRecReasonCode.Success) + { + _bufferWriter.WriteByte((byte)packet.ReasonCode); _propertiesWriter.WriteTo(_bufferWriter); _propertiesWriter.Reset(); - - // The payload is the past part of the packet. But it is not added here in order to keep - // memory allocation low. - - byte fixedHeader = 0; - - if (packet.Retain) - { - fixedHeader |= 0x01; - } - - fixedHeader |= (byte)((byte)packet.QualityOfServiceLevel << 1); - - if (packet.Dup) - { - fixedHeader |= 0x08; - } - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader); } - byte EncodePubRecPacket(MqttPubRecPacket packet) + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PubRec); + } + + byte EncodePubRelPacket(MqttPubRelPacket packet) + { + ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); + + _propertiesWriter.WriteReasonString(packet.ReasonString); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); + + if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubRelReasonCode.Success) { - ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - - _propertiesWriter.WriteReasonString(packet.ReasonString); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - - _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); - - if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubRecReasonCode.Success) - { - _bufferWriter.WriteByte((byte)packet.ReasonCode); - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); - } - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PubRec); - } - - byte EncodePubRelPacket(MqttPubRelPacket packet) - { - ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - - _propertiesWriter.WriteReasonString(packet.ReasonString); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - - _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); - - if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubRelReasonCode.Success) - { - _bufferWriter.WriteByte((byte)packet.ReasonCode); - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); - } - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02); - } - - byte EncodeSubAckPacket(MqttSubAckPacket packet) - { - if (packet.ReasonCodes?.Any() != true) - { - throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3]."); - } - - ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - - _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); - - _propertiesWriter.WriteReasonString(packet.ReasonString); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - + _bufferWriter.WriteByte((byte)packet.ReasonCode); _propertiesWriter.WriteTo(_bufferWriter); _propertiesWriter.Reset(); - - foreach (var reasonCode in packet.ReasonCodes) - { - _bufferWriter.WriteByte((byte)reasonCode); - } - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.SubAck); } - byte EncodeSubscribePacket(MqttSubscribePacket packet) + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02); + } + + byte EncodeSubAckPacket(MqttSubAckPacket packet) + { + if (packet.ReasonCodes?.Any() != true) { - if (packet.TopicFilters?.Any() != true) - { - throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3]."); - } - - ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - - _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); - - if (packet.SubscriptionIdentifier > 0) - { - _propertiesWriter.WriteSubscriptionIdentifier(packet.SubscriptionIdentifier); - } - - _propertiesWriter.WriteUserProperties(packet.UserProperties); - - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); - - if (packet.TopicFilters?.Count > 0) - { - foreach (var topicFilter in packet.TopicFilters) - { - _bufferWriter.WriteString(topicFilter.Topic); - - var options = (byte)topicFilter.QualityOfServiceLevel; - - if (topicFilter.NoLocal) - { - options |= 1 << 2; - } - - if (topicFilter.RetainAsPublished) - { - options |= 1 << 3; - } - - if (topicFilter.RetainHandling != MqttRetainHandling.SendAtSubscribe) - { - options |= (byte)((byte)topicFilter.RetainHandling << 4); - } - - _bufferWriter.WriteByte(options); - } - } - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02); + throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3]."); } - byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet) + ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); + + _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); + + _propertiesWriter.WriteReasonString(packet.ReasonString); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); + + foreach (var reasonCode in packet.ReasonCodes) { - ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - - _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); - - _propertiesWriter.WriteReasonString(packet.ReasonString); - _propertiesWriter.WriteUserProperties(packet.UserProperties); - - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); - - foreach (var reasonCode in packet.ReasonCodes) - { - _bufferWriter.WriteByte((byte)reasonCode); - } - - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck); + _bufferWriter.WriteByte((byte)reasonCode); } - byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet) + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.SubAck); + } + + byte EncodeSubscribePacket(MqttSubscribePacket packet) + { + if (packet.TopicFilters?.Any() != true) { - if (packet.TopicFilters?.Any() != true) - { - throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2]."); - } + throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3]."); + } - ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); + ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); - _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); + _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); - _propertiesWriter.WriteUserProperties(packet.UserProperties); + if (packet.SubscriptionIdentifier > 0) + { + _propertiesWriter.WriteSubscriptionIdentifier(packet.SubscriptionIdentifier); + } - _propertiesWriter.WriteTo(_bufferWriter); - _propertiesWriter.Reset(); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); + + if (packet.TopicFilters?.Count > 0) + { foreach (var topicFilter in packet.TopicFilters) { - _bufferWriter.WriteString(topicFilter); - } + _bufferWriter.WriteString(topicFilter.Topic); - return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Unsubscribe, 0x02); + var options = (byte)topicFilter.QualityOfServiceLevel; + + if (topicFilter.NoLocal) + { + options |= 1 << 2; + } + + if (topicFilter.RetainAsPublished) + { + options |= 1 << 3; + } + + if (topicFilter.RetainHandling != MqttRetainHandling.SendAtSubscribe) + { + options |= (byte)((byte)topicFilter.RetainHandling << 4); + } + + _bufferWriter.WriteByte(options); + } } - static void ThrowIfPacketIdentifierIsInvalid(ushort packetIdentifier, MqttPacket packet) - { - // SUBSCRIBE, UNSUBSCRIBE, and PUBLISH(in cases where QoS > 0) Control Packets MUST contain a non-zero 16 - bit Packet Identifier[MQTT - 2.3.1 - 1]. + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02); + } - if (packetIdentifier == 0) - { - throw new MqttProtocolViolationException($"Packet identifier is not set for {packet.GetType().Name}."); - } + byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet) + { + ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); + + _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); + + _propertiesWriter.WriteReasonString(packet.ReasonString); + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); + + foreach (var reasonCode in packet.ReasonCodes) + { + _bufferWriter.WriteByte((byte)reasonCode); + } + + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck); + } + + byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet) + { + if (packet.TopicFilters?.Any() != true) + { + throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2]."); + } + + ThrowIfPacketIdentifierIsInvalid(packet.PacketIdentifier, packet); + + _bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier); + + _propertiesWriter.WriteUserProperties(packet.UserProperties); + + _propertiesWriter.WriteTo(_bufferWriter); + _propertiesWriter.Reset(); + + foreach (var topicFilter in packet.TopicFilters) + { + _bufferWriter.WriteString(topicFilter); + } + + return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Unsubscribe, 0x02); + } + + static void ThrowIfPacketIdentifierIsInvalid(ushort packetIdentifier, MqttPacket packet) + { + // SUBSCRIBE, UNSUBSCRIBE, and PUBLISH(in cases where QoS > 0) Control Packets MUST contain a non-zero 16 - bit Packet Identifier[MQTT - 2.3.1 - 1]. + + if (packetIdentifier == 0) + { + throw new MqttProtocolViolationException($"Packet identifier is not set for {packet.GetType().Name}."); } } -} +} \ No newline at end of file