From 603639126e2c95a763eb63b0b1f655ecee66a8d8 Mon Sep 17 00:00:00 2001
From: Christian Kratky <christian.kratky@googlemail.com>
Date: Tue, 24 Jul 2018 13:49:23 +0200
Subject: [PATCH] Fix issue in processing of server messages.

---
 .../Server/MqttClientSessionsManager.cs       | 83 +++++++++++--------
 1 file changed, 49 insertions(+), 34 deletions(-)

diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs
index 34080a00..48732a6d 100644
--- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs
+++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs
@@ -44,7 +44,7 @@ namespace MQTTnet.Server
 
         public void Start()
         {
-            Task.Factory.StartNew(() => ProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
+            Task.Factory.StartNew(() => TryProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
         }
 
         public void Stop()
@@ -133,54 +133,69 @@ namespace MQTTnet.Server
             _messageQueue?.Dispose();
         }
 
-        private void ProcessQueuedApplicationMessages(CancellationToken cancellationToken)
+        private void TryProcessQueuedApplicationMessages(CancellationToken cancellationToken)
         {
             while (!cancellationToken.IsCancellationRequested)
             {
                 try
                 {
-                    var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken);
-                    var sender = enqueuedApplicationMessage.Sender;
-                    var applicationMessage = enqueuedApplicationMessage.PublishPacket.ToApplicationMessage();
-
-                    var interceptorContext = InterceptApplicationMessage(sender, applicationMessage);
-                    if (interceptorContext != null)
-                    {
-                        if (interceptorContext.CloseConnection)
-                        {
-                            enqueuedApplicationMessage.Sender.Stop(MqttClientDisconnectType.NotClean);
-                        }
-
-                        if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
-                        {
-                            return;
-                        }
-
-                        applicationMessage = interceptorContext.ApplicationMessage;
-                    }
-
-                    Server.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
-
-                    if (applicationMessage.Retain)
-                    {
-                        _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).GetAwaiter().GetResult();
-                    }
-
-                    foreach (var clientSession in GetSessions())
-                    {
-                        clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket());
-                    }
+                    TryProcessNextQueuedApplicationMessage(cancellationToken);
                 }
                 catch (OperationCanceledException)
                 {
                 }
                 catch (Exception exception)
                 {
-                    _logger.Error(exception, "Unhandled exception while processing queued application message.");
+                    _logger.Error(exception, "Unhandled exception while processing queued application messages.");
                 }
             }
         }
 
+        private void TryProcessNextQueuedApplicationMessage(CancellationToken cancellationToken)
+        {
+            try
+            {
+                var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken);
+                var sender = enqueuedApplicationMessage.Sender;
+                var applicationMessage = enqueuedApplicationMessage.PublishPacket.ToApplicationMessage();
+
+                var interceptorContext = InterceptApplicationMessage(sender, applicationMessage);
+                if (interceptorContext != null)
+                {
+                    if (interceptorContext.CloseConnection)
+                    {
+                        enqueuedApplicationMessage.Sender.Stop(MqttClientDisconnectType.NotClean);
+                    }
+
+                    if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
+                    {
+                        return;
+                    }
+
+                    applicationMessage = interceptorContext.ApplicationMessage;
+                }
+
+                Server.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
+
+                if (applicationMessage.Retain)
+                {
+                    _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).GetAwaiter().GetResult();
+                }
+
+                foreach (var clientSession in GetSessions())
+                {
+                    clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket());
+                }
+            }
+            catch (OperationCanceledException)
+            {
+            }
+            catch (Exception exception)
+            {
+                _logger.Error(exception, "Unhandled exception while processing next queued application message.");
+            }
+        }
+
         private List<MqttClientSession> GetSessions()
         {
             lock (_sessions)