diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 00179800..f9ede290 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -13,11 +13,12 @@ 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] Reduced async tasks count by moving dedicated keep alive tasks per connection to shared one. * [Server] Session takeover and keep alive timeout are now properly set in DISCONNECT packet. -* [Server] Performance improvements. +* [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 diff --git a/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs b/Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs index 21d646ba..7419fae0 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 248d2c32..0b4ac3ad 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -102,6 +102,7 @@ namespace MQTTnet.Formatter.V5 MaximumPacketSize = connAckPacket.Properties?.MaximumPacketSize, ReasonString = connAckPacket.Properties?.ReasonString, ReceiveMaximum = connAckPacket.Properties?.ReceiveMaximum, + 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/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 a6e259db..3f7fe081 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs @@ -10,14 +10,9 @@ namespace MQTTnet.Formatter.V5 { readonly IMqttPacketWriter _packetWriter; - public MqttV500PacketEncoder() - : this(new MqttPacketWriter()) - { - } - public MqttV500PacketEncoder(IMqttPacketWriter packetWriter) { - _packetWriter = packetWriter; + _packetWriter = packetWriter ?? throw new ArgumentNullException(nameof(packetWriter)); } public ArraySegment Encode(MqttBasePacket packet) @@ -201,6 +196,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/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/Formatter/V5/MqttV500PropertiesReader.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs index 443cf177..bb8beace 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs @@ -82,6 +82,17 @@ 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 28931c80..6dd028cd 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs @@ -132,6 +132,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/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index 534b2433..2a132cb2 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -191,10 +191,6 @@ namespace MQTTnet.Implementations } await stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); - - // This subsequent call is required to check whether the socket is still connected. - // Without this call a broken connection is only recognized at the next call. - await stream.WriteAsync(PlatformAbstractionLayer.EmptyByteArray, 0, 0, cancellationToken); } } catch (ObjectDisposedException) diff --git a/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs b/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs index 4cd003e7..d9dba9a2 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; } diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index 3d69e2d5..f3f13350 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -446,7 +446,12 @@ namespace MQTTnet.Server await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false); await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); - await SendAsync(new MqttPubRelPacket { PacketIdentifier = publishPacket.PacketIdentifier }, cancellationToken).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 d7403096..433c2a8e 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; using System.Threading.Tasks; @@ -45,6 +46,8 @@ namespace MQTTnet.Tests.Mockups ServerLogger.LogMessagePublished += (s, e) => { + Debug.WriteLine(e.LogMessage.ToString()); + if (e.LogMessage.Level == MqttNetLogLevel.Error) { lock (_serverErrors) @@ -56,6 +59,8 @@ namespace MQTTnet.Tests.Mockups ClientLogger.LogMessagePublished += (s, e) => { + Debug.WriteLine(e.LogMessage.ToString()); + if (e.LogMessage.Level == MqttNetLogLevel.Error) { lock (_clientErrors) @@ -159,7 +164,7 @@ namespace MQTTnet.Tests.Mockups 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 b1774f74..73ed3f48 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 { @@ -743,5 +747,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); + } + } } }