From c22b6033e3eb9195bea9fbd8088a9e551cb75ad6 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sun, 12 May 2019 17:26:36 +0200 Subject: [PATCH] Accept multiple SubscriptionIdentifiers for PUBLISH packets. --- .../Formatter/V5/MqttV500DataConverter.cs | 5 +- .../Formatter/V5/MqttV500PacketDecoder.cs | 53 +++++++++++++------ .../Formatter/V5/MqttV500PacketEncoder.cs | 4 +- .../Formatter/V5/MqttV500PropertiesReader.cs | 32 +++++------ .../Formatter/V5/MqttV500PropertiesWriter.cs | 13 +++++ Source/MQTTnet/MqttApplicationMessage.cs | 2 +- .../MQTTnet/MqttApplicationMessageBuilder.cs | 21 ++++---- .../Packets/MqttPublishPacketProperties.cs | 4 +- .../MqttPacketSerializer_Tests.cs | 3 +- 9 files changed, 87 insertions(+), 50 deletions(-) diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs index 2a14f403..92766177 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -33,13 +33,14 @@ namespace MQTTnet.Formatter.V5 MessageExpiryInterval = applicationMessage.MessageExpiryInterval, PayloadFormatIndicator = applicationMessage.PayloadFormatIndicator, ResponseTopic = applicationMessage.ResponseTopic, - SubscriptionIdentifier = applicationMessage.SubscriptionIdentifier, + SubscriptionIdentifiers = applicationMessage.SubscriptionIdentifiers, TopicAlias = applicationMessage.TopicAlias } }; if (applicationMessage.UserProperties != null) { + packet.Properties.UserProperties = new List(); packet.Properties.UserProperties.AddRange(applicationMessage.UserProperties); } @@ -67,7 +68,7 @@ namespace MQTTnet.Formatter.V5 ContentType = publishPacket.Properties?.ContentType, CorrelationData = publishPacket.Properties?.CorrelationData, MessageExpiryInterval = publishPacket.Properties?.MessageExpiryInterval, - SubscriptionIdentifier = publishPacket.Properties?.SubscriptionIdentifier, + SubscriptionIdentifiers = publishPacket.Properties?.SubscriptionIdentifiers, TopicAlias = publishPacket.Properties?.TopicAlias, PayloadFormatIndicator = publishPacket.Properties?.PayloadFormatIndicator, UserProperties = publishPacket.Properties?.UserProperties ?? new List() diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs index fa53c5d8..4c394e46 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Linq; using MQTTnet.Adapter; using MQTTnet.Exceptions; @@ -117,7 +118,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -155,7 +156,12 @@ namespace MQTTnet.Formatter.V5 } else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier) { - packet.WillMessage.SubscriptionIdentifier = propertiesReader.ReadSubscriptionIdentifier(); + if (packet.WillMessage.SubscriptionIdentifiers == null) + { + packet.WillMessage.SubscriptionIdentifiers = new List(); + } + + packet.WillMessage.SubscriptionIdentifiers.Add(propertiesReader.ReadSubscriptionIdentifier()); } else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.ContentType) { @@ -168,7 +174,12 @@ namespace MQTTnet.Formatter.V5 } else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + if (packet.WillMessage.UserProperties == null) + { + packet.WillMessage.UserProperties = new List(); + } + + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -271,7 +282,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -309,7 +320,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -339,7 +350,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -389,7 +400,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -421,7 +432,7 @@ namespace MQTTnet.Formatter.V5 { if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -456,7 +467,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -530,7 +541,12 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier) { - packet.Properties.SubscriptionIdentifier = propertiesReader.ReadSubscriptionIdentifier(); + if (packet.Properties.SubscriptionIdentifiers == null) + { + packet.Properties.SubscriptionIdentifiers = new List(); + } + + packet.Properties.SubscriptionIdentifiers.Add(propertiesReader.ReadSubscriptionIdentifier()); } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.ContentType) { @@ -538,7 +554,12 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + if (packet.Properties.UserProperties == null) + { + packet.Properties.UserProperties = new List(); + } + + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -581,7 +602,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -619,7 +640,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -657,7 +678,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -695,7 +716,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { @@ -740,7 +761,7 @@ namespace MQTTnet.Formatter.V5 } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) { - propertiesReader.FillUserProperties(packet.Properties.UserProperties); + propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties); } else { diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs index 89a76bf1..23b7f1a6 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs @@ -150,7 +150,7 @@ namespace MQTTnet.Formatter.V5 willPropertiesWriter.WriteTopicAlias(packet.WillMessage.TopicAlias); willPropertiesWriter.WriteResponseTopic(packet.WillMessage.ResponseTopic); willPropertiesWriter.WriteCorrelationData(packet.WillMessage.CorrelationData); - willPropertiesWriter.WriteSubscriptionIdentifier(packet.WillMessage.SubscriptionIdentifier); + willPropertiesWriter.WriteSubscriptionIdentifiers(packet.WillMessage.SubscriptionIdentifiers); willPropertiesWriter.WriteContentType(packet.WillMessage.ContentType); willPropertiesWriter.WriteUserProperties(packet.WillMessage.UserProperties); @@ -258,7 +258,7 @@ namespace MQTTnet.Formatter.V5 propertiesWriter.WriteTopicAlias(packet.Properties.TopicAlias); propertiesWriter.WriteResponseTopic(packet.Properties.ResponseTopic); propertiesWriter.WriteCorrelationData(packet.Properties.CorrelationData); - propertiesWriter.WriteSubscriptionIdentifier(packet.Properties.SubscriptionIdentifier); + propertiesWriter.WriteSubscriptionIdentifiers(packet.Properties.SubscriptionIdentifiers); propertiesWriter.WriteContentType(packet.Properties.ContentType); propertiesWriter.WriteUserProperties(packet.Properties.UserProperties); } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs index a61cb066..df323aee 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs @@ -42,7 +42,7 @@ namespace MQTTnet.Formatter.V5 return true; } - public void FillUserProperties(List userProperties) + public void AddUserPropertyTo(List userProperties) { if (userProperties == null) throw new ArgumentNullException(nameof(userProperties)); @@ -67,17 +67,17 @@ namespace MQTTnet.Formatter.V5 return _body.ReadWithLengthPrefix(); } - public bool? ReadRetainAvailable() + public bool ReadRetainAvailable() { return _body.ReadBoolean(); } - public uint? ReadSessionExpiryInterval() + public uint ReadSessionExpiryInterval() { return _body.ReadFourByteInteger(); } - public ushort? ReadReceiveMaximum() + public ushort ReadReceiveMaximum() { return _body.ReadTwoByteInteger(); } @@ -92,17 +92,17 @@ namespace MQTTnet.Formatter.V5 return _body.ReadStringWithLengthPrefix(); } - public ushort? ReadTopicAliasMaximum() + public ushort ReadTopicAliasMaximum() { return _body.ReadTwoByteInteger(); } - public uint? ReadMaximumPacketSize() + public uint ReadMaximumPacketSize() { return _body.ReadFourByteInteger(); } - public ushort? ReadServerKeepAlive() + public ushort ReadServerKeepAlive() { return _body.ReadTwoByteInteger(); } @@ -112,22 +112,22 @@ namespace MQTTnet.Formatter.V5 return _body.ReadStringWithLengthPrefix(); } - public bool? ReadSharedSubscriptionAvailable() + public bool ReadSharedSubscriptionAvailable() { return _body.ReadBoolean(); } - public bool? ReadSubscriptionIdentifiersAvailable() + public bool ReadSubscriptionIdentifiersAvailable() { return _body.ReadBoolean(); } - public bool? ReadWildcardSubscriptionAvailable() + public bool ReadWildcardSubscriptionAvailable() { return _body.ReadBoolean(); } - public uint? ReadSubscriptionIdentifier() + public uint ReadSubscriptionIdentifier() { return _body.ReadVariableLengthInteger(); } @@ -137,12 +137,12 @@ namespace MQTTnet.Formatter.V5 return (MqttPayloadFormatIndicator)_body.ReadByte(); } - public uint? ReadMessageExpiryInterval() + public uint ReadMessageExpiryInterval() { return _body.ReadFourByteInteger(); } - public ushort? ReadTopicAlias() + public ushort ReadTopicAlias() { return _body.ReadTwoByteInteger(); } @@ -162,17 +162,17 @@ namespace MQTTnet.Formatter.V5 return _body.ReadStringWithLengthPrefix(); } - public uint? ReadWillDelayInterval() + public uint ReadWillDelayInterval() { return _body.ReadFourByteInteger(); } - public bool? RequestResponseInformation() + public bool RequestResponseInformation() { return _body.ReadBoolean(); } - public bool? RequestProblemInformation() + public bool RequestProblemInformation() { return _body.ReadBoolean(); } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs index 16532304..f632586b 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs @@ -80,6 +80,19 @@ namespace MQTTnet.Formatter.V5 WriteAsVariableLengthInteger(MqttPropertyId.SubscriptionIdentifier, value); } + public void WriteSubscriptionIdentifiers(IEnumerable value) + { + if (value == null) + { + return; + } + + foreach (var subscriptionIdentifier in value) + { + WriteAsVariableLengthInteger(MqttPropertyId.SubscriptionIdentifier, subscriptionIdentifier); + } + } + public void WriteTopicAlias(ushort? value) { Write(MqttPropertyId.TopicAlias, value); diff --git a/Source/MQTTnet/MqttApplicationMessage.cs b/Source/MQTTnet/MqttApplicationMessage.cs index 06bd97a7..6522d42f 100644 --- a/Source/MQTTnet/MqttApplicationMessage.cs +++ b/Source/MQTTnet/MqttApplicationMessage.cs @@ -28,6 +28,6 @@ namespace MQTTnet public byte[] CorrelationData { get; set; } - public uint? SubscriptionIdentifier { get; set; } + public List SubscriptionIdentifiers { get; set; } } } diff --git a/Source/MQTTnet/MqttApplicationMessageBuilder.cs b/Source/MQTTnet/MqttApplicationMessageBuilder.cs index cc962b9f..7d116890 100644 --- a/Source/MQTTnet/MqttApplicationMessageBuilder.cs +++ b/Source/MQTTnet/MqttApplicationMessageBuilder.cs @@ -18,7 +18,7 @@ namespace MQTTnet private string _responseTopic; private byte[] _correlationData; private ushort? _topicAlias; - private uint? _subscriptionIdentifier; + private List _subscriptionIdentifiers; private uint? _messageExpiryInterval; private MqttPayloadFormatIndicator? _payloadFormatIndicator; private List _userProperties; @@ -43,7 +43,7 @@ namespace MQTTnet { _payload = payload.ToArray(); } - + return this; } @@ -176,7 +176,12 @@ namespace MQTTnet /// public MqttApplicationMessageBuilder WithSubscriptionIdentifier(uint subscriptionIdentifier) { - _subscriptionIdentifier = subscriptionIdentifier; + if (_subscriptionIdentifiers == null) + { + _subscriptionIdentifiers = new List(); + } + + _subscriptionIdentifiers.Add(subscriptionIdentifier); return this; } @@ -215,16 +220,12 @@ namespace MQTTnet ResponseTopic = _responseTopic, CorrelationData = _correlationData, TopicAlias = _topicAlias, - SubscriptionIdentifier = _subscriptionIdentifier, + SubscriptionIdentifiers = _subscriptionIdentifiers, MessageExpiryInterval = _messageExpiryInterval, - PayloadFormatIndicator = _payloadFormatIndicator + PayloadFormatIndicator = _payloadFormatIndicator, + UserProperties = _userProperties }; - if (_userProperties?.Any() == true) - { - applicationMessage.UserProperties = _userProperties; - } - return applicationMessage; } } diff --git a/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs b/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs index 0de4d812..ba2d8c63 100644 --- a/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs @@ -15,9 +15,9 @@ namespace MQTTnet.Packets public byte[] CorrelationData { get; set; } - public List UserProperties { get; } = new List(); + public List UserProperties { get; set; } - public uint? SubscriptionIdentifier { get; set; } + public List SubscriptionIdentifiers { get; set; } public string ContentType { get; set; } } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs index 429a2312..7f29b574 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.IO; using System.Linq; using System.Text; @@ -234,7 +235,7 @@ namespace MQTTnet.Tests [TestMethod] public void SerializeV500_MqttPublishPacket() { - var prop = new MqttPublishPacketProperties(); + var prop = new MqttPublishPacketProperties {UserProperties = new List()}; prop.ResponseTopic = "/Response";