Merge

Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs

@@ -6,17 +6,16 @@ namespace MQTTnet.Client
     public interface IMqttClientOptions
     {
         string ClientId { get; }
-
-        IMqttClientCredentials Credentials { get; }
         bool CleanSession { get; }
-        MqttApplicationMessage WillMessage { get; }
+        IMqttClientCredentials Credentials { get; }
+        MqttProtocolVersion ProtocolVersion { get; }
+        IMqttClientChannelOptions ChannelOptions { get; }
         
         TimeSpan CommunicationTimeout { get; }
         TimeSpan KeepAlivePeriod { get; }
         TimeSpan? KeepAliveSendInterval { get; }
+        MqttReceivedApplicationMessageProcessingMode ReceivedApplicationMessageProcessingMode { get; }
 
-        MqttProtocolVersion ProtocolVersion { get; }
-
-        IMqttClientChannelOptions ChannelOptions { get; }
+        MqttApplicationMessage WillMessage { get; }
     }
 }

Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs

@@ -63,7 +63,7 @@ namespace MQTTnet.Client
 
                 StartReceivingPackets(_cancellationTokenSource.Token);
 
-                var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false);
+                var connectResponse = await AuthenticateAsync(options.WillMessage, _cancellationTokenSource.Token).ConfigureAwait(false);
                 _logger.Verbose<MqttClient>("MQTT connection with server established.");
 
                 _sendTracker.Restart();
@@ -94,7 +94,7 @@ namespace MQTTnet.Client
             {
                 if (IsConnected && !_cancellationTokenSource.IsCancellationRequested)
                 {
-                    await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false);
+                    await SendAsync(new MqttDisconnectPacket(), _cancellationTokenSource.Token).ConfigureAwait(false);
                 }
             }
             finally
@@ -115,7 +115,7 @@ namespace MQTTnet.Client
                 TopicFilters = topicFilters.ToList()
             };
 
-            var response = await SendAndReceiveAsync<MqttSubAckPacket>(subscribePacket).ConfigureAwait(false);
+            var response = await SendAndReceiveAsync<MqttSubAckPacket>(subscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false);
 
             if (response.SubscribeReturnCodes.Count != subscribePacket.TopicFilters.Count)
             {
@@ -137,7 +137,7 @@ namespace MQTTnet.Client
                 TopicFilters = topicFilters.ToList()
             };
 
-            await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket).ConfigureAwait(false);
+            await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false);
         }
 
         public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
@@ -154,7 +154,7 @@ namespace MQTTnet.Client
                     case MqttQualityOfServiceLevel.AtMostOnce:
                         {
                             // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
-                            await SendAsync(qosGroup.Cast<MqttBasePacket>().ToArray()).ConfigureAwait(false);
+                            await SendAsync(qosGroup, _cancellationTokenSource.Token).ConfigureAwait(false);
                             break;
                         }
                     case MqttQualityOfServiceLevel.AtLeastOnce:
@@ -162,7 +162,7 @@ namespace MQTTnet.Client
                             foreach (var publishPacket in qosGroup)
                             {
                                 publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
-                                await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket).ConfigureAwait(false);
+                                await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, _cancellationTokenSource.Token).ConfigureAwait(false);
                             }
 
                             break;
@@ -173,13 +173,13 @@ namespace MQTTnet.Client
                             {
                                 publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
 
-                                var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
+                                var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, _cancellationTokenSource.Token).ConfigureAwait(false);
                                 var pubRelPacket = new MqttPubRelPacket
                                 {
                                     PacketIdentifier = pubRecPacket.PacketIdentifier
                                 };
 
-                                await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket).ConfigureAwait(false);
+                                await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, _cancellationTokenSource.Token).ConfigureAwait(false);
                             }
 
                             break;
@@ -200,7 +200,7 @@ namespace MQTTnet.Client
             _adapter?.Dispose();
         }
 
