Introduce asynchronous message processing

Build/MQTTnet.nuspec

@@ -17,6 +17,8 @@
 * [LowLevelMqttClient] Added low level MQTT client in order to provide more flexibility when using the MQTT protocol. This client requires detailed knowledge about the MQTT protocol.
 * [Client] Improve connection stability (thanks to @jltjohanlindqvist).
 * [Client] Support WithConnectionUri to configure client (thanks to @PMExtra).
+* [Client] Support PublishAsync with QoS 1 and QoS 2 from within an ApplicationMessageReceivedHandler (#648, #587, thanks to @PSanetra).
+* [Client] Fixed MqttCommunicationTimedOutExceptions, caused by a long running ApplicationMessageReceivedHandler, which blocked MQTT packets from being processed (#829, thanks to @PSanetra).
 * [ManagedClient] Added builder class for MqttClientUnsubscribeOptions (thanks to @dominikviererbe).
 * [ManagedClient] Added support for persisted sessions (thansk to @PMExtra).
 * [ManagedClient] Fixed a memory leak (thanks to @zawodskoj).

Source/MQTTnet/Client/MqttClient.cs

@@ -34,6 +34,9 @@ public class MqttClient : Disposable, IMqttClient
         private CancellationTokenSource _backgroundCancellationTokenSource;
         private Task _packetReceiverTask;
         private Task _keepAlivePacketsSenderTask;
+        private Task _publishPacketReceiverTask;
+
+        private AsyncQueue<MqttPublishPacket> _publishPacketReceiverQueue;
 
         private IMqttChannelAdapter _adapter;
         private bool _cleanDisconnectInitiated;
@@ -88,6 +91,9 @@ public async Task<MqttClientAuthenticateResult> ConnectAsync(IMqttClientOptions
                     await _adapter.ConnectAsync(options.CommunicationTimeout, combined.Token).ConfigureAwait(false);
                     _logger.Verbose("Connection with server established.");
 
+                    _publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>();
+                    _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken);
+
                     _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken);
 
                     authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, combined.Token).ConfigureAwait(false);
@@ -230,6 +236,9 @@ private void Cleanup()
             _backgroundCancellationTokenSource?.Dispose();
             _backgroundCancellationTokenSource = null;
 
+            _publishPacketReceiverQueue?.Dispose();
+            _publishPacketReceiverQueue = null;
+
             _adapter?.Dispose();
             _adapter = null;
         }
@@ -300,9 +309,12 @@ private async Task DisconnectInternalAsync(Task sender, Exception exception, Mqt
             try
             {
                 var receiverTask = WaitForTaskAsync(_packetReceiverTask, sender);
+                var publishPacketReceiverTask = WaitForTaskAsync(_publishPacketReceiverTask, sender);
                 var keepAliveTask = WaitForTaskAsync(_keepAlivePacketsSenderTask, sender);
 
-                await Task.WhenAll(receiverTask, keepAliveTask).ConfigureAwait(false);
+                await Task.WhenAll(receiverTask, publishPacketReceiverTask, keepAliveTask).ConfigureAwait(false);
+
+                _publishPacketReceiverQueue.Dispose();
             }
             catch (Exception e)
             {
@@ -522,7 +534,7 @@ private async Task TryProcessReceivedPacketAsync(MqttBasePacket packet, Cancella
 
                 if (packet is MqttPublishPacket publishPacket)
                 {
-                    await TryProcessReceivedPublishPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false);
+                    EnqueueReceivedPublishPacket(publishPacket);
                 }
                 else if (packet is MqttPubRelPacket pubRelPacket)
                 {
@@ -584,47 +596,71 @@ await SendAsync(new MqttPubCompPacket
             }
         }
 
-        private async Task TryProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        private void EnqueueReceivedPublishPacket(MqttPublishPacket publishPacket)
         {
             try
             {
-                if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
-                {
-                    await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
-                }
-                else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
+                _publishPacketReceiverQueue.Enqueue(publishPacket);
+            }
+            catch (Exception exception)
+            {
+                _logger.Error(exception, "Error while enqueueing application message.");
+            }
+        }
+
+        private async Task ProcessReceivedPublishPackets(CancellationToken cancellationToken)
+        {
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                try
                 {
-                    if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
+                    var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken);
+
+                    if (!publishPacketDequeueResult.IsSuccess)
+                    {
+                        return;
+                    }
+
+                    var publishPacket = publishPacketDequeueResult.Item;
+
+                    if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
                     {
-                        await SendAsync(new MqttPubAckPacket
+                        await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
+                    }
+                    else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
+                    {
+                        if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
                         {
-                            PacketIdentifier = publishPacket.PacketIdentifier,
-                            ReasonCode = MqttPubAckReasonCode.Success
-                        }, cancellationToken).ConfigureAwait(false);
+                            await SendAsync(new MqttPubAckPacket
+                            {
+                                PacketIdentifier = publishPacket.PacketIdentifier,
+                                ReasonCode = MqttPubAckReasonCode.Success
+                            }, cancellationToken).ConfigureAwait(false);
+                        }
                     }
-                }
-                else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
-                {
-                    if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
+                    else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
                     {
-                        var pubRecPacket = new MqttPubRecPacket
+                        if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
                         {
-                            PacketIdentifier = publishPacket.PacketIdentifier,
-                            ReasonCode = MqttPubRecReasonCode.Success
-                        };
+                            var pubRecPacket = new MqttPubRecPacket
+                            {
+                                PacketIdentifier = publishPacket.PacketIdentifier,
+                                ReasonCode = MqttPubRecReasonCode.Success
+                            };
 
-                        await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
+                            await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
+                        }
+                    }
+                    else
+                    {
+                        throw new MqttProtocolViolationException("Received a not supported QoS level.");
                     }
                 }
