From 14224e4180bd54dea70a1dcfd226fc26dfa55522 Mon Sep 17 00:00:00 2001 From: Max Gortman Date: Tue, 3 Nov 2020 12:19:38 -0800 Subject: [PATCH 1/6] Add support for Maximum QoS property on CONNACK --- .../Client/Connecting/MqttClientAuthenticateResult.cs | 3 +++ Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs | 1 + Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs | 4 ++++ Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs | 1 + .../MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs | 10 ++++++++++ .../MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs | 11 +++++++++++ Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs | 3 +++ 7 files changed, 33 insertions(+) diff --git a/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs b/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs index 21d646ba..f7ff7253 100644 --- a/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs +++ b/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs @@ -1,5 +1,6 @@ using System.Collections.Generic; using MQTTnet.Packets; +using MQTTnet.Protocol; namespace MQTTnet.Client.Connecting { @@ -25,6 +26,8 @@ namespace MQTTnet.Client.Connecting public string ReasonString { get; set; } public ushort? ReceiveMaximum { get; set; } + + public MqttQualityOfServiceLevel? MaximumQoS { get; set; } public string ResponseInformation { get; set; } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs index f55c3476..b5d7103d 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -93,6 +93,7 @@ namespace MQTTnet.Formatter.V5 MaximumPacketSize = connAckPacket.Properties?.MaximumPacketSize, ReasonString = connAckPacket.Properties?.ReasonString, ReceiveMaximum = connAckPacket.Properties?.ReceiveMaximum, + MaximumQoS = connAckPacket.Properties?.MaximumQoS, ResponseInformation = connAckPacket.Properties?.ResponseInformation, TopicAliasMaximum = connAckPacket.Properties?.TopicAliasMaximum, ServerReference = connAckPacket.Properties?.ServerReference, diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs index cdbceda7..be8d74f4 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs @@ -254,6 +254,10 @@ namespace MQTTnet.Formatter.V5 { packet.Properties.ReceiveMaximum = propertiesReader.ReadReceiveMaximum(); } + else if (propertiesReader.CurrentPropertyId == MqttPropertyId.MaximumQoS) + { + packet.Properties.MaximumQoS = propertiesReader.ReadMaximumQoS(); + } else if (propertiesReader.CurrentPropertyId == MqttPropertyId.AssignedClientIdentifier) { packet.Properties.AssignedClientIdentifier = propertiesReader.ReadAssignedClientIdentifier(); diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs index 9ebc7408..daff40f8 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs @@ -203,6 +203,7 @@ namespace MQTTnet.Formatter.V5 propertiesWriter.WriteAuthenticationData(packet.Properties.AuthenticationData); propertiesWriter.WriteRetainAvailable(packet.Properties.RetainAvailable); propertiesWriter.WriteReceiveMaximum(packet.Properties.ReceiveMaximum); + propertiesWriter.WriteMaximumQoS(packet.Properties.MaximumQoS); propertiesWriter.WriteAssignedClientIdentifier(packet.Properties.AssignedClientIdentifier); propertiesWriter.WriteTopicAliasMaximum(packet.Properties.TopicAliasMaximum); propertiesWriter.WriteReasonString(packet.Properties.ReasonString); diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs index 443cf177..3282dbda 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs @@ -82,6 +82,16 @@ namespace MQTTnet.Formatter.V5 return _body.ReadTwoByteInteger(); } + public MqttQualityOfServiceLevel ReadMaximumQoS() + { + byte value = _body.ReadByte(); + if (value > 1) + { + throw new MqttProtocolViolationException($"Unexpected Maximum QoS value: {value}"); + } + return (MqttQualityOfServiceLevel)value; + } + public string ReadAssignedClientIdentifier() { return _body.ReadStringWithLengthPrefix(); diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs index f632586b..cb1b23ed 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs @@ -133,6 +133,17 @@ namespace MQTTnet.Formatter.V5 Write(MqttPropertyId.ReceiveMaximum, value); } + public void WriteMaximumQoS(MqttQualityOfServiceLevel? value) + { + if (!value.HasValue || value.Value > MqttQualityOfServiceLevel.AtLeastOnce) + { + return; + } + + _packetWriter.Write((byte)MqttPropertyId.MaximumQoS); + _packetWriter.Write((byte)value.Value); + } + public void WriteMaximumPacketSize(uint? value) { WriteAsFourByteInteger(MqttPropertyId.MaximumPacketSize, value); diff --git a/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs b/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs index d59735c6..686f2886 100644 --- a/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs +++ b/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs @@ -1,4 +1,5 @@ using System.Collections.Generic; +using MQTTnet.Protocol; namespace MQTTnet.Packets { @@ -8,6 +9,8 @@ namespace MQTTnet.Packets public ushort? ReceiveMaximum { get; set; } + public MqttQualityOfServiceLevel? MaximumQoS { get; set; } + public bool? RetainAvailable { get; set; } public uint? MaximumPacketSize { get; set; } From 06c9d27c872fa422eee1a8bf6d31301e966a2a6d Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 4 Nov 2020 21:28:40 +0100 Subject: [PATCH 2/6] Fixed bug in PubRel packet generation (MQTTv5 only). --- .../Formatter/V5/MqttV500PacketEncoder.cs | 8 +--- .../Formatter/V5/MqttV500PacketFormatter.cs | 11 ++--- Source/MQTTnet/Server/MqttClientConnection.cs | 7 +++- .../Mockups/TestEnvironment.cs | 17 ++++++++ Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs | 40 +++++++++++++++++++ 5 files changed, 67 insertions(+), 16 deletions(-) diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs index 9ebc7408..976ad12a 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs @@ -10,15 +10,9 @@ namespace MQTTnet.Formatter.V5 { private readonly IMqttPacketWriter _packetWriter; - public MqttV500PacketEncoder() - : this(new MqttPacketWriter()) - { - - } - public MqttV500PacketEncoder(IMqttPacketWriter packetWriter) { - _packetWriter = packetWriter; + _packetWriter = packetWriter ?? throw new ArgumentNullException(nameof(packetWriter)); } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs index 9a52a98d..a34aa536 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs @@ -4,15 +4,10 @@ using MQTTnet.Packets; namespace MQTTnet.Formatter.V5 { - public class MqttV500PacketFormatter : IMqttPacketFormatter + public sealed class MqttV500PacketFormatter : IMqttPacketFormatter { - private readonly MqttV500PacketEncoder _encoder; - private readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder(); - - public MqttV500PacketFormatter() - { - _encoder = new MqttV500PacketEncoder(); - } + readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder(); + readonly MqttV500PacketEncoder _encoder; public MqttV500PacketFormatter(IMqttPacketWriter writer) { diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index d501c8eb..cd304956 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -452,7 +452,12 @@ namespace MQTTnet.Server await SendAsync(publishPacket).ConfigureAwait(false); await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); - await SendAsync(new MqttPubRelPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false); + await SendAsync(new MqttPubRelPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = MqttPubRelReasonCode.Success + }, cancellationToken).ConfigureAwait(false); + await awaiter2.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); } } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index 5800bf43..533d4f0b 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -5,6 +5,7 @@ using MQTTnet.Diagnostics; using MQTTnet.Server; using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Threading.Tasks; @@ -43,6 +44,8 @@ namespace MQTTnet.Tests.Mockups ServerLogger.LogMessagePublished += (s, e) => { + Debug.WriteLine(e.LogMessage.ToString()); + if (e.LogMessage.Level == MqttNetLogLevel.Error) { lock (_serverErrors) @@ -54,6 +57,8 @@ namespace MQTTnet.Tests.Mockups ClientLogger.LogMessagePublished += (s, e) => { + Debug.WriteLine(e.LogMessage.ToString()); + if (e.LogMessage.Level == MqttNetLogLevel.Error) { lock (_clientErrors) @@ -125,6 +130,18 @@ namespace MQTTnet.Tests.Mockups return client; } + public async Task ConnectClientAsync(Action optionsBuilder) + { + var options = new MqttClientOptionsBuilder(); + options = options.WithTcpServer("localhost", ServerPort); + optionsBuilder?.Invoke(options); + + var client = CreateClient(); + await client.ConnectAsync(options.Build()).ConfigureAwait(false); + + return client; + } + public void ThrowIfLogErrors() { lock (_serverErrors) diff --git a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs index 16405de2..893bc9eb 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs @@ -15,6 +15,10 @@ using System.Net.Sockets; using System.Text; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Client.Receiving; +using MQTTnet.Diagnostics; +using MQTTnet.Extensions.ManagedClient; +using MQTTnet.Formatter; namespace MQTTnet.Tests { @@ -741,5 +745,41 @@ namespace MQTTnet.Tests Assert.AreEqual(null, receivedMessage.Payload); } } + + [TestMethod] + public async Task Subscribe_With_QoS2() + { + using (var testEnvironment = new TestEnvironment()) + { + await testEnvironment.StartServerAsync(); + var client1 = await testEnvironment.ConnectClientAsync(o => o.WithProtocolVersion(MqttProtocolVersion.V500)); + var client2 = await testEnvironment.ConnectClientAsync(o => o.WithProtocolVersion(MqttProtocolVersion.V500)); + + var disconnectedFired = false; + client1.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(c => + { + disconnectedFired = true; + }); + + var messageReceived = false; + client1.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(c => + { + messageReceived = true; + }); + + await client1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("topic1").WithExactlyOnceQoS().Build()); + + await Task.Delay(500); + + var message = new MqttApplicationMessageBuilder().WithTopic("topic1").WithPayload("Hello World").WithExactlyOnceQoS().WithRetainFlag().Build(); + + await client2.PublishAsync(message); + await Task.Delay(500); + + Assert.IsTrue(messageReceived); + Assert.IsTrue(client1.IsConnected); + Assert.IsFalse(disconnectedFired); + } + } } } From 26af205ab0aa9510b3952032e17feda2119a04be Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 4 Nov 2020 21:29:26 +0100 Subject: [PATCH 3/6] Update docs. --- Build/MQTTnet.nuspec | 1 + 1 file changed, 1 insertion(+) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 97d0d54b..a1bd0e53 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -15,6 +15,7 @@ * [Core] Added support for TLS 1.3 (requires .NET Core 3.1+) (thanks to @Dvergatal). * [RpcClient] Moved the RPC topic validation to the topic generation strategy. * [RcpClient] Added interface for _MqttRpcClient_. +* [Server] Fixed bug in PubRel packet generation (MQTTv5 only). Copyright Christian Kratky 2016-2020 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor From 344cf674e6426f7a2550e36e2f703dddd337ae6f Mon Sep 17 00:00:00 2001 From: Max Gortman Date: Thu, 5 Nov 2020 16:55:50 -0800 Subject: [PATCH 4/6] PR fixes --- .../MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs | 2 +- Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs | 2 +- Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs | 1 + 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs b/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs index f7ff7253..7419fae0 100644 --- a/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs +++ b/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs @@ -27,7 +27,7 @@ namespace MQTTnet.Client.Connecting public ushort? ReceiveMaximum { get; set; } - public MqttQualityOfServiceLevel? MaximumQoS { get; set; } + public MqttQualityOfServiceLevel MaximumQoS { get; set; } public string ResponseInformation { get; set; } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs index b5d7103d..7f9bd92f 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -93,7 +93,7 @@ namespace MQTTnet.Formatter.V5 MaximumPacketSize = connAckPacket.Properties?.MaximumPacketSize, ReasonString = connAckPacket.Properties?.ReasonString, ReceiveMaximum = connAckPacket.Properties?.ReceiveMaximum, - MaximumQoS = connAckPacket.Properties?.MaximumQoS, + MaximumQoS = connAckPacket.Properties?.MaximumQoS ?? MqttQualityOfServiceLevel.ExactlyOnce, ResponseInformation = connAckPacket.Properties?.ResponseInformation, TopicAliasMaximum = connAckPacket.Properties?.TopicAliasMaximum, ServerReference = connAckPacket.Properties?.ServerReference, diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs index 3282dbda..bb8beace 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs @@ -89,6 +89,7 @@ namespace MQTTnet.Formatter.V5 { throw new MqttProtocolViolationException($"Unexpected Maximum QoS value: {value}"); } + return (MqttQualityOfServiceLevel)value; } From a35dee9352478276feaf41a1b315d5dd1c8946be Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 6 Nov 2020 20:03:33 +0100 Subject: [PATCH 5/6] Update docs. --- Build/MQTTnet.nuspec | 1 + 1 file changed, 1 insertion(+) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index a1bd0e53..18ef9d4b 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -13,6 +13,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol. * [Core] Added support for TLS 1.3 (requires .NET Core 3.1+) (thanks to @Dvergatal). +* [Core] Added support for _MaximumQoS_ transmit when using MQTTv5 (thanks to @nayato). * [RpcClient] Moved the RPC topic validation to the topic generation strategy. * [RcpClient] Added interface for _MqttRpcClient_. * [Server] Fixed bug in PubRel packet generation (MQTTv5 only). From 4608930a37fdda17bffe994be00e3afac81f2385 Mon Sep 17 00:00:00 2001 From: Stefan Ossendorf Date: Sun, 8 Nov 2020 14:31:01 +0100 Subject: [PATCH 6/6] Cancellation token removed since the method does not support it Cancellation token removed because the SendAsync() method does not support it. --- Source/MQTTnet/Server/MqttClientConnection.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index cd304956..23b22257 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -456,7 +456,7 @@ namespace MQTTnet.Server { PacketIdentifier = publishPacket.PacketIdentifier, ReasonCode = MqttPubRelReasonCode.Success - }, cancellationToken).ConfigureAwait(false); + }).ConfigureAwait(false); await awaiter2.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); }