diff --git a/Source/MQTTnet/Internal/AsyncBlockingQueue.cs b/Source/MQTTnet/Internal/AsyncBlockingQueue.cs new file mode 100644 index 00000000..678be116 --- /dev/null +++ b/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 : IDisposable + { + private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0); + private readonly ConcurrentQueue _queue = new ConcurrentQueue(); + + public void Enqueue(TItem item) + { + _queue.Enqueue(item); + _semaphore.Release(); + } + + public async Task 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(); + } + } +} diff --git a/Source/MQTTnet/MessageStream/MqttMessageStream.cs b/Source/MQTTnet/MessageStream/MqttMessageStream.cs deleted file mode 100644 index ef881473..00000000 --- a/Source/MQTTnet/MessageStream/MqttMessageStream.cs +++ /dev/null @@ -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 _queue = new LinkedList(); - private readonly LinkedList> _waitHandles = new LinkedList>(); - - private readonly SemaphoreSlim _lock = new SemaphoreSlim(1, 1); - - private readonly BlockingQueue _packets = new BlockingQueue(); - - 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 TakeAsync(CancellationToken cancellationToken) - { - lock (_packets) - { - var packet = _packets.Dequeue(); - return Task.FromResult(packet); - } - - - //while (!cancellationToken.IsCancellationRequested) - //{ - // TaskCompletionSource waitHandle; - // lock (_queue) - // { - // if (_queue.Count > 0) - // { - // var node = _queue.First; - // _queue.RemoveFirst(); - - // return node.Value; - // } - - // waitHandle = new TaskCompletionSource(); - // _waitHandles.Add(waitHandle); - // } - - // await waitHandle.Task; - //} - - //return null; - } - } -} diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index 4981b06a..4b75ff8c 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/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); - if (enqueuedApplicationMessage == null) + queuedApplicationMessage = await Session.ApplicationMessagesQueue.TakeAsync(_cancellationToken.Token).ConfigureAwait(false); + if (queuedApplicationMessage == null) { return; } @@ -392,11 +392,11 @@ namespace MQTTnet.Server return; } - publishPacket = _dataConverter.CreatePublishPacket(enqueuedApplicationMessage.ApplicationMessage); - publishPacket.QualityOfServiceLevel = enqueuedApplicationMessage.QualityOfServiceLevel; + publishPacket = _dataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage); + 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) diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 07640d94..a88d315c 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/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 _messageQueue = new BlockingCollection(); + private readonly AsyncQueue _messageQueue = new AsyncQueue(); private readonly SemaphoreSlim _createConnectionGate = new SemaphoreSlim(1, 1); private readonly ConcurrentDictionary _connections = new ConcurrentDictionary(); @@ -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 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 applicationMessage = enqueuedApplicationMessage.ApplicationMessage; + var sender = queuedApplicationMessage.Sender; + 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); } diff --git a/Tests/MQTTnet.Core.Tests/AsyncQueue_Tests.cs b/Tests/MQTTnet.Core.Tests/AsyncQueue_Tests.cs new file mode 100644 index 00000000..ebe3b191 --- /dev/null +++ b/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(); + 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(); + + 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); + } + } +}