YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-28 01:02:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

Accept multiple SubscriptionIdentifiers for PUBLISH packets.

Browse Source
This commit is contained in:
Christian Kratky
2019-05-12 17:26:36 +02:00
parent 27476a9194
commit c22b6033e3
9 changed files with 87 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs
+3 -2
View File
@@ -33,13 +33,14 @@ namespace MQTTnet.Formatter.V5
MessageExpiryInterval = applicationMessage.MessageExpiryInterval, MessageExpiryInterval = applicationMessage.MessageExpiryInterval,
PayloadFormatIndicator = applicationMessage.PayloadFormatIndicator, PayloadFormatIndicator = applicationMessage.PayloadFormatIndicator,
ResponseTopic = applicationMessage.ResponseTopic, ResponseTopic = applicationMessage.ResponseTopic,
SubscriptionIdentifier = applicationMessage.SubscriptionIdentifier, SubscriptionIdentifiers = applicationMessage.SubscriptionIdentifiers,
TopicAlias = applicationMessage.TopicAlias TopicAlias = applicationMessage.TopicAlias
} }
}; };
if (applicationMessage.UserProperties != null) if (applicationMessage.UserProperties != null)
{ {
packet.Properties.UserProperties = new List<MqttUserProperty>();
packet.Properties.UserProperties.AddRange(applicationMessage.UserProperties); packet.Properties.UserProperties.AddRange(applicationMessage.UserProperties);
} }
@@ -67,7 +68,7 @@ namespace MQTTnet.Formatter.V5
ContentType = publishPacket.Properties?.ContentType, ContentType = publishPacket.Properties?.ContentType,
CorrelationData = publishPacket.Properties?.CorrelationData, CorrelationData = publishPacket.Properties?.CorrelationData,
MessageExpiryInterval = publishPacket.Properties?.MessageExpiryInterval, MessageExpiryInterval = publishPacket.Properties?.MessageExpiryInterval,
SubscriptionIdentifier = publishPacket.Properties?.SubscriptionIdentifier, SubscriptionIdentifiers = publishPacket.Properties?.SubscriptionIdentifiers,
TopicAlias = publishPacket.Properties?.TopicAlias, TopicAlias = publishPacket.Properties?.TopicAlias,
PayloadFormatIndicator = publishPacket.Properties?.PayloadFormatIndicator, PayloadFormatIndicator = publishPacket.Properties?.PayloadFormatIndicator,
UserProperties = publishPacket.Properties?.UserProperties ?? new List<MqttUserProperty>() UserProperties = publishPacket.Properties?.UserProperties ?? new List<MqttUserProperty>()
Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs
+37 -16
View File
@@ -1,4 +1,5 @@
using System; using System;
using System.Collections.Generic;
using System.Linq; using System.Linq;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
@@ -117,7 +118,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -155,7 +156,12 @@ namespace MQTTnet.Formatter.V5
} }
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier) else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier)
{ {
packet.WillMessage.SubscriptionIdentifier = propertiesReader.ReadSubscriptionIdentifier(); if (packet.WillMessage.SubscriptionIdentifiers == null)
{
packet.WillMessage.SubscriptionIdentifiers = new List<uint>();
}
packet.WillMessage.SubscriptionIdentifiers.Add(propertiesReader.ReadSubscriptionIdentifier());
} }
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.ContentType) else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.ContentType)
{ {
@@ -168,7 +174,12 @@ namespace MQTTnet.Formatter.V5
} }
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); if (packet.WillMessage.UserProperties == null)
{
packet.WillMessage.UserProperties = new List<MqttUserProperty>();
}
propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -271,7 +282,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -309,7 +320,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -339,7 +350,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -389,7 +400,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -421,7 +432,7 @@ namespace MQTTnet.Formatter.V5
{ {
if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -456,7 +467,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -530,7 +541,12 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier)
{ {
packet.Properties.SubscriptionIdentifier = propertiesReader.ReadSubscriptionIdentifier(); if (packet.Properties.SubscriptionIdentifiers == null)
{
packet.Properties.SubscriptionIdentifiers = new List<uint>();
}
packet.Properties.SubscriptionIdentifiers.Add(propertiesReader.ReadSubscriptionIdentifier());
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.ContentType) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.ContentType)
{ {
@@ -538,7 +554,12 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); if (packet.Properties.UserProperties == null)
{
packet.Properties.UserProperties = new List<MqttUserProperty>();
}
propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -581,7 +602,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -619,7 +640,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -657,7 +678,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -695,7 +716,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
@@ -740,7 +761,7 @@ namespace MQTTnet.Formatter.V5
} }
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty) else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{ {
propertiesReader.FillUserProperties(packet.Properties.UserProperties); propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
} }
else else
{ {
Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs
+2 -2
View File
@@ -150,7 +150,7 @@ namespace MQTTnet.Formatter.V5
willPropertiesWriter.WriteTopicAlias(packet.WillMessage.TopicAlias); willPropertiesWriter.WriteTopicAlias(packet.WillMessage.TopicAlias);
willPropertiesWriter.WriteResponseTopic(packet.WillMessage.ResponseTopic); willPropertiesWriter.WriteResponseTopic(packet.WillMessage.ResponseTopic);
willPropertiesWriter.WriteCorrelationData(packet.WillMessage.CorrelationData); willPropertiesWriter.WriteCorrelationData(packet.WillMessage.CorrelationData);
willPropertiesWriter.WriteSubscriptionIdentifier(packet.WillMessage.SubscriptionIdentifier); willPropertiesWriter.WriteSubscriptionIdentifiers(packet.WillMessage.SubscriptionIdentifiers);
willPropertiesWriter.WriteContentType(packet.WillMessage.ContentType); willPropertiesWriter.WriteContentType(packet.WillMessage.ContentType);
willPropertiesWriter.WriteUserProperties(packet.WillMessage.UserProperties); willPropertiesWriter.WriteUserProperties(packet.WillMessage.UserProperties);
@@ -258,7 +258,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteTopicAlias(packet.Properties.TopicAlias); propertiesWriter.WriteTopicAlias(packet.Properties.TopicAlias);
propertiesWriter.WriteResponseTopic(packet.Properties.ResponseTopic); propertiesWriter.WriteResponseTopic(packet.Properties.ResponseTopic);
propertiesWriter.WriteCorrelationData(packet.Properties.CorrelationData); propertiesWriter.WriteCorrelationData(packet.Properties.CorrelationData);
propertiesWriter.WriteSubscriptionIdentifier(packet.Properties.SubscriptionIdentifier); propertiesWriter.WriteSubscriptionIdentifiers(packet.Properties.SubscriptionIdentifiers);
propertiesWriter.WriteContentType(packet.Properties.ContentType); propertiesWriter.WriteContentType(packet.Properties.ContentType);
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties); propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
} }
Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs
+16 -16
View File
@@ -42,7 +42,7 @@ namespace MQTTnet.Formatter.V5
return true; return true;
} }
public void FillUserProperties(List<MqttUserProperty> userProperties) public void AddUserPropertyTo(List<MqttUserProperty> userProperties)
{ {
if (userProperties == null) throw new ArgumentNullException(nameof(userProperties)); if (userProperties == null) throw new ArgumentNullException(nameof(userProperties));
@@ -67,17 +67,17 @@ namespace MQTTnet.Formatter.V5
return _body.ReadWithLengthPrefix(); return _body.ReadWithLengthPrefix();
} }
public bool? ReadRetainAvailable() public bool ReadRetainAvailable()
{ {
return _body.ReadBoolean(); return _body.ReadBoolean();
} }
public uint? ReadSessionExpiryInterval() public uint ReadSessionExpiryInterval()
{ {
return _body.ReadFourByteInteger(); return _body.ReadFourByteInteger();
} }
public ushort? ReadReceiveMaximum() public ushort ReadReceiveMaximum()
{ {
return _body.ReadTwoByteInteger(); return _body.ReadTwoByteInteger();
} }
@@ -92,17 +92,17 @@ namespace MQTTnet.Formatter.V5
return _body.ReadStringWithLengthPrefix(); return _body.ReadStringWithLengthPrefix();
} }
public ushort? ReadTopicAliasMaximum() public ushort ReadTopicAliasMaximum()
{ {
return _body.ReadTwoByteInteger(); return _body.ReadTwoByteInteger();
} }
public uint? ReadMaximumPacketSize() public uint ReadMaximumPacketSize()
{ {
return _body.ReadFourByteInteger(); return _body.ReadFourByteInteger();
} }
public ushort? ReadServerKeepAlive() public ushort ReadServerKeepAlive()
{ {
return _body.ReadTwoByteInteger(); return _body.ReadTwoByteInteger();
} }
@@ -112,22 +112,22 @@ namespace MQTTnet.Formatter.V5
return _body.ReadStringWithLengthPrefix(); return _body.ReadStringWithLengthPrefix();
} }
public bool? ReadSharedSubscriptionAvailable() public bool ReadSharedSubscriptionAvailable()
{ {
return _body.ReadBoolean(); return _body.ReadBoolean();
} }
public bool? ReadSubscriptionIdentifiersAvailable() public bool ReadSubscriptionIdentifiersAvailable()
{ {
return _body.ReadBoolean(); return _body.ReadBoolean();
} }
public bool? ReadWildcardSubscriptionAvailable() public bool ReadWildcardSubscriptionAvailable()
{ {
return _body.ReadBoolean(); return _body.ReadBoolean();
} }
public uint? ReadSubscriptionIdentifier() public uint ReadSubscriptionIdentifier()
{ {
return _body.ReadVariableLengthInteger(); return _body.ReadVariableLengthInteger();
} }
@@ -137,12 +137,12 @@ namespace MQTTnet.Formatter.V5
return (MqttPayloadFormatIndicator)_body.ReadByte(); return (MqttPayloadFormatIndicator)_body.ReadByte();
} }
public uint? ReadMessageExpiryInterval() public uint ReadMessageExpiryInterval()
{ {
return _body.ReadFourByteInteger(); return _body.ReadFourByteInteger();
} }
public ushort? ReadTopicAlias() public ushort ReadTopicAlias()
{ {
return _body.ReadTwoByteInteger(); return _body.ReadTwoByteInteger();
} }
@@ -162,17 +162,17 @@ namespace MQTTnet.Formatter.V5
return _body.ReadStringWithLengthPrefix(); return _body.ReadStringWithLengthPrefix();
} }
public uint? ReadWillDelayInterval() public uint ReadWillDelayInterval()
{ {
return _body.ReadFourByteInteger(); return _body.ReadFourByteInteger();
} }
public bool? RequestResponseInformation() public bool RequestResponseInformation()
{ {
return _body.ReadBoolean(); return _body.ReadBoolean();
} }
public bool? RequestProblemInformation() public bool RequestProblemInformation()
{ {
return _body.ReadBoolean(); return _body.ReadBoolean();
} }
Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs
+13
View File
@@ -80,6 +80,19 @@ namespace MQTTnet.Formatter.V5
WriteAsVariableLengthInteger(MqttPropertyId.SubscriptionIdentifier, value); WriteAsVariableLengthInteger(MqttPropertyId.SubscriptionIdentifier, value);
} }
public void WriteSubscriptionIdentifiers(IEnumerable<uint> value)
{
if (value == null)
{
return;
}
foreach (var subscriptionIdentifier in value)
{
WriteAsVariableLengthInteger(MqttPropertyId.SubscriptionIdentifier, subscriptionIdentifier);
}
}
public void WriteTopicAlias(ushort? value) public void WriteTopicAlias(ushort? value)
{ {
Write(MqttPropertyId.TopicAlias, value); Write(MqttPropertyId.TopicAlias, value);
Source/MQTTnet/MqttApplicationMessage.cs
+1 -1
View File
@@ -28,6 +28,6 @@ namespace MQTTnet
public byte[] CorrelationData { get; set; } public byte[] CorrelationData { get; set; }
public uint? SubscriptionIdentifier { get; set; } public List<uint> SubscriptionIdentifiers { get; set; }
} }
} }
Source/MQTTnet/MqttApplicationMessageBuilder.cs
+11 -10
View File
@@ -18,7 +18,7 @@ namespace MQTTnet
private string _responseTopic; private string _responseTopic;
private byte[] _correlationData; private byte[] _correlationData;
private ushort? _topicAlias; private ushort? _topicAlias;
private uint? _subscriptionIdentifier; private List<uint> _subscriptionIdentifiers;
private uint? _messageExpiryInterval; private uint? _messageExpiryInterval;
private MqttPayloadFormatIndicator? _payloadFormatIndicator; private MqttPayloadFormatIndicator? _payloadFormatIndicator;
private List<MqttUserProperty> _userProperties; private List<MqttUserProperty> _userProperties;
@@ -43,7 +43,7 @@ namespace MQTTnet
{ {
_payload = payload.ToArray(); _payload = payload.ToArray();
} }
return this; return this;
} }
@@ -176,7 +176,12 @@ namespace MQTTnet
/// </summary> /// </summary>
public MqttApplicationMessageBuilder WithSubscriptionIdentifier(uint subscriptionIdentifier) public MqttApplicationMessageBuilder WithSubscriptionIdentifier(uint subscriptionIdentifier)
{ {
_subscriptionIdentifier = subscriptionIdentifier; if (_subscriptionIdentifiers == null)
{
_subscriptionIdentifiers = new List<uint>();
}
_subscriptionIdentifiers.Add(subscriptionIdentifier);
return this; return this;
} }
@@ -215,16 +220,12 @@ namespace MQTTnet
ResponseTopic = _responseTopic, ResponseTopic = _responseTopic,
CorrelationData = _correlationData, CorrelationData = _correlationData,
TopicAlias = _topicAlias, TopicAlias = _topicAlias,
SubscriptionIdentifier = _subscriptionIdentifier, SubscriptionIdentifiers = _subscriptionIdentifiers,
MessageExpiryInterval = _messageExpiryInterval, MessageExpiryInterval = _messageExpiryInterval,
PayloadFormatIndicator = _payloadFormatIndicator PayloadFormatIndicator = _payloadFormatIndicator,
UserProperties = _userProperties
}; };
if (_userProperties?.Any() == true)
{
applicationMessage.UserProperties = _userProperties;
}
return applicationMessage; return applicationMessage;
} }
} }
Source/MQTTnet/Packets/MqttPublishPacketProperties.cs
+2 -2
View File
@@ -15,9 +15,9 @@ namespace MQTTnet.Packets
public byte[] CorrelationData { get; set; } public byte[] CorrelationData { get; set; }
public List<MqttUserProperty> UserProperties { get; } = new List<MqttUserProperty>(); public List<MqttUserProperty> UserProperties { get; set; }
public uint? SubscriptionIdentifier { get; set; } public List<uint> SubscriptionIdentifiers { get; set; }
public string ContentType { get; set; } public string ContentType { get; set; }
} }
Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs
+2 -1
View File
@@ -1,4 +1,5 @@
using System; using System;
using System.Collections.Generic;
using System.IO; using System.IO;
using System.Linq; using System.Linq;
using System.Text; using System.Text;
@@ -234,7 +235,7 @@ namespace MQTTnet.Tests
[TestMethod] [TestMethod]
public void SerializeV500_MqttPublishPacket() public void SerializeV500_MqttPublishPacket()
{ {
var prop = new MqttPublishPacketProperties(); var prop = new MqttPublishPacketProperties {UserProperties = new List<MqttUserProperty>()};
prop.ResponseTopic = "/Response"; prop.ResponseTopic = "/Response";