From db1fc7625d9ae5c512bd42747c4aa901db75035b Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Fri, 11 Nov 2022 08:14:15 +0100 Subject: [PATCH] 1546 possible bug in server repeatedly sending the same message to client subscribed with cleansession=false (#1569) --- .github/workflows/ReleaseNotes.md | 3 +- .../IManagedMqttClient.cs | 2 +- .../ManagedMqttClient.cs | 12 +- .../Client/LowLevelMqttClient_Tests.cs | 10 +- .../Client/ManagedMqttClient_Tests.cs | 61 ++++++- .../Internal/MqttPacketBusItem_Tests.cs | 43 +++++ .../Internal/MqttPacketBus_Tests.cs | 168 +++++++++--------- .../TestApplicationMessageReceivedHandler.cs | 29 ++- .../MQTTnet.Tests/Mockups/TestEnvironment.cs | 21 ++- Source/MQTTnet.Tests/Server/QoS_Tests.cs | 51 ++++++ Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 4 +- Source/MQTTnet/Client/MqttClientExtensions.cs | 51 +++--- .../Formatter/ReadFixedHeaderResult.cs | 6 +- Source/MQTTnet/Internal/MqttPacketBusItem.cs | 48 +++-- Source/MQTTnet/Server/Internal/MqttClient.cs | 8 +- Source/MQTTnet/Server/Internal/MqttSession.cs | 50 +++++- Source/MQTTnet/Server/MqttServerExtensions.cs | 78 +++++++- .../Server/Status/MqttSessionStatus.cs | 2 +- 18 files changed, 475 insertions(+), 172 deletions(-) create mode 100644 Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 608f1589..ace201e0 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,7 +1,8 @@ * [Core] Fixed several dead locks by implementing a new async signal (#1552) * [Client] All waiting tasks will now fail when the client connection gets closed. This fixes several freezes (#1561) -* [ManagedClient] The topic is no longer declared as invalid if a topic alias is used (#1395). +* [ManagedClient] The topic is no longer declared as invalid if a topic alias is used (#1395) * [Server] Fixed duplicated invocation of the event _ClientAcknowledgedPublishPacketAsync_ for QoS level 2 (#1550) * [Server] Fixed issue in upgrading and downgrading of QoS levels for subscriptions and retained messages (#1560) * [Server] Fixed memory leak when old sessions are discarded (#1553) * [Server] When using MQTTv5 the reason code _NoMatchingSubscribers_ is returned when no client is subscribed to a published topic (#802) +* [Server] Fix sending of duplicate messages with QoS > 0 when session is recovered and session is reused (#1546) diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs index 20f51d83..49add6a3 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs @@ -43,7 +43,7 @@ namespace MQTTnet.Extensions.ManagedClient Task StartAsync(ManagedMqttClientOptions options); - Task StopAsync(); + Task StopAsync(bool cleanDisconnect = true); Task SubscribeAsync(ICollection topicFilters); diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index c979627c..0c04c663 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -62,6 +62,7 @@ namespace MQTTnet.Extensions.ManagedClient CancellationTokenSource _publishingCancellationToken; ManagedMqttClientStorageManager _storageManager; + bool _isCleanDisconnect; public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger) { @@ -255,10 +256,12 @@ namespace MQTTnet.Extensions.ManagedClient _logger.Info("Started"); } - public async Task StopAsync() + public async Task StopAsync(bool cleanDisconnect = true) { ThrowIfDisposed(); + _isCleanDisconnect = cleanDisconnect; + StopPublishing(); StopMaintainingConnection(); @@ -382,9 +385,12 @@ namespace MQTTnet.Extensions.ManagedClient { try { - using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout)) + if (_isCleanDisconnect) { - await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false); + using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout)) + { + await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false); + } } } catch (OperationCanceledException) diff --git a/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs b/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs index ae8fcebe..e7366d3e 100644 --- a/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs @@ -64,7 +64,7 @@ namespace MQTTnet.Tests.Client await client.ConnectAsync(options, CancellationToken.None).ConfigureAwait(false); } - + [TestMethod] [ExpectedException(typeof(MqttCommunicationException))] public async Task Connect_To_Wrong_Host() @@ -80,6 +80,8 @@ namespace MQTTnet.Tests.Client { using (var testEnvironment = CreateTestEnvironment()) { + testEnvironment.IgnoreServerLogErrors = true; + testEnvironment.ServerPort = 8364; var server = await testEnvironment.StartServer(); @@ -116,6 +118,8 @@ namespace MQTTnet.Tests.Client { using (var testEnvironment = CreateTestEnvironment()) { + testEnvironment.IgnoreServerLogErrors = true; + var server = await testEnvironment.StartServer(); using (var lowLevelClient = testEnvironment.CreateLowLevelClient()) @@ -123,7 +127,7 @@ namespace MQTTnet.Tests.Client Assert.IsFalse(lowLevelClient.IsConnected); var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).WithTimeout(TimeSpan.FromSeconds(1)).Build(); - + await lowLevelClient.ConnectAsync(clientOptions, CancellationToken.None); Assert.IsTrue(lowLevelClient.IsConnected); @@ -148,7 +152,7 @@ namespace MQTTnet.Tests.Client catch { } - + Assert.IsFalse(lowLevelClient.IsConnected); } } diff --git a/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs b/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs index 554cc663..bfa662ac 100644 --- a/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs @@ -11,6 +11,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Extensions.ManagedClient; +using MQTTnet.Formatter; using MQTTnet.Internal; using MQTTnet.Protocol; using MQTTnet.Server; @@ -21,6 +22,65 @@ namespace MQTTnet.Tests.Client [TestClass] public sealed class ManagedMqttClient_Tests : BaseTestClass { + [TestMethod] + public async Task Receive_While_Not_Cleanly_Disconnected() + { + using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500)) + { + await testEnvironment.StartServer(o => o.WithPersistentSessions()); + + var senderClient = await testEnvironment.ConnectClient(); + + // Prepare managed client. + var managedClient = testEnvironment.Factory.CreateManagedMqttClient(); + await managedClient.SubscribeAsync("#"); + var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient); + + var managedClientOptions = new ManagedMqttClientOptions + { + ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder() + .WithTcpServer("127.0.0.1", testEnvironment.ServerPort) + .WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed") + .WithCleanSession(false) + .Build() + }; + + await managedClient.StartAsync(managedClientOptions); + await LongTestDelay(); + + // Send test data. + await senderClient.PublishStringAsync("topic1"); + await LongTestDelay(); + + receivedMessages.AssertReceivedCountEquals(1); + + // Stop the managed client but keep session at server (not clean disconnect required). + await managedClient.StopAsync(false); + await LongTestDelay(); + + // Send new messages in the meantime. + await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce); + await LongTestDelay(); + + // Start the managed client, it should receive the new message. + await managedClient.StartAsync(managedClientOptions); + await LongTestDelay(); + + receivedMessages.AssertReceivedCountEquals(2); + + // Stop and start again, no new message should be received. + for (var i = 0; i < 3; i++) + { + await managedClient.StopAsync(false); + await LongTestDelay(); + await managedClient.StartAsync(managedClientOptions); + await LongTestDelay(); + } + + receivedMessages.AssertReceivedCountEquals(2); + } + } + [TestMethod] public async Task Connect_To_Invalid_Server() { @@ -33,7 +93,6 @@ namespace MQTTnet.Tests.Client var managedClient = testEnvironment.Factory.CreateManagedMqttClient(mqttClient); ConnectingFailedEventArgs connectingFailedEventArgs = null; - managedClient.ConnectingFailedAsync += args => { connectingFailedEventArgs = args; diff --git a/Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs b/Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs new file mode 100644 index 00000000..912d38a0 --- /dev/null +++ b/Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs @@ -0,0 +1,43 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Internal; +using MQTTnet.Packets; + +namespace MQTTnet.Tests.Internal +{ + [TestClass] + public sealed class MqttPacketBusItem_Tests + { + [TestMethod] + public void Fire_Completed_Event() + { + var eventFired = false; + + var item = new MqttPacketBusItem(new MqttPublishPacket()); + item.Completed += (_, __) => + { + eventFired = true; + }; + + item.Complete(); + + Assert.IsTrue(eventFired); + } + + [TestMethod] + [ExpectedException(typeof(TaskCanceledException))] + public async Task Wait_Packet_Bus_Item_After_Already_Canceled() + { + var item = new MqttPacketBusItem(new MqttPublishPacket()); + + // Finish the item before the actual + item.Cancel(); + + await item.WaitAsync(); + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs b/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs index b09159fd..40ff02aa 100644 --- a/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs +++ b/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs @@ -13,69 +13,26 @@ namespace MQTTnet.Tests.Internal { [TestClass] public sealed class MqttPacketBus_Tests - { - [TestMethod] - [ExpectedException(typeof(TaskCanceledException))] - public async Task Wait_With_Empty_Bus() - { - var bus = new MqttPacketBus(); - - using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1))) - { - await bus.DequeueItemAsync(timeout.Token); - } - } - - [TestMethod] - public async Task Fill_From_Different_Task() - { - const int MessageCount = 500; - - var delayRandom = new Random(); - - var bus = new MqttPacketBus(); - - _ = Task.Run( - () => - { - for (int i = 0; i < MessageCount; i++) - { - bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health); - - Thread.Sleep(delayRandom.Next(0, 10)); - } - }); - - for (var i = 0; i < MessageCount; i++) - { - using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10))) - { - await bus.DequeueItemAsync(timeout.Token); - } - } - - Assert.AreEqual(0, bus.TotalItemsCount); - } - + { [TestMethod] public void Alternate_Priorities() { var bus = new MqttPacketBus(); - + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - + bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); - + bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); Assert.AreEqual(9, bus.TotalItemsCount); - + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); @@ -85,33 +42,10 @@ namespace MQTTnet.Tests.Internal Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); - + Assert.AreEqual(0, bus.TotalItemsCount); } - - [TestMethod] - public void Export_Packets_Without_Dequeue() - { - var bus = new MqttPacketBus(); - bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - - Assert.AreEqual(3, bus.TotalItemsCount); - - var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control); - Assert.AreEqual(0, exportedPackets.Count); - - exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health); - Assert.AreEqual(0, exportedPackets.Count); - - exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data); - Assert.AreEqual(3, exportedPackets.Count); - - Assert.AreEqual(3, bus.TotalItemsCount); - } - [TestMethod] public void Await_Single_Packet() { @@ -121,31 +55,97 @@ namespace MQTTnet.Tests.Internal var item1 = new MqttPacketBusItem(new MqttPublishPacket()); var item2 = new MqttPacketBusItem(new MqttPublishPacket()); - + var item3 = new MqttPacketBusItem(new MqttPublishPacket()); - item3.Delivered += (_, __) => + item3.Completed += (_, __) => { delivered = true; }; - + bus.EnqueueItem(item1, MqttPacketBusPartition.Data); bus.EnqueueItem(item2, MqttPacketBusPartition.Data); bus.EnqueueItem(item3, MqttPacketBusPartition.Data); - + Assert.IsFalse(delivered); - bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered(); - + bus.DequeueItemAsync(CancellationToken.None).Result.Complete(); + Assert.IsFalse(delivered); - - bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered(); - + + bus.DequeueItemAsync(CancellationToken.None).Result.Complete(); + Assert.IsFalse(delivered); - - bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered(); - + + bus.DequeueItemAsync(CancellationToken.None).Result.Complete(); + // The third packet has the event attached. Assert.IsTrue(delivered); } + + [TestMethod] + public void Export_Packets_Without_Dequeue() + { + var bus = new MqttPacketBus(); + + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + + Assert.AreEqual(3, bus.TotalItemsCount); + + var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control); + Assert.AreEqual(0, exportedPackets.Count); + + exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health); + Assert.AreEqual(0, exportedPackets.Count); + + exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data); + Assert.AreEqual(3, exportedPackets.Count); + + Assert.AreEqual(3, bus.TotalItemsCount); + } + + [TestMethod] + public async Task Fill_From_Different_Task() + { + const int MessageCount = 500; + + var delayRandom = new Random(); + + var bus = new MqttPacketBus(); + + _ = Task.Run( + () => + { + for (var i = 0; i < MessageCount; i++) + { + bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health); + + Thread.Sleep(delayRandom.Next(0, 10)); + } + }); + + for (var i = 0; i < MessageCount; i++) + { + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10))) + { + await bus.DequeueItemAsync(timeout.Token); + } + } + + Assert.AreEqual(0, bus.TotalItemsCount); + } + + [TestMethod] + [ExpectedException(typeof(OperationCanceledException))] + public async Task Wait_With_Empty_Bus() + { + var bus = new MqttPacketBus(); + + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1))) + { + await bus.DequeueItemAsync(timeout.Token); + } + } } } \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs b/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs index c1a7c482..7556709a 100644 --- a/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs +++ b/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs @@ -2,8 +2,10 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System; using System.Collections.Generic; using System.Linq; +using System.Text; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; @@ -17,7 +19,12 @@ namespace MQTTnet.Tests.Mockups public TestApplicationMessageReceivedHandler(IMqttClient mqttClient) { - mqttClient.ApplicationMessageReceivedAsync += MqttClientOnApplicationMessageReceivedAsync; + if (mqttClient == null) + { + throw new ArgumentNullException(nameof(mqttClient)); + } + + mqttClient.ApplicationMessageReceivedAsync += OnApplicationMessageReceivedAsync; } public List ReceivedEventArgs @@ -39,7 +46,25 @@ namespace MQTTnet.Tests.Mockups } } - Task MqttClientOnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs) + public string GeneratePayloadSequence() + { + var sequence = new StringBuilder(); + + lock (_receivedEventArgs) + { + foreach (var receivedEventArg in _receivedEventArgs) + { + var payload = receivedEventArg.ApplicationMessage.ConvertPayloadToString(); + + // An empty payload is not part of the sequence! + sequence.Append(payload); + } + } + + return sequence.ToString(); + } + + Task OnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs) { lock (_receivedEventArgs) { diff --git a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs index 6b79a963..44d692d9 100644 --- a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs +++ b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs @@ -11,6 +11,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Diagnostics; +using MQTTnet.Extensions.ManagedClient; using MQTTnet.Extensions.Rpc; using MQTTnet.Formatter; using MQTTnet.Internal; @@ -109,7 +110,7 @@ namespace MQTTnet.Tests.Mockups configureOptions.Invoke(optionsBuilder); var options = optionsBuilder.Build(); - + var client = CreateClient(); if (timeout == TimeSpan.Zero) @@ -137,7 +138,7 @@ namespace MQTTnet.Tests.Mockups options = options.WithTcpServer("127.0.0.1", ServerPort); var client = CreateClient(); - + if (timeout == TimeSpan.Zero) { await client.ConnectAsync(options.Build()).ConfigureAwait(false); @@ -161,7 +162,7 @@ namespace MQTTnet.Tests.Mockups } var client = CreateClient(); - + if (timeout == TimeSpan.Zero) { await client.ConnectAsync(options).ConfigureAwait(false); @@ -199,6 +200,16 @@ namespace MQTTnet.Tests.Mockups return new TestApplicationMessageReceivedHandler(mqttClient); } + public TestApplicationMessageReceivedHandler CreateApplicationMessageHandler(IManagedMqttClient managedClient) + { + if (managedClient == null) + { + throw new ArgumentNullException(nameof(managedClient)); + } + + return new TestApplicationMessageReceivedHandler(managedClient.InternalClient); + } + public IMqttClient CreateClient() { lock (_clients) @@ -325,7 +336,7 @@ namespace MQTTnet.Tests.Mockups var options = optionsBuilder.Build(); var server = CreateServer(options); await server.StartAsync(); - + // The OS has chosen the port to we have to properly expose it to the tests. ServerPort = options.DefaultEndpointOptions.Port; return server; @@ -344,7 +355,7 @@ namespace MQTTnet.Tests.Mockups var options = optionsBuilder.Build(); var server = CreateServer(options); await server.StartAsync(); - + // The OS has chosen the port to we have to properly expose it to the tests. ServerPort = options.DefaultEndpointOptions.Port; return server; diff --git a/Source/MQTTnet.Tests/Server/QoS_Tests.cs b/Source/MQTTnet.Tests/Server/QoS_Tests.cs index 7e5a9491..d6c9e644 100644 --- a/Source/MQTTnet.Tests/Server/QoS_Tests.cs +++ b/Source/MQTTnet.Tests/Server/QoS_Tests.cs @@ -15,6 +15,57 @@ namespace MQTTnet.Tests.Server [TestClass] public sealed class QoS_Tests : BaseTestClass { + [TestMethod] + public async Task Preserve_Message_Order_For_Queued_Messages() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(o => o.WithPersistentSessions()); + + // Create a session which will contain the messages. + var dummyClient = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false)); + await dummyClient.SubscribeAsync("#", MqttQualityOfServiceLevel.AtLeastOnce); + dummyClient.Dispose(); + + // Now inject messages which are appended to the queue of the client. + await server.InjectApplicationMessage("T", "0", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + + await server.InjectApplicationMessage("T", "2", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + await server.InjectApplicationMessage("T", "1", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + + await server.InjectApplicationMessage("T", "4", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + await server.InjectApplicationMessage("T", "3", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + + await server.InjectApplicationMessage("T", "6", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + await server.InjectApplicationMessage("T", "5", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + + await server.InjectApplicationMessage("T", "8", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + await server.InjectApplicationMessage("T", "7", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + + await server.InjectApplicationMessage("T", "9", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + + await LongTestDelay(); + + // Create a new client for the existing message. + var client = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false)); + var messages = testEnvironment.CreateApplicationMessageHandler(client); + + await LongTestDelay(); + + var payloadSequence = messages.GeneratePayloadSequence(); + Assert.AreEqual("0214365879", payloadSequence); + + // Disconnect and reconnect to make sure that the server will not send the messages twice. + await client.DisconnectAsync(); + await LongTestDelay(); + await client.ReconnectAsync(); + await LongTestDelay(); + + payloadSequence = messages.GeneratePayloadSequence(); + Assert.AreEqual("0214365879", payloadSequence); + } + } + [TestMethod] public async Task Fire_Event_On_Client_Acknowledges_QoS_0() { diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 4b030150..308d5477 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -86,7 +86,7 @@ namespace MQTTnet.Adapter await Task.WhenAny(connectTask, timeout.Task).ConfigureAwait(false); if (timeout.Task.IsCompleted && !connectTask.IsCompleted) { - throw new OperationCanceledException("MQTT connect cancelled.", cancellationToken); + throw new OperationCanceledException("MQTT connect canceled.", cancellationToken); } // Make sure that the exception from the connect task gets thrown. @@ -317,7 +317,7 @@ namespace MQTTnet.Adapter if (cancellationToken.IsCancellationRequested) { - return ReadFixedHeaderResult.Cancelled; + return ReadFixedHeaderResult.Canceled; } if (bytesRead == 0) diff --git a/Source/MQTTnet/Client/MqttClientExtensions.cs b/Source/MQTTnet/Client/MqttClientExtensions.cs index 9f936bdd..ba917aad 100644 --- a/Source/MQTTnet/Client/MqttClientExtensions.cs +++ b/Source/MQTTnet/Client/MqttClientExtensions.cs @@ -14,26 +14,6 @@ namespace MQTTnet.Client { public static class MqttClientExtensions { - public static async Task TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default) - { - if (client == null) - { - throw new ArgumentNullException(nameof(client)); - } - - try - { - await client.PingAsync(cancellationToken).ConfigureAwait(false); - return true; - } - catch - { - // Ignore errors. - } - - return false; - } - public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reason = MqttClientDisconnectReason.NormalDisconnection, string reasonString = null) { if (client == null) @@ -89,6 +69,17 @@ namespace MQTTnet.Client return mqttClient.PublishBinaryAsync(topic, payloadBuffer, qualityOfServiceLevel, retain, cancellationToken); } + public static Task ReconnectAsync(this IMqttClient client, CancellationToken cancellationToken = default) + { + if (client.Options == null) + { + throw new InvalidOperationException( + "The MQTT client was not connected before. A reconnect is only permitted when the client was already connected or at least tried to."); + } + + return client.ConnectAsync(client.Options, cancellationToken); + } + public static Task SendExtendedAuthenticationExchangeDataAsync(this IMqttClient client, MqttExtendedAuthenticationExchangeData data) { if (client == null) @@ -137,6 +128,26 @@ namespace MQTTnet.Client return mqttClient.SubscribeAsync(subscribeOptions, cancellationToken); } + public static async Task TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default) + { + if (client == null) + { + throw new ArgumentNullException(nameof(client)); + } + + try + { + await client.PingAsync(cancellationToken).ConfigureAwait(false); + return true; + } + catch + { + // Ignore errors. + } + + return false; + } + public static Task UnsubscribeAsync(this IMqttClient mqttClient, string topic, CancellationToken cancellationToken = default) { if (mqttClient == null) diff --git a/Source/MQTTnet/Formatter/ReadFixedHeaderResult.cs b/Source/MQTTnet/Formatter/ReadFixedHeaderResult.cs index 58663933..a471a6a7 100644 --- a/Source/MQTTnet/Formatter/ReadFixedHeaderResult.cs +++ b/Source/MQTTnet/Formatter/ReadFixedHeaderResult.cs @@ -6,9 +6,9 @@ namespace MQTTnet.Formatter { public struct ReadFixedHeaderResult { - public static ReadFixedHeaderResult Cancelled { get; } = new ReadFixedHeaderResult + public static ReadFixedHeaderResult Canceled { get; } = new ReadFixedHeaderResult { - IsCancelled = true + IsCanceled = true }; public static ReadFixedHeaderResult ConnectionClosed { get; } = new ReadFixedHeaderResult @@ -16,7 +16,7 @@ namespace MQTTnet.Formatter IsConnectionClosed = true }; - public bool IsCancelled { get; set; } + public bool IsCanceled { get; set; } public bool IsConnectionClosed { get; set; } diff --git a/Source/MQTTnet/Internal/MqttPacketBusItem.cs b/Source/MQTTnet/Internal/MqttPacketBusItem.cs index 91224533..b94654cf 100644 --- a/Source/MQTTnet/Internal/MqttPacketBusItem.cs +++ b/Source/MQTTnet/Internal/MqttPacketBusItem.cs @@ -10,40 +10,36 @@ namespace MQTTnet.Internal { public sealed class MqttPacketBusItem { - readonly AsyncTaskCompletionSource _promise = new AsyncTaskCompletionSource(); - + readonly AsyncTaskCompletionSource _promise = new AsyncTaskCompletionSource(); + public MqttPacketBusItem(MqttPacket packet) { Packet = packet ?? throw new ArgumentNullException(nameof(packet)); } - + + public event EventHandler Completed; + public MqttPacket Packet { get; } - public event EventHandler Delivered; - - public Task WaitForDeliveryAsync() - { - return _promise.Task; - } - - public void MarkAsDelivered() - { - if (_promise.TrySetResult(0)) - { - Delivered?.Invoke(this, EventArgs.Empty); - } - } - - public void MarkAsFailed(Exception exception) - { - if (exception == null) throw new ArgumentNullException(nameof(exception)); - - _promise.TrySetException(exception); - } - - public void MarkAsCancelled() + public void Cancel() { _promise.TrySetCanceled(); } + + public void Complete() + { + _promise.TrySetResult(true); + Completed?.Invoke(this, EventArgs.Empty); + } + + public void Fail(Exception exception) + { + _promise.TrySetException(exception); + } + + public Task WaitAsync() + { + return _promise.Task; + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index f9f7e695..f505b00c 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -507,15 +507,15 @@ namespace MQTTnet.Server try { await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false); - packetBusItem.MarkAsDelivered(); + packetBusItem.Complete(); } catch (OperationCanceledException) { - packetBusItem.MarkAsCancelled(); + packetBusItem.Cancel(); } catch (Exception exception) { - packetBusItem.MarkAsFailed(exception); + packetBusItem.Fail(exception); } finally { @@ -563,7 +563,7 @@ namespace MQTTnet.Server catch (ObjectDisposedException) { // This can happen when connections are created and dropped very quickly. - // It is not an issue if the cancellation token cannot be cancelled multiple times. + // It is not an issue if the cancellation token cannot be canceled multiple times. } } diff --git a/Source/MQTTnet/Server/Internal/MqttSession.cs b/Source/MQTTnet/Server/Internal/MqttSession.cs index a5a045bf..b40c2645 100644 --- a/Source/MQTTnet/Server/Internal/MqttSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttSession.cs @@ -4,7 +4,6 @@ using System; using System.Collections; -using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -25,7 +24,8 @@ namespace MQTTnet.Server readonly MqttServerOptions _serverOptions; readonly MqttClientSubscriptionsManager _subscriptionsManager; - readonly ConcurrentDictionary _unacknowledgedPublishPackets = new ConcurrentDictionary(); + // Do not use a dictionary in order to keep the ordering of the messages. + readonly List _unacknowledgedPublishPackets = new List(); // Bookkeeping to know if this is a subscribing client; lazy initialize later. HashSet _subscribedTopics; @@ -72,7 +72,14 @@ namespace MQTTnet.Server public MqttPublishPacket AcknowledgePublishPacket(ushort packetIdentifier) { - _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out var publishPacket); + MqttPublishPacket publishPacket; + + lock (_unacknowledgedPublishPackets) + { + publishPacket = _unacknowledgedPublishPackets.FirstOrDefault(p => p.PacketIdentifier.Equals(packetIdentifier)); + _unacknowledgedPublishPackets.Remove(publishPacket); + } + return publishPacket; } @@ -130,7 +137,10 @@ namespace MQTTnet.Server { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); - _unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket; + lock (_unacknowledgedPublishPackets) + { + _unacknowledgedPublishPackets.Add(publishPacket); + } } _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data); @@ -145,8 +155,10 @@ namespace MQTTnet.Server { // This will only return the matching PUBLISH packet but does not remove it. // This is required for QoS 2. - _unacknowledgedPublishPackets.TryGetValue(packetIdentifier, out var publishPacket); - return publishPacket; + lock (_unacknowledgedPublishPackets) + { + return _unacknowledgedPublishPackets.FirstOrDefault(p => p.PacketIdentifier.Equals(packetIdentifier)); + } } public void Recover() @@ -154,9 +166,33 @@ namespace MQTTnet.Server // TODO: Keep the bus and only insert pending items again. // TODO: Check if packet identifier must be restarted or not. // TODO: Recover package identifier. + + /* + The Session state in the Client consists of: + · QoS 1 and QoS 2 messages which have been sent to the Server, but have not been completely acknowledged. + · QoS 2 messages which have been received from the Server, but have not been completely acknowledged. + + The Session state in the Server consists of: + · The existence of a Session, even if the rest of the Session state is empty. + · The Client’s subscriptions. + · QoS 1 and QoS 2 messages which have been sent to the Client, but have not been completely acknowledged. + · QoS 1 and QoS 2 messages pending transmission to the Client. + · QoS 2 messages which have been received from the Client, but have not been completely acknowledged. + · Optionally, QoS 0 messages pending transmission to the Client. + */ + + // Create a copy of all currently unacknowledged publish packets and clear the storage. + // We must re-enqueue them in order to trigger other code. + List unacknowledgedPublishPackets; + lock (_unacknowledgedPublishPackets) + { + unacknowledgedPublishPackets = _unacknowledgedPublishPackets.ToList(); + _unacknowledgedPublishPackets.Clear(); + } + _packetBus.Clear(); - foreach (var publishPacket in _unacknowledgedPublishPackets.Values.ToList()) + foreach (var publishPacket in unacknowledgedPublishPackets) { EnqueueDataPacket(new MqttPacketBusItem(publishPacket)); } diff --git a/Source/MQTTnet/Server/MqttServerExtensions.cs b/Source/MQTTnet/Server/MqttServerExtensions.cs index 22d8538c..13746e50 100644 --- a/Source/MQTTnet/Server/MqttServerExtensions.cs +++ b/Source/MQTTnet/Server/MqttServerExtensions.cs @@ -3,29 +3,89 @@ // See the LICENSE file in the project root for more information. using System; +using System.Text; using System.Threading.Tasks; +using MQTTnet.Internal; using MQTTnet.Packets; +using MQTTnet.Protocol; namespace MQTTnet.Server { public static class MqttServerExtensions { + public static Task InjectApplicationMessage( + this MqttServer server, + string topic, + string payload = null, + MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce, + bool retain = false) + { + if (server == null) + { + throw new ArgumentNullException(nameof(server)); + } + + if (topic == null) + { + throw new ArgumentNullException(nameof(topic)); + } + + var payloadBuffer = EmptyBuffer.Array; + if (payload is string stringPayload) + { + payloadBuffer = Encoding.UTF8.GetBytes(stringPayload); + } + + return server.InjectApplicationMessage( + new InjectedMqttApplicationMessage( + new MqttApplicationMessage + { + Topic = topic, + Payload = payloadBuffer, + QualityOfServiceLevel = qualityOfServiceLevel, + Retain = retain + })); + } + public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters) { - if (server == null) throw new ArgumentNullException(nameof(server)); - if (clientId == null) throw new ArgumentNullException(nameof(clientId)); - if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); + if (server == null) + { + throw new ArgumentNullException(nameof(server)); + } + + if (clientId == null) + { + throw new ArgumentNullException(nameof(clientId)); + } + + if (topicFilters == null) + { + throw new ArgumentNullException(nameof(topicFilters)); + } return server.SubscribeAsync(clientId, topicFilters); } - + public static Task SubscribeAsync(this MqttServer server, string clientId, string topic) { - if (server == null) throw new ArgumentNullException(nameof(server)); - if (clientId == null) throw new ArgumentNullException(nameof(clientId)); - if (topic == null) throw new ArgumentNullException(nameof(topic)); + if (server == null) + { + throw new ArgumentNullException(nameof(server)); + } - return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build()); + if (clientId == null) + { + throw new ArgumentNullException(nameof(clientId)); + } + + if (topic == null) + { + throw new ArgumentNullException(nameof(topic)); + } + + var topicFilters = new MqttTopicFilterBuilder().WithTopic(topic).Build(); + return server.SubscribeAsync(clientId, topicFilters); } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index 6cea318b..c6df1da5 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -45,7 +45,7 @@ namespace MQTTnet.Server var packetBusItem = new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage)); _session.EnqueueDataPacket(packetBusItem); - return packetBusItem.WaitForDeliveryAsync(); + return packetBusItem.WaitAsync(); } public Task DeleteAsync()