Add queue with async support to free up threads.

Source/MQTTnet/Internal/AsyncBlockingQueue.cs

@@ -0,0 +1,39 @@
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Internal
+{
+    public sealed class AsyncQueue<TItem> : IDisposable
+    {
+        private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
+        private readonly ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
+
+        public void Enqueue(TItem item)
+        {
+            _queue.Enqueue(item);
+            _semaphore.Release();
+        }
+
+        public async Task<TItem> DequeueAsync(CancellationToken cancellationToken)
+        {
+            while (true)
+            {
+                await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+
+                cancellationToken.ThrowIfCancellationRequested();
+
+                if (_queue.TryDequeue(out var item))
+                {
+                    return item;
+                }
+            }
+        }
+
+        public void Dispose()
+        {
+            _semaphore?.Dispose();
+        }
+    }
+}

Source/MQTTnet/MessageStream/MqttMessageStream.cs

@@ -1,79 +0,0 @@
-using System;
-using System.Collections.Generic;
-
-using System.Runtime.CompilerServices;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Internal;
-using MQTTnet.Packets;
-
-namespace MQTTnet.MessageStream
-{
-    public class MqttMessageStream
-    {
-        private readonly LinkedList<MqttBasePacket> _queue = new LinkedList<MqttBasePacket>();
-        private readonly LinkedList<TaskCompletionSource<MqttBasePacket>> _waitHandles = new LinkedList<TaskCompletionSource<MqttBasePacket>>();
-
-        private readonly SemaphoreSlim _lock = new SemaphoreSlim(1, 1);
-
-        private readonly BlockingQueue<MqttBasePacket> _packets = new BlockingQueue<MqttBasePacket>();
-
-        public void Enqueue(MqttBasePacket packet)
-        {
-            if (packet == null) throw new ArgumentNullException(nameof(packet));
-
-            
-
-            lock (_queue)
-            {
-                //_queue.AddLast(packet);
-
-                _packets.Enqueue(packet);
-            }
-
-            //lock (_queue)
-            //{
-            //    _queue.AddLast(packet);
-
-            //    foreach (var waitHandle in _waitHandles)
-            //    {
-            //        waitHandle.TrySetResult(true);
-            //    }
-
-            //    _waitHandles.Clear();
-            //}
-        }
-        
-        public Task<MqttBasePacket> TakeAsync(CancellationToken cancellationToken)
-        {
-            lock (_packets)
-            {
-                var packet = _packets.Dequeue();
-                return Task.FromResult(packet);
-            }
-            
-
-            //while (!cancellationToken.IsCancellationRequested)
-            //{
-            //    TaskCompletionSource<bool> waitHandle;
-            //    lock (_queue)
-            //    {
-            //        if (_queue.Count > 0)
-            //        {
-            //            var node = _queue.First;
-            //            _queue.RemoveFirst();
-
-            //            return node.Value;
-            //        }
-
-            //        waitHandle = new TaskCompletionSource<bool>();
-            //        _waitHandles.Add(waitHandle);
-            //    }
-
-            //    await waitHandle.Task;
-            //}
-
-            //return null;
-        }
-    }
-}

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -340,7 +340,7 @@ namespace MQTTnet.Server
             return SendAsync(response);
         }
 
-        private async Task HandleIncomingPublishPacketWithQoS2Async(MqttPublishPacket publishPacket)
+        private Task HandleIncomingPublishPacketWithQoS2Async(MqttPublishPacket publishPacket)
         {
             var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket);
             _sessionsManager.DispatchApplicationMessage(applicationMessage, this);
@@ -351,7 +351,7 @@ namespace MQTTnet.Server
                 ReasonCode = MqttPubRecReasonCode.Success
             };
 
-            await SendAsync(pubRecPacket).ConfigureAwait(false);
+            return SendAsync(pubRecPacket);
 
             ////Task.Run(async () =>
             ////{