-        private async Task<MqttConnAckPacket> AuthenticateAsync(MqttApplicationMessage willApplicationMessage)
+        private async Task<MqttConnAckPacket> AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken)
         {
             var connectPacket = new MqttConnectPacket
             {
@@ -212,7 +212,7 @@ namespace MQTTnet.Client
                 WillMessage = willApplicationMessage
             };
 
-            var response = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket).ConfigureAwait(false);
+            var response = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket, _cancellationTokenSource.Token).ConfigureAwait(false);
             if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
             {
                 throw new MqttConnectingFailedException(response.ConnectReturnCode);
@@ -257,14 +257,12 @@ namespace MQTTnet.Client
 
             try
             {
-                if (_packetReceiverTask != null && _packetReceiverTask != sender)
-                {
-                    _packetReceiverTask.Wait();
-                }
+                await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false);
+                await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false);
 
                 if (_keepAliveMessageSenderTask != null && _keepAliveMessageSenderTask != sender)
                 {
-                    _keepAliveMessageSenderTask.Wait();
+                    await _keepAliveMessageSenderTask.ConfigureAwait(false);
                 }
 
                 if (_adapter != null)
@@ -290,101 +288,19 @@ namespace MQTTnet.Client
             }
         }
 
-        private async Task ProcessReceivedPacketAsync(MqttBasePacket packet)
-        {
-            try
-            {
-                if (packet is MqttPublishPacket publishPacket)
-                {
-                    await ProcessReceivedPublishPacketAsync(publishPacket).ConfigureAwait(false);
-                    return;
-                }
-
-                if (packet is MqttPingReqPacket)
-                {
-                    await SendAsync(new MqttPingRespPacket()).ConfigureAwait(false);
-                    return;
-                }
-
-                if (packet is MqttDisconnectPacket)
-                {
-                    await DisconnectAsync().ConfigureAwait(false);
-                    return;
-                }
-
-                if (packet is MqttPubRelPacket pubRelPacket)
-                {
-                    await ProcessReceivedPubRelPacket(pubRelPacket).ConfigureAwait(false);
-                    return;
-                }
-
-                _packetDispatcher.Dispatch(packet);
-            }
-            catch (Exception exception)
-            {
-                _logger.Error<MqttClient>(exception, "Unhandled exception while processing received packet.");
-            }
-        }
-
-        private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)
-        {
-            try
-            {
-                var applicationMessage = publishPacket.ToApplicationMessage();
-                ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage));
-            }
-            catch (Exception exception)
-            {
-                _logger.Error<MqttClient>(exception, "Unhandled exception while handling application message.");
-            }
-        }
-
-        private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket)
-        {
-            if (_cancellationTokenSource.IsCancellationRequested)
-            {
-                return Task.FromResult(0);
-            }
-
-            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
-            {
-                FireApplicationMessageReceivedEvent(publishPacket);
-                return Task.FromResult(0);
-            }
-
-            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
-            {
-                FireApplicationMessageReceivedEvent(publishPacket);
-                return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
-            }
-
-            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
-            {
-                // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
-                FireApplicationMessageReceivedEvent(publishPacket);
-                return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
-            }
-
-            throw new MqttCommunicationException("Received a not supported QoS level.");
-        }
-
-        private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket)
-        {
-            var response = new MqttPubCompPacket
-            {
-                PacketIdentifier = pubRelPacket.PacketIdentifier
-            };
-
-            return SendAsync(response);
-        }
-
-        private Task SendAsync(params MqttBasePacket[] packets)
+        private Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken)
         {
             _sendTracker.Restart();
-            return _adapter.SendPacketsAsync(_options.CommunicationTimeout, packets, _cancellationTokenSource.Token);
+            return _adapter.SendPacketsAsync(_options.CommunicationTimeout, cancellationToken, new[] { packet });
         }
 
-        private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket
+        private Task SendAsync(IEnumerable<MqttBasePacket> packets, CancellationToken cancellationToken)
+        {
+            _sendTracker.Restart();
+            return _adapter.SendPacketsAsync(_options.CommunicationTimeout, cancellationToken, packets);
+        }
+
+        private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket
         {
             _sendTracker.Restart();
 
@@ -429,7 +345,7 @@ namespace MQTTnet.Client
 
                     if (_sendTracker.Elapsed > keepAliveSendInterval)
                     {
-                        await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
+                        await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false);
                     }
 
                     await Task.Delay(keepAliveSendInterval, cancellationToken).ConfigureAwait(false);
@@ -472,7 +388,19 @@ namespace MQTTnet.Client
                         return;
                     }
 
-                    StartProcessReceivedPacket(packet, cancellationToken);
+                    if (packet == null)
+                    {
+                        continue;
+                    }
+
+                    if (_options.ReceivedApplicationMessageProcessingMode == MqttReceivedApplicationMessageProcessingMode.SingleThread)
+                    {
+                        await ProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false);
+                    }
+                    else if (_options.ReceivedApplicationMessageProcessingMode == MqttReceivedApplicationMessageProcessingMode.DedicatedThread)
+                    {
+                        StartProcessReceivedPacketAsync(packet, cancellationToken);
+                    }
                 }
             }
             catch (Exception exception)