-                else
+                catch (Exception exception)
                 {
-                    throw new MqttProtocolViolationException("Received a not supported QoS level.");
+                    _logger.Error(exception, "Error while handling application message.");
                 }
             }
-            catch (Exception exception)
-            {
-                _logger.Error(exception, "Error while handling application message.");
-            }
         }
 
         private async Task<MqttClientPublishResult> PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken)

Source/MQTTnet/Internal/AsyncBlockingQueue.cs → Source/MQTTnet/Internal/AsyncQueue.cs

@@ -23,9 +23,15 @@ public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationTo
         {
             while (!cancellationToken.IsCancellationRequested)
             {
-                await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
-
-                cancellationToken.ThrowIfCancellationRequested();
+                try
+                {
+                    await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+                    cancellationToken.ThrowIfCancellationRequested();
+                }
+                catch (OperationCanceledException)
+                {
+                    return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
+                }
 
                 if (_queue.TryDequeue(out var item))
                 {

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -23,6 +23,8 @@ public sealed class MqttClientConnection : IDisposable
         readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();
 
         readonly IMqttRetainedMessagesManager _retainedMessagesManager;
+        readonly Func<Task> _onStart;
+        readonly Func<MqttClientDisconnectType, Task> _onStop;
         readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
         readonly MqttClientSessionsManager _sessionsManager;
 
@@ -34,7 +36,7 @@ public sealed class MqttClientConnection : IDisposable
         readonly string _endpoint;
         readonly DateTime _connectedTimestamp;
 
-        Task<MqttClientDisconnectType> _packageReceiverTask;
+        volatile Task _packageReceiverTask;
         DateTime _lastPacketReceivedTimestamp;
         DateTime _lastNonKeepAlivePacketReceivedTimestamp;
 
@@ -43,7 +45,7 @@ public sealed class MqttClientConnection : IDisposable
         long _receivedApplicationMessagesCount;
         long _sentApplicationMessagesCount;
 
-        bool _isTakeover;
+        volatile bool _isTakeover;
 
         public MqttClientConnection(
             MqttConnectPacket connectPacket,
@@ -52,12 +54,16 @@ public MqttClientConnection(
             IMqttServerOptions serverOptions,
             MqttClientSessionsManager sessionsManager,
             IMqttRetainedMessagesManager retainedMessagesManager,
+            Func<Task> onStart,
+            Func<MqttClientDisconnectType, Task> onStop,
             IMqttNetLogger logger)
         {
             Session = session ?? throw new ArgumentNullException(nameof(session));
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
+            _onStart = onStart ?? throw new ArgumentNullException(nameof(onStart));
+            _onStop = onStop ?? throw new ArgumentNullException(nameof(onStop));
 
             _channelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
             _dataConverter = _channelAdapter.PacketFormatterAdapter.DataConverter;
@@ -80,15 +86,13 @@ public MqttClientConnection(
 
         public MqttClientSession Session { get; }
 
-        public bool IsFinalized { get; set; }
-
         public Task StopAsync(bool isTakeover = false)
         {
             _isTakeover = isTakeover;
+            var task = _packageReceiverTask;
 
             StopInternal();
 
-            var task = _packageReceiverTask;
             if (task != null)
             {
                 return task;
@@ -127,17 +131,18 @@ public void Dispose()
             _cancellationToken.Dispose();
         }
 
-        public Task<MqttClientDisconnectType> RunAsync(MqttConnectionValidatorContext connectionValidatorContext)
+        public Task RunAsync(MqttConnectionValidatorContext connectionValidatorContext)
         {
             _packageReceiverTask = RunInternalAsync(connectionValidatorContext);
             return _packageReceiverTask;
         }
 
-        async Task<MqttClientDisconnectType> RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext)
+        async Task RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext)
         {
             var disconnectType = MqttClientDisconnectType.NotClean;
             try
             {
+                await _onStart();
                 _logger.Info("Client '{0}': Session started.", ClientId);
 
                 _channelAdapter.ReadingPacketStartedCallback = OnAdapterReadingPacketStarted;
@@ -241,6 +246,11 @@ await SendAsync(
             }
             finally
             {
+                if (_isTakeover)
+                {
+                    disconnectType = MqttClientDisconnectType.Takeover;
+                }
+
                 if (Session.WillMessage != null)
                 {
                     _sessionsManager.DispatchApplicationMessage(Session.WillMessage, this);
@@ -255,14 +265,16 @@ await SendAsync(
                 _logger.Info("Client '{0}': Connection stopped.", ClientId);
 
                 _packageReceiverTask = null;
-            }
 
-            if (_isTakeover)
-            {
-                return MqttClientDisconnectType.Takeover;
+                try
+                {
+                    await _onStop(disconnectType);
+                }
+                catch (Exception e)
+                {
+                    _logger.Error(e, "client '{0}': Error while cleaning up", ClientId);
+                }
             }
-
-            return disconnectType;
         }
 
         void StopInternal()