@@ -374,15 +374,15 @@ namespace MQTTnet.Server
 
         private async Task SendPendingPacketsAsync()
         {
-            MqttPendingApplicationMessage enqueuedApplicationMessage = null;
+            MqttPendingApplicationMessage queuedApplicationMessage = null;
             MqttPublishPacket publishPacket = null;
 
             try
             {
                 while (!_cancellationToken.IsCancellationRequested)
                 {
-                    enqueuedApplicationMessage = await Session.ApplicationMessagesQueue.TakeAsync(_cancellationToken.Token).ConfigureAwait(false);
+                    queuedApplicationMessage = await Session.ApplicationMessagesQueue.TakeAsync(_cancellationToken.Token).ConfigureAwait(false);
-                    if (enqueuedApplicationMessage == null)
+                    if (queuedApplicationMessage == null)
                     {
                         return;
                     }
@@ -392,11 +392,11 @@ namespace MQTTnet.Server
                         return;
                     }
 
-                    publishPacket = _dataConverter.CreatePublishPacket(enqueuedApplicationMessage.ApplicationMessage);
+                    publishPacket = _dataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage);
-                    publishPacket.QualityOfServiceLevel = enqueuedApplicationMessage.QualityOfServiceLevel;
+                    publishPacket.QualityOfServiceLevel = queuedApplicationMessage.QualityOfServiceLevel;
 
                     // Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9].
-                    publishPacket.Retain = enqueuedApplicationMessage.IsRetainedMessage;
+                    publishPacket.Retain = queuedApplicationMessage.IsRetainedMessage;
 
                     if (publishPacket.QualityOfServiceLevel > 0)
                     {
@@ -406,9 +406,9 @@ namespace MQTTnet.Server
                     if (_serverOptions.ClientMessageQueueInterceptor != null)
                     {
                         var context = new MqttClientMessageQueueInterceptorContext(
-                            enqueuedApplicationMessage.SenderClientId,
+                            queuedApplicationMessage.SenderClientId,
                             ClientId,
-                            enqueuedApplicationMessage.ApplicationMessage);
+                            queuedApplicationMessage.ApplicationMessage);
 
                         if (_serverOptions.ClientMessageQueueInterceptor != null)
                         {
@@ -448,7 +448,7 @@ namespace MQTTnet.Server
                         }
                     }
 
-                    _logger.Verbose("Enqueued application message sent (ClientId: {0}).", ClientId);
+                    _logger.Verbose("Queued application message sent (ClientId: {0}).", ClientId);
 
                     // TODO:
                     //Interlocked.Increment(ref _sentPacketsCount);
@@ -475,9 +475,9 @@ namespace MQTTnet.Server
 
                 if (publishPacket?.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
                 {
-                    enqueuedApplicationMessage.IsDuplicate = true;
+                    queuedApplicationMessage.IsDuplicate = true;
 
-                    Session.ApplicationMessagesQueue.Enqueue(enqueuedApplicationMessage);
+                    Session.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage);
                 }
 
                 if (!_cancellationToken.Token.IsCancellationRequested)

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -5,6 +5,7 @@ using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
+using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Server.Status;
@@ -13,7 +14,7 @@ namespace MQTTnet.Server
 {
     public class MqttClientSessionsManager : IDisposable
     {
-        private readonly BlockingCollection<MqttEnqueuedApplicationMessage> _messageQueue = new BlockingCollection<MqttEnqueuedApplicationMessage>();
+        private readonly AsyncQueue<MqttEnqueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttEnqueuedApplicationMessage>();
 
         private readonly SemaphoreSlim _createConnectionGate = new SemaphoreSlim(1, 1);
         private readonly ConcurrentDictionary<string, MqttClientConnection> _connections = new ConcurrentDictionary<string, MqttClientConnection>();
@@ -99,7 +100,7 @@ namespace MQTTnet.Server
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            _messageQueue.Add(new MqttEnqueuedApplicationMessage(applicationMessage, sender), _cancellationToken);
+            _messageQueue.Enqueue(new MqttEnqueuedApplicationMessage(applicationMessage, sender));
         }
 
         public Task SubscribeAsync(string clientId, ICollection<TopicFilter> topicFilters)
@@ -174,17 +175,17 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken);
+                var queuedApplicationMessage = await _messageQueue.DequeueAsync(cancellationToken).ConfigureAwait(false);
 
-                var sender = enqueuedApplicationMessage.Sender;
+                var sender = queuedApplicationMessage.Sender;
-                var applicationMessage = enqueuedApplicationMessage.ApplicationMessage;
+                var applicationMessage = queuedApplicationMessage.ApplicationMessage;
 
                 var interceptorContext = await InterceptApplicationMessageAsync(sender, applicationMessage).ConfigureAwait(false);
                 if (interceptorContext != null)
                 {
                     if (interceptorContext.CloseConnection)
                     {
-                        await enqueuedApplicationMessage.Sender.StopAsync().ConfigureAwait(false);
+                        await queuedApplicationMessage.Sender.StopAsync().ConfigureAwait(false);
                     }
 
                     if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
@@ -205,7 +206,7 @@ namespace MQTTnet.Server
                 foreach (var clientSession in _sessions.Values)
                 {
                     clientSession.EnqueueApplicationMessage(
-                        enqueuedApplicationMessage.ApplicationMessage,
+                        queuedApplicationMessage.ApplicationMessage,
                         sender?.ClientId,
                         false);
                 }

Tests/MQTTnet.Core.Tests/AsyncQueue_Tests.cs

@@ -0,0 +1,51 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Internal;
+
+namespace MQTTnet.Tests
+{
+    [TestClass]
+    public class AsyncQueue_Tests
+    {
+        [TestMethod]
+        public async Task Preserve_Order()
+        {
+            var queue = new AsyncQueue<string>();
+            queue.Enqueue("1");
+            queue.Enqueue("2");
+            queue.Enqueue("3");
+
+            Assert.AreEqual("1", await queue.DequeueAsync(CancellationToken.None));
+            Assert.AreEqual("2", await queue.DequeueAsync(CancellationToken.None));
+            Assert.AreEqual("3", await queue.DequeueAsync(CancellationToken.None));
+        }
+
+        [TestMethod]
+        public async Task Preserve_ProcessAsync()
+        {
+            var queue = new AsyncQueue<int>();
+
+            var sum = 0;
+            var worker = Task.Run(async () => 
+            {
+                while (sum < 6)
+                {
+                    sum += await queue.DequeueAsync(CancellationToken.None);
+                }
+            });
+            
+            queue.Enqueue(1);
+            await Task.Delay(500);
+
+            queue.Enqueue(2);
+            await Task.Delay(500);
+
+            queue.Enqueue(3);
+            await Task.Delay(500);
+
+            Assert.AreEqual(6, sum);
+            Assert.AreEqual(TaskStatus.RanToCompletion, worker.Status);
+        }
+    }
+}