@@ -498,9 +426,74 @@ namespace MQTTnet.Client
             }
         }
 
-        private void StartProcessReceivedPacket(MqttBasePacket packet, CancellationToken cancellationToken)
+        private async Task ProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
         {
-            Task.Run(() => ProcessReceivedPacketAsync(packet), cancellationToken);
+            try
+            {
+                if (packet is MqttPublishPacket publishPacket)
+                {
+                    await ProcessReceivedPublishPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false);
+                    return;
+                }
+
+                if (packet is MqttPingReqPacket)
+                {
+                    await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false);
+                    return;
+                }
+
+                if (packet is MqttDisconnectPacket)
+                {
+                    await DisconnectAsync().ConfigureAwait(false);
+                    return;
+                }
+
+                if (packet is MqttPubRelPacket pubRelPacket)
+                {
+                    await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false);
+                    return;
+                }
+
+                _packetDispatcher.Dispatch(packet);
+            }
+            catch (Exception exception)
+            {
+                _logger.Error<MqttClient>(exception, "Unhandled exception while processing received packet.");
+            }
+        }
+
+        private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        {
+            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
+            {
+                FireApplicationMessageReceivedEvent(publishPacket);
+                return Task.FromResult(0);
+            }
+
+            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
+            {
+                FireApplicationMessageReceivedEvent(publishPacket);
+                return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, cancellationToken);
+            }
+
+            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
+            {
+                // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
+                FireApplicationMessageReceivedEvent(publishPacket);
+                return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, cancellationToken);
+            }
+
+            throw new MqttCommunicationException("Received a not supported QoS level.");
+        }
+
+        private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
+        {
+            var response = new MqttPubCompPacket
+            {
+                PacketIdentifier = pubRelPacket.PacketIdentifier
+            };
+
+            return SendAsync(response, cancellationToken);
         }
 
         private void StartReceivingPackets(CancellationToken cancellationToken)
@@ -512,5 +505,46 @@ namespace MQTTnet.Client
         {
             _keepAliveMessageSenderTask = Task.Run(() => SendKeepAliveMessagesAsync(cancellationToken), cancellationToken);
         }
+
+        private void StartProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
+        {
+#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            Task.Run(() => ProcessReceivedPacketAsync(packet, cancellationToken), cancellationToken);
+#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+        }
+
+        private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)
+        {
+            try
+            {
+                var applicationMessage = publishPacket.ToApplicationMessage();
+                ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage));
+            }
+            catch (Exception exception)
+            {
+                _logger.Error<MqttClient>(exception, "Unhandled exception while handling application message.");
+            }
+        }
+
+        private static async Task WaitForTaskAsync(Task task, Task sender)
+        {
+            if (task == sender)
+            {
+                return;
+            }
+
+            if (task.IsCanceled || task.IsCompleted || task.IsFaulted)
+            {
+                return;
+            }
+
+            try
+            {
+                await task.ConfigureAwait(false);
+            }
+            catch (TaskCanceledException)
+            {
+            }
+        }
     }
 }

Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs

@@ -5,22 +5,17 @@ namespace MQTTnet.Client
 {
     public class MqttClientOptions : IMqttClientOptions
     {
-        public MqttApplicationMessage WillMessage { get; set; }
-
         public string ClientId { get; set; } = Guid.NewGuid().ToString("N");
-
         public bool CleanSession { get; set; } = true;
-
         public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials();
-
-        public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15);
-
-        public TimeSpan? KeepAliveSendInterval { get; set; }
+        public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
+        public IMqttClientChannelOptions ChannelOptions { get; set; }
 
         public TimeSpan CommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
+        public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15);
+        public TimeSpan? KeepAliveSendInterval { get; set; }
+        public MqttReceivedApplicationMessageProcessingMode ReceivedApplicationMessageProcessingMode { get; set; } = MqttReceivedApplicationMessageProcessingMode.SingleThread;
 
-        public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
-
-        public IMqttClientChannelOptions ChannelOptions { get; set; }
+        public MqttApplicationMessage WillMessage { get; set; }
     }
 }

Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs

@@ -114,6 +114,13 @@ namespace MQTTnet.Client
             return this;
         }
 
+        public MqttClientOptionsBuilder WithReceivedApplicationMessageProcessingMode(
+            MqttReceivedApplicationMessageProcessingMode mode)
+        {
+            _options.ReceivedApplicationMessageProcessingMode = mode;
+            return this;
+        }
+
         public IMqttClientOptions Build()
         {
             if (_tlsOptions != null)

Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs

@@ -0,0 +1,8 @@
+namespace MQTTnet.Client
+{
+    public enum MqttReceivedApplicationMessageProcessingMode
+    {
+        SingleThread,
+        DedicatedThread
+    }
+}

Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs

@@ -0,0 +1,8 @@
+namespace MQTTnet.Server
+{
+    public enum MqttClientDisconnectType
+    {
+        Clean,
+        NotClean
+    }
+}

Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs

@@ -127,7 +127,7 @@ namespace MQTTnet.Server
 
                 if (!cancellationToken.IsCancellationRequested)
                 {
-                    await _clientSession.StopAsync().ConfigureAwait(false);
+                    await _clientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
                 }
             }
         }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs

@@ -101,7 +101,7 @@ namespace MQTTnet.Server
             return _wasCleanDisconnect;
         }
 
-        public Task StopAsync(bool wasCleanDisconnect = false)
+        public async Task StopAsync(MqttClientDisconnectType type)
         {
             try
             {
@@ -110,7 +110,7 @@ namespace MQTTnet.Server
                     return Task.FromResult(0);
                 }
 
-                _wasCleanDisconnect = wasCleanDisconnect;
+                _wasCleanDisconnect = type == MqttClientDisconnectType.Clean;
 
                 _cancellationTokenSource?.Cancel(false);
                 PendingMessagesQueue.WaitForCompletion();
@@ -119,7 +119,7 @@ namespace MQTTnet.Server
                 var willMessage = _willMessage;
                 _willMessage = null; // clear willmessage so it is send just once
 
-                if (_willMessage != null && !wasCleanDisconnect)
+                if (willMessage != null && !_wasCleanDisconnect)
                 {
                     _sessionsManager.StartDispatchApplicationMessage(this, willMessage);
                 }
@@ -186,7 +186,7 @@ namespace MQTTnet.Server
         private Task StopDueToKeepAliveTimeoutAsync()
         {
             _logger.Info<MqttClientSession>("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId);
-            return StopAsync();
+            return StopAsync(MqttClientDisconnectType.NotClean);
         }
 
         private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -203,15 +203,18 @@ namespace MQTTnet.Server
             catch (OperationCanceledException)
             {
             }
-            catch (MqttCommunicationException exception)
-            {
-                _logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
-                await StopAsync().ConfigureAwait(false);
-            }
             catch (Exception exception)
             {
-                _logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
-                await StopAsync().ConfigureAwait(false);
+                if (exception is MqttCommunicationException)
+                {
+                    _logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
+                }
+                else
+                {
+                    _logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
+                }
+
+                await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
             }
         }
 
@@ -260,16 +263,16 @@ namespace MQTTnet.Server
 
             if (packet is MqttDisconnectPacket)
             {
-                return StopAsync(true);
+                return StopAsync(MqttClientDisconnectType.Clean);
             }
 
             if (packet is MqttConnectPacket)
             {
-                return StopAsync();
+                return StopAsync(MqttClientDisconnectType.NotClean);
             }
 
             _logger.Warning<MqttClientSession>("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
-            return StopAsync();
+            return StopAsync(MqttClientDisconnectType.NotClean);
         }
 
         private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters)
@@ -288,7 +291,7 @@ namespace MQTTnet.Server
 
             if (subscribeResult.CloseConnection)
             {
-                await StopAsync().ConfigureAwait(false);
+                await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
             }
 
             await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);

Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs

@@ -120,7 +120,7 @@ namespace MQTTnet.Server
             {
                 foreach (var session in _sessions)
                 {
-                    await session.Value.StopAsync().ConfigureAwait(false);
+                    await session.Value.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
                 }
 
                 _sessions.Clear();
@@ -232,7 +232,7 @@ namespace MQTTnet.Server
                     {
                         _sessions.Remove(connectPacket.ClientId);
 
-                        await clientSession.StopAsync().ConfigureAwait(false);
+                        await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false);
                         clientSession.Dispose();
                         clientSession = null;