diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 28ac47c3..2ecfc072 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -11,12 +11,7 @@ false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). -* [Core] Fixed missing properties from PUBLISH packet in _MqttApplicationMessage_ (thanks to @pcbing). -* [Core] Fixed wrong encoding of PUBREL and PUBCOMP packets for MQTTv5 (thanks to @perphilipp). -* [Client] Added the authentication result to the disconnected handler (only set when connecting failed). -* [Client] Added new overloads for _MqttClientOptionsBuilder_. -* [Server] Fixed a bug which returns wrong flag for existing session in CONNACK packet (thanks to @avengerstark). -* [nuget] .NET Framework builds are now using 4.5.2 or 4.6.1 builds instead of netstandard 2.0. +* [Core] Fixed issues in MQTTv5 message encoding and decoding. Copyright Christian Kratky 2016-2019 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 diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 8db55134..8d877f68 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -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 diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs index 8d2adbf8..49cf551b 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs @@ -7,6 +7,8 @@ namespace MQTTnet.Client.Subscribing { public List TopicFilters { get; set; } = new List(); + public uint? SubscriptionIdentifier { get; set; } + public List UserProperties { get; set; } = new List(); } } diff --git a/Source/MQTTnet/Exceptions/MqttUnexpectedDisconnectReceivedException.cs b/Source/MQTTnet/Exceptions/MqttUnexpectedDisconnectReceivedException.cs new file mode 100644 index 00000000..8edb2b2d --- /dev/null +++ b/Source/MQTTnet/Exceptions/MqttUnexpectedDisconnectReceivedException.cs @@ -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 UserProperties { get; } + + public string ServerReference { get; } + } +} diff --git a/Source/MQTTnet/Formatter/IMqttPacketWriter.cs b/Source/MQTTnet/Formatter/IMqttPacketWriter.cs index cce98cf5..5ea6ca34 100644 --- a/Source/MQTTnet/Formatter/IMqttPacketWriter.cs +++ b/Source/MQTTnet/Formatter/IMqttPacketWriter.cs @@ -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); diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs index 3866dd8e..2a14f403 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -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; diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs index 46692cca..fa53c5d8 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs @@ -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(); } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs index acb4202c..89a76bf1 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs @@ -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); } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs index e30cb3f2..a61cb066 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs @@ -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() diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs index a7cb89f3..16532304 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs @@ -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) diff --git a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs index 26b36ec1..ef245e90 100644 --- a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs +++ b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs @@ -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) { diff --git a/Source/MQTTnet/Protocol/MqttPropertyID.cs b/Source/MQTTnet/Protocol/MqttPropertyID.cs index 2e93a158..1fc08ce0 100644 --- a/Source/MQTTnet/Protocol/MqttPropertyID.cs +++ b/Source/MQTTnet/Protocol/MqttPropertyID.cs @@ -9,7 +9,7 @@ CorrelationData = 9, SubscriptionIdentifier = 11, SessionExpiryInterval = 17, - AssignedClientIdentifer = 18, + AssignedClientIdentifier = 18, ServerKeepAlive = 19, AuthenticationMethod = 21, AuthenticationData = 22, diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs index 93fc2a60..52b3d5f6 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs @@ -10,12 +10,47 @@ using MQTTnet.Client.Unsubscribing; using MQTTnet.Formatter; using MQTTnet.Protocol; using MQTTnet.Server; +using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.MQTTv5 { [TestClass] public class Client_Tests { + [TestMethod] + public async Task Connect_With_New_Mqtt_Features() + { + using (var testEnvironment = new TestEnvironment()) + { + await testEnvironment.StartServerAsync(); + + // This test can be also executed against "broker.hivemq.com" to validate package format. + var client = await testEnvironment.ConnectClientAsync( + new MqttClientOptionsBuilder() + //.WithTcpServer("broker.hivemq.com") + .WithTcpServer("127.0.0.1", testEnvironment.ServerPort) + .WithProtocolVersion(MqttProtocolVersion.V500) + .WithTopicAliasMaximum(20) + .WithReceiveMaximum(20) + .WithWillMessage(new MqttApplicationMessageBuilder().WithTopic("abc").Build()) + .WithWillDelayInterval(20) + .Build()); + + await client.SubscribeAsync("a"); + + await client.PublishAsync(new MqttApplicationMessageBuilder() + .WithTopic("a") + .WithPayload("x") + .WithUserProperty("a", "1") + .WithUserProperty("b", "2") + .WithPayloadFormatIndicator(MqttPayloadFormatIndicator.CharacterData) + .WithAtLeastOnceQoS() + .Build()); + + await Task.Delay(500); + } + } + [TestMethod] public async Task Connect() { @@ -63,7 +98,16 @@ namespace MQTTnet.Tests.MQTTv5 await server.StartAsync(new MqttServerOptions()); await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").WithProtocolVersion(MqttProtocolVersion.V500).Build()); - var result = await client.SubscribeAsync("a", MqttQualityOfServiceLevel.AtLeastOnce); + + var result = await client.SubscribeAsync(new MqttClientSubscribeOptions() + { + SubscriptionIdentifier = 1, + TopicFilters = new List + { + new TopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce} + } + }); + await client.DisconnectAsync(); Assert.AreEqual(1, result.Items.Count); diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs new file mode 100644 index 00000000..e3020cea --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs @@ -0,0 +1,40 @@ +using System.Threading; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Client.Options; +using MQTTnet.Formatter; +using MQTTnet.Tests.Mockups; + +namespace MQTTnet.Tests.MQTTv5 +{ + [TestClass] + public class Server_Tests + { + [TestMethod] + public async Task Will_Message_Send() + { + using (var testEnvironment = new TestEnvironment()) + { + var receivedMessagesCount = 0; + + await testEnvironment.StartServerAsync(); + + var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build(); + + var clientOptions = new MqttClientOptionsBuilder().WithWillMessage(willMessage).WithProtocolVersion(MqttProtocolVersion.V500); + + var c1 = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500)); + c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); + await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build()); + + var c2 = await testEnvironment.ConnectClientAsync(clientOptions); + c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent. + + await Task.Delay(1000); + + Assert.AreEqual(1, receivedMessagesCount); + } + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index 776857bf..5cda5374 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -90,12 +90,24 @@ namespace MQTTnet.Tests.Mockups public async Task ConnectClientAsync(MqttClientOptionsBuilder options) { + if (options == null) throw new ArgumentNullException(nameof(options)); + var client = CreateClient(); await client.ConnectAsync(options.WithTcpServer("localhost", ServerPort).Build()); return client; } + public async Task ConnectClientAsync(IMqttClientOptions options) + { + if (options == null) throw new ArgumentNullException(nameof(options)); + + var client = CreateClient(); + await client.ConnectAsync(options); + + return client; + } + public void ThrowIfLogErrors() { lock (_serverErrors)