diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index da658a7..71f17db 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -12,6 +12,8 @@ MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). * [Server] Added new method which exposes all retained messages. * [Server] Removed (wrong) setter from the server options interface. +* [ManagedClient] Added max pending messages count option. +* [ManagedClient] Add pending messages overflow strategy option. Copyright Christian Kratky 2016-2018 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/README.md b/README.md index fc2acea..cb405dd 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov * Performance optimized (processing ~70.000 messages / second)* * Interfaces included for mocking and testing * Access to internal trace messages -* Unit tested (~90 tests) +* Unit tested (~100 tests) \* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_. diff --git a/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedEventArgs.cs b/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedEventArgs.cs new file mode 100644 index 0000000..0f1613d --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedEventArgs.cs @@ -0,0 +1,14 @@ +using System; + +namespace MQTTnet.Extensions.ManagedClient +{ + public class ApplicationMessageSkippedEventArgs : EventArgs + { + public ApplicationMessageSkippedEventArgs(ManagedMqttApplicationMessage applicationMessage) + { + ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); + } + + public ManagedMqttApplicationMessage ApplicationMessage { get; } + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs index b5a6c50..d003584 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs @@ -15,6 +15,7 @@ namespace MQTTnet.Extensions.ManagedClient event EventHandler Disconnected; event EventHandler ApplicationMessageProcessed; + event EventHandler ApplicationMessageSkipped; event EventHandler ConnectingFailed; event EventHandler SynchronizingSubscriptionsFailed; diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs index 8d48bab..f59aca7 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs @@ -1,5 +1,6 @@ using System; using MQTTnet.Client; +using MQTTnet.Server; namespace MQTTnet.Extensions.ManagedClient { @@ -12,5 +13,9 @@ namespace MQTTnet.Extensions.ManagedClient TimeSpan ConnectionCheckInterval { get; } IManagedMqttClientStorage Storage { get; } + + int MaxPendingMessages { get; } + + MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 2b28deb..16d8e9b 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -7,13 +6,17 @@ using System.Threading.Tasks; using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; +using MQTTnet.Internal; using MQTTnet.Protocol; +using MQTTnet.Server; +using MqttClientConnectedEventArgs = MQTTnet.Client.MqttClientConnectedEventArgs; +using MqttClientDisconnectedEventArgs = MQTTnet.Client.MqttClientDisconnectedEventArgs; namespace MQTTnet.Extensions.ManagedClient { public class ManagedMqttClient : IManagedMqttClient { - private readonly BlockingCollection _messageQueue = new BlockingCollection(); + private readonly BlockingQueue _messageQueue = new BlockingQueue(); private readonly Dictionary _subscriptions = new Dictionary(); private readonly HashSet _unsubscriptions = new HashSet(); @@ -50,6 +53,7 @@ namespace MQTTnet.Extensions.ManagedClient public event EventHandler ApplicationMessageReceived; public event EventHandler ApplicationMessageProcessed; + public event EventHandler ApplicationMessageSkipped; public event EventHandler ConnectingFailed; public event EventHandler SynchronizingSubscriptionsFailed; @@ -75,7 +79,7 @@ namespace MQTTnet.Extensions.ManagedClient foreach (var message in messages) { - _messageQueue.Add(message); + _messageQueue.Enqueue(message); } } @@ -93,10 +97,7 @@ namespace MQTTnet.Extensions.ManagedClient StopPublishing(); StopMaintainingConnection(); - while (_messageQueue.Any()) - { - _messageQueue.Take(); - } + _messageQueue.Clear(); return Task.FromResult(0); } @@ -112,12 +113,38 @@ namespace MQTTnet.Extensions.ManagedClient { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + ManagedMqttApplicationMessage skippedMessage = null; + lock (_messageQueue) + { + if (_messageQueue.Count > _options.MaxPendingMessages) + { + if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage) + { + _logger.Verbose("Skipping publish of new application message because internal queue is full."); + ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(applicationMessage)); + return; + } + + if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage) + { + skippedMessage = _messageQueue.RemoveFirst(); + _logger.Verbose("Removed oldest application message from internal queue because it is full."); + ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(skippedMessage)); + } + } + + _messageQueue.Enqueue(applicationMessage); + } + if (_storageManager != null) { + if (skippedMessage != null) + { + await _storageManager.RemoveAsync(skippedMessage).ConfigureAwait(false); + } + await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false); } - - _messageQueue.Add(applicationMessage); } public Task SubscribeAsync(IEnumerable topicFilters) @@ -157,7 +184,6 @@ namespace MQTTnet.Extensions.ManagedClient public void Dispose() { - _messageQueue?.Dispose(); _connectionCancellationToken?.Dispose(); _publishingCancellationToken?.Dispose(); } @@ -228,7 +254,7 @@ namespace MQTTnet.Extensions.ManagedClient { while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected) { - var message = _messageQueue.Take(cancellationToken); + var message = _messageQueue.Dequeue(); if (message == null) { continue; @@ -268,7 +294,7 @@ namespace MQTTnet.Extensions.ManagedClient if (message.ApplicationMessage.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { - _messageQueue.Add(message); + _messageQueue.Enqueue(message); } } catch (Exception exception) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs index cecf227..cba801b 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs @@ -1,5 +1,6 @@ using System; using MQTTnet.Client; +using MQTTnet.Server; namespace MQTTnet.Extensions.ManagedClient { @@ -12,5 +13,9 @@ namespace MQTTnet.Extensions.ManagedClient public TimeSpan ConnectionCheckInterval { get; set; } = TimeSpan.FromSeconds(1); public IManagedMqttClientStorage Storage { get; set; } + + public int MaxPendingMessages { get; set; } = int.MaxValue; + + public MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; set; } = MqttPendingMessagesOverflowStrategy.DropNewMessage; } } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs index ef774c5..3ea5b05 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs @@ -1,5 +1,6 @@ using System; using MQTTnet.Client; +using MQTTnet.Server; namespace MQTTnet.Extensions.ManagedClient { @@ -8,6 +9,18 @@ namespace MQTTnet.Extensions.ManagedClient private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions(); private MqttClientOptionsBuilder _clientOptionsBuilder; + public ManagedMqttClientOptionsBuilder WithMaxPendingMessages(int value) + { + _options.MaxPendingMessages = value; + return this; + } + + public ManagedMqttClientOptionsBuilder WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy value) + { + _options.PendingMessagesOverflowStrategy = value; + return this; + } + public ManagedMqttClientOptionsBuilder WithAutoReconnectDelay(TimeSpan value) { _options.AutoReconnectDelay = value; diff --git a/Source/MQTTnet/Internal/BlockingQueue.cs b/Source/MQTTnet/Internal/BlockingQueue.cs new file mode 100644 index 0000000..ae6cc98 --- /dev/null +++ b/Source/MQTTnet/Internal/BlockingQueue.cs @@ -0,0 +1,77 @@ +using System; +using System.Collections.Generic; +using System.Threading; + +namespace MQTTnet.Internal +{ + public class BlockingQueue + { + private readonly object _syncRoot = new object(); + private readonly LinkedList _items = new LinkedList(); + private readonly ManualResetEvent _gate = new ManualResetEvent(false); + + public int Count + { + get + { + lock (_syncRoot) + { + return _items.Count; + } + } + } + + public void Enqueue(TItem item) + { + if (item == null) throw new ArgumentNullException(nameof(item)); + + lock (_syncRoot) + { + _items.AddLast(item); + _gate.Set(); + } + } + + public TItem Dequeue() + { + while (true) + { + lock (_syncRoot) + { + if (_items.Count > 0) + { + var item = _items.First.Value; + _items.RemoveFirst(); + + return item; + } + + if (_items.Count == 0) + { + _gate.Reset(); + } + } + + _gate.WaitOne(); + } + } + + public TItem RemoveFirst() + { + lock (_syncRoot) + { + var item = _items.First; + _items.RemoveFirst(); + return item.Value; + } + } + + public void Clear() + { + lock (_syncRoot) + { + _items.Clear(); + } + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/BlockingQueueTests.cs b/Tests/MQTTnet.Core.Tests/BlockingQueueTests.cs new file mode 100644 index 0000000..80e0d72 --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/BlockingQueueTests.cs @@ -0,0 +1,108 @@ +using System.Threading; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Internal; + +namespace MQTTnet.Core.Tests +{ + [TestClass] + public class BlockingQueueTests + { + [TestMethod] + public void Preserve_Order() + { + var queue = new BlockingQueue(); + queue.Enqueue("a"); + queue.Enqueue("b"); + queue.Enqueue("c"); + + Assert.AreEqual(3, queue.Count); + + Assert.AreEqual("a", queue.Dequeue()); + Assert.AreEqual("b", queue.Dequeue()); + Assert.AreEqual("c", queue.Dequeue()); + } + + [TestMethod] + public void Remove_First_Items() + { + var queue = new BlockingQueue(); + queue.Enqueue("a"); + queue.Enqueue("b"); + queue.Enqueue("c"); + + Assert.AreEqual("a", queue.RemoveFirst()); + Assert.AreEqual("b", queue.RemoveFirst()); + + Assert.AreEqual(1, queue.Count); + + Assert.AreEqual("c", queue.Dequeue()); + } + + [TestMethod] + public void Clear_Items() + { + var queue = new BlockingQueue(); + queue.Enqueue("a"); + queue.Enqueue("b"); + queue.Enqueue("c"); + + Assert.AreEqual(3, queue.Count); + + queue.Clear(); + + Assert.AreEqual(0, queue.Count); + } + + [TestMethod] + public async Task Wait_For_Item() + { + var queue = new BlockingQueue(); + + string item = null; +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Run(() => + { + item = queue.Dequeue(); + }); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + + await Task.Delay(100); + + Assert.AreEqual(0, queue.Count); + Assert.AreEqual(null, item); + + queue.Enqueue("x"); + + await Task.Delay(100); + + Assert.AreEqual("x", item); + Assert.AreEqual(0, queue.Count); + } + + [TestMethod] + public void Wait_For_Times() + { + var number = 0; + + var queue = new BlockingQueue(); + +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Run(async () => + { + while (true) + { + queue.Enqueue(1); + await Task.Delay(100); + } + }); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + + while (number < 50) + { + queue.Dequeue(); + Interlocked.Increment(ref number); + } + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj index 20cd045..2baaa28 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj +++ b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj @@ -13,6 +13,7 @@ + diff --git a/Tests/MQTTnet.Core.Tests/ManagedMqttClientTests.cs b/Tests/MQTTnet.Core.Tests/ManagedMqttClientTests.cs new file mode 100644 index 0000000..670b8a1 --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/ManagedMqttClientTests.cs @@ -0,0 +1,38 @@ +using MQTTnet.Extensions.ManagedClient; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Server; + +namespace MQTTnet.Core.Tests +{ + [TestClass] + public class ManagedMqttClientTests + { + [TestMethod] + public async Task Drop_New_Messages_On_Full_Queue() + { + var factory = new MqttFactory(); + var managedClient = factory.CreateManagedMqttClient(); + + var clientOptions = new ManagedMqttClientOptionsBuilder() + .WithMaxPendingMessages(5) + .WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy.DropNewMessage); + + clientOptions.WithClientOptions(o => o.WithTcpServer("localhost")); + + await managedClient.StartAsync(clientOptions.Build()); + + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "1" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "2" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "3" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "4" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "5" }); + + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "6" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "7" }); + await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "8" }); + + Assert.AreEqual(5, managedClient.PendingApplicationMessagesCount); + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index fc40377..b37dfdf 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -340,6 +340,8 @@ namespace MQTTnet.Core.Tests } }); + await Task.Delay(100); + var retainedMessages = server.GetRetainedMessages(); Assert.AreEqual(ClientCount, retainedMessages.Count); diff --git a/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs index 7b5ad77..c405f5f 100644 --- a/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs +++ b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs @@ -45,7 +45,9 @@ namespace MQTTnet.Core.Tests stopwatch.Stop(); times.Add(stopwatch.Elapsed); stopwatch.Restart(); - } + } + + await server.StopAsync(); } } }