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

Fix issues in MQTTv5 packet encoding and decoding.

Browse Source
This commit is contained in:
Christian Kratky
2019-05-12 00:18:48 +02:00
parent ac88f0a05a
commit 12973a4179
15 changed files with 243 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+3
View File
@@ -469,6 +469,9 @@ namespace MQTTnet.Client
}
else if (packet is MqttDisconnectPacket)
{
// Also dispatch disconnect to waiting threads to generate a proper exception.
_packetDispatcher.Dispatch(packet);
await DisconnectAsync(null, cancellationToken).ConfigureAwait(false);
}
else
Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs
+2
View File
@@ -7,6 +7,8 @@ namespace MQTTnet.Client.Subscribing
{
public List<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>();
public uint? SubscriptionIdentifier { get; set; }
public List<MqttUserProperty> UserProperties { get; set; } = new List<MqttUserProperty>();
}
}
Source/MQTTnet/Exceptions/MqttUnexpectedDisconnectReceivedException.cs
+29
View File
@@ -0,0 +1,29 @@
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Exceptions
{
public class MqttUnexpectedDisconnectReceivedException : MqttCommunicationException
{
public MqttUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket)
: base($"Unexpected DISCONNECT (Reason code={disconnectPacket.ReasonCode}) received.")
{
ReasonCode = disconnectPacket.ReasonCode;
SessionExpiryInterval = disconnectPacket.Properties?.SessionExpiryInterval;
ReasonString = disconnectPacket.Properties?.ReasonString;
ServerReference = disconnectPacket.Properties?.ServerReference;
UserProperties = disconnectPacket.Properties?.UserProperties;
}
public MqttDisconnectReasonCode? ReasonCode { get; }
public uint? SessionExpiryInterval { get; }
public string ReasonString { get; }
public List<MqttUserProperty> UserProperties { get; }
public string ServerReference { get; }
}
}
Source/MQTTnet/Formatter/IMqttPacketWriter.cs
-1
View File
@@ -8,7 +8,6 @@
void Write(byte returnCode);
void WriteWithLengthPrefix(byte[] payload);
void Write(ushort keepAlivePeriod);
void Write(IMqttPacketWriter propertyWriter);
void WriteVariableLengthInteger(uint length);
void Write(byte[] payload, int v, int length);
Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs
+1
View File
@@ -158,6 +158,7 @@ namespace MQTTnet.Formatter.V5
};
packet.TopicFilters.AddRange(options.TopicFilters);
packet.Properties.SubscriptionIdentifier = options.SubscriptionIdentifier;
packet.Properties.UserProperties.AddRange(options.UserProperties);
return packet;
Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs
+49 -6
View File
@@ -91,10 +91,6 @@ namespace MQTTnet.Formatter.V5
{
packet.Properties.AuthenticationMethod = propertiesReader.ReadAuthenticationMethod();
}
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.WillDelayInterval)
{
packet.Properties.WillDelayInterval = propertiesReader.ReadWillDelayInterval();
}
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.AuthenticationData)
{
packet.Properties.AuthenticationData = propertiesReader.ReadAuthenticationData();
@@ -125,7 +121,7 @@ namespace MQTTnet.Formatter.V5
}
else
{
propertiesReader.ThrowInvalidPropertyIdException(typeof(MqttConnAckPacket));
propertiesReader.ThrowInvalidPropertyIdException(typeof(MqttConnectPacket));
}
}
@@ -133,6 +129,53 @@ namespace MQTTnet.Formatter.V5
if (packet.WillMessage != null)
{
var willPropertiesReader = new MqttV500PropertiesReader(body);
while (willPropertiesReader.MoveNext())
{
if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.PayloadFormatIndicator)
{
packet.WillMessage.PayloadFormatIndicator = propertiesReader.ReadPayloadFormatIndicator();
}
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.MessageExpiryInterval)
{
packet.WillMessage.MessageExpiryInterval = propertiesReader.ReadMessageExpiryInterval();
}
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.TopicAlias)
{
packet.WillMessage.TopicAlias = propertiesReader.ReadTopicAlias();
}
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.ResponseTopic)
{
packet.WillMessage.ResponseTopic = propertiesReader.ReadResponseTopic();
}
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.CorrelationData)
{
packet.WillMessage.CorrelationData = propertiesReader.ReadCorrelationData();
}
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier)
{
packet.WillMessage.SubscriptionIdentifier = propertiesReader.ReadSubscriptionIdentifier();
}
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.ContentType)
{
packet.WillMessage.ContentType = propertiesReader.ReadContentType();
}
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.WillDelayInterval)
{
// This is a special case!
packet.Properties.WillDelayInterval = propertiesReader.ReadWillDelayInterval();
}
else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
{
propertiesReader.FillUserProperties(packet.Properties.UserProperties);
}
else
{
propertiesReader.ThrowInvalidPropertyIdException(typeof(MqttPublishPacket));
}
}
packet.WillMessage.Topic = body.ReadStringWithLengthPrefix();
packet.WillMessage.Payload = body.ReadWithLengthPrefix();
}
@@ -186,7 +229,7 @@ namespace MQTTnet.Formatter.V5
{
packet.Properties.ReceiveMaximum = propertiesReader.ReadReceiveMaximum();
}
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.AssignedClientIdentifer)
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.AssignedClientIdentifier)
{
packet.Properties.AssignedClientIdentifier = propertiesReader.ReadAssignedClientIdentifier();
}
Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs
+30 -16
View File
@@ -127,24 +127,38 @@ namespace MQTTnet.Formatter.V5
var propertiesWriter = new MqttV500PropertiesWriter();
if (packet.Properties != null)
{
propertiesWriter.WriteWillDelayInterval(packet.Properties.WillDelayInterval);
propertiesWriter.WriteSessionExpiryInterval(packet.Properties.SessionExpiryInterval);
propertiesWriter.WriteAuthenticationMethod(packet.Properties.AuthenticationMethod);
propertiesWriter.WriteAuthenticationData(packet.Properties.AuthenticationData);
propertiesWriter.WriteRequestProblemInformation(packet.Properties.RequestProblemInformation);
propertiesWriter.WriteRequestResponseInformation(packet.Properties.RequestResponseInformation);
propertiesWriter.WriteReceiveMaximum(packet.Properties.ReceiveMaximum);
propertiesWriter.WriteTopicAlias(packet.Properties.TopicAliasMaximum);
propertiesWriter.WriteTopicAliasMaximum(packet.Properties.TopicAliasMaximum);
propertiesWriter.WriteMaximumPacketSize(packet.Properties.MaximumPacketSize);
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
}
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
packetWriter.WriteWithLengthPrefix(packet.ClientId);
if (packet.WillMessage != null)
{
var willPropertiesWriter = new MqttV500PropertiesWriter();
willPropertiesWriter.WritePayloadFormatIndicator(packet.WillMessage.PayloadFormatIndicator);
willPropertiesWriter.WriteMessageExpiryInterval(packet.WillMessage.MessageExpiryInterval);
willPropertiesWriter.WriteTopicAlias(packet.WillMessage.TopicAlias);
willPropertiesWriter.WriteResponseTopic(packet.WillMessage.ResponseTopic);
willPropertiesWriter.WriteCorrelationData(packet.WillMessage.CorrelationData);
willPropertiesWriter.WriteSubscriptionIdentifier(packet.WillMessage.SubscriptionIdentifier);
willPropertiesWriter.WriteContentType(packet.WillMessage.ContentType);
willPropertiesWriter.WriteUserProperties(packet.WillMessage.UserProperties);
// This is a special case!
willPropertiesWriter.WriteWillDelayInterval(packet.Properties?.WillDelayInterval);
willPropertiesWriter.WriteTo(packetWriter);
packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic);
packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload);
}
@@ -202,7 +216,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
}
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
}
@@ -249,7 +263,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
}
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
if (packet.Payload?.Length > 0)
{
@@ -300,7 +314,7 @@ namespace MQTTnet.Formatter.V5
if (packetWriter.Length > 0 || packet.ReasonCode.Value != MqttPubAckReasonCode.Success)
{
packetWriter.Write((byte)packet.ReasonCode.Value);
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
}
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
@@ -328,7 +342,7 @@ namespace MQTTnet.Formatter.V5
if (packetWriter.Length > 0 || packet.ReasonCode.Value != MqttPubRecReasonCode.Success)
{
packetWriter.Write((byte)packet.ReasonCode.Value);
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
}
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
@@ -355,7 +369,7 @@ namespace MQTTnet.Formatter.V5
if (propertiesWriter.Length > 0 || packet.ReasonCode.Value != MqttPubRelReasonCode.Success)
{
packetWriter.Write((byte)packet.ReasonCode.Value);
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
}
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
@@ -382,7 +396,7 @@ namespace MQTTnet.Formatter.V5
if (propertiesWriter.Length > 0 || packet.ReasonCode.Value != MqttPubCompReasonCode.Success)
{
packetWriter.Write((byte)packet.ReasonCode.Value);
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
}
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
@@ -403,7 +417,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
}
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
if (packet.TopicFilters?.Count > 0)
{
@@ -450,7 +464,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
}
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
foreach (var reasonCode in packet.ReasonCodes)
{
@@ -474,7 +488,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
}
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
foreach (var topicFilter in packet.TopicFilters)
{
@@ -499,7 +513,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
}
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
foreach (var reasonCode in packet.ReasonCodes)
{
@@ -527,7 +541,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
}
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Disconnect);
}
@@ -555,7 +569,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
}
propertiesWriter.WriteToPacket(packetWriter);
propertiesWriter.WriteTo(packetWriter);
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Auth);
}
Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs
+3 -14
View File
@@ -1,6 +1,5 @@
using System;
using System.Collections.Generic;
using System.Linq;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -47,20 +46,10 @@ namespace MQTTnet.Formatter.V5
{
if (userProperties == null) throw new ArgumentNullException(nameof(userProperties));
var userPropertiesLength = _body.ReadVariableLengthInteger();
if (userPropertiesLength == 0)
{
return;
}
var name = _body.ReadStringWithLengthPrefix();
var value = _body.ReadStringWithLengthPrefix();
var targetPosition = _body.Offset + userPropertiesLength;
while (_body.Offset < targetPosition)
{
var name = _body.ReadStringWithLengthPrefix();
var value = _body.ReadStringWithLengthPrefix();
userProperties.Add(new MqttUserProperty(name, value));
}
userProperties.Add(new MqttUserProperty(name, value));
}
public string ReadReasonString()
Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs
+17 -10
View File
@@ -19,16 +19,12 @@ namespace MQTTnet.Formatter.V5
return;
}
var propertyWriter = new MqttPacketWriter();
foreach (var property in userProperties)
{
propertyWriter.WriteWithLengthPrefix(property.Name);
propertyWriter.WriteWithLengthPrefix(property.Value);
_packetWriter.Write((byte)MqttPropertyId.UserProperty);
_packetWriter.WriteWithLengthPrefix(property.Name);
_packetWriter.WriteWithLengthPrefix(property.Value);
}
_packetWriter.Write((byte)MqttPropertyId.UserProperty);
_packetWriter.WriteVariableLengthInteger((uint)propertyWriter.Length);
_packetWriter.Write(propertyWriter);
}
public void WriteCorrelationData(byte[] value)
@@ -66,7 +62,7 @@ namespace MQTTnet.Formatter.V5
Write(MqttPropertyId.AuthenticationMethod, value);
}
public void WriteToPacket(IMqttPacketWriter packetWriter)
public void WriteTo(IMqttPacketWriter packetWriter)
{
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -121,7 +117,7 @@ namespace MQTTnet.Formatter.V5
public void WriteReceiveMaximum(ushort? value)
{
Write(MqttPropertyId.RequestResponseInformation, value);
Write(MqttPropertyId.ReceiveMaximum, value);
}
public void WriteMaximumPacketSize(uint? value)
@@ -136,7 +132,7 @@ namespace MQTTnet.Formatter.V5
public void WriteAssignedClientIdentifier(string value)
{
Write(MqttPropertyId.AssignedClientIdentifer, value);
Write(MqttPropertyId.AssignedClientIdentifier, value);
}
public void WriteTopicAliasMaximum(ushort? value)
@@ -180,6 +176,17 @@ namespace MQTTnet.Formatter.V5
_packetWriter.Write(value.Value ? (byte)0x1 : (byte)0x0);
}
private void Write(MqttPropertyId id, byte? value)
{
if (!value.HasValue)
{
return;
}
_packetWriter.Write((byte)id);
_packetWriter.Write(value.Value);
}
private void Write(MqttPropertyId id, ushort? value)
{
if (!value.HasValue)
Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs
+10
View File
@@ -23,6 +23,16 @@ namespace MQTTnet.PacketDispatcher
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packet is MqttDisconnectPacket disconnectPacket)
{
foreach (var packetAwaiter in _packetAwaiters)
{
packetAwaiter.Value.Fail(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
}
return;
}
ushort identifier = 0;
if (packet is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier.HasValue)
{
Source/MQTTnet/Protocol/MqttPropertyID.cs
+1 -1
View File
@@ -9,7 +9,7 @@
CorrelationData = 9,
SubscriptionIdentifier = 11,
SessionExpiryInterval = 17,
AssignedClientIdentifer = 18,
AssignedClientIdentifier = 18,
ServerKeepAlive = 19,
AuthenticationMethod = 21,
AuthenticationData = 22,