From 043cfdb3f9e23c6cb29b31064bd44afef7d46e05 Mon Sep 17 00:00:00 2001
From: JanEggers <jan.eggers@ferag.com>
Date: Mon, 23 Jul 2018 17:58:29 +0200
Subject: [PATCH] fixed deadlock and object disposed exception

---
 Source/MQTTnet/Server/MqttClientSession.cs  | 43 +++++++++++++--------
 Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 40 +++++++++++++++++++
 2 files changed, 67 insertions(+), 16 deletions(-)

diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs
index 8ef401c2..8ddb8455 100644
--- a/Source/MQTTnet/Server/MqttClientSession.cs
+++ b/Source/MQTTnet/Server/MqttClientSession.cs
@@ -126,18 +126,11 @@ namespace MQTTnet.Server
                     _logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId);
                 }
 
-                Stop(MqttClientDisconnectType.NotClean);
+                Stop(MqttClientDisconnectType.NotClean, true);
             }
             finally
             {
-                if (_adapter != null)
-                {
-                    _adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
-                    _adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
-                    await Cleanup().ConfigureAwait(false);
-                }
-
-                _adapter = null;
+                await Cleanup().ConfigureAwait(false);
 
                 _cleanupHandle?.Dispose();
                 _cleanupHandle = null;
@@ -153,8 +146,18 @@ namespace MQTTnet.Server
         {
             try
             {
-                await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
-                _adapter.Dispose();
+                var adapter = _adapter;
+                if (adapter == null)
+                {
+                    return;
+                }
+
+                _adapter = null;
+
+                adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
+                adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
+                await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
+                adapter.Dispose();
             }
             catch (Exception exception)
             {
@@ -163,6 +166,11 @@ namespace MQTTnet.Server
         }
 
         public void Stop(MqttClientDisconnectType type)
+        {
+            Stop(type, false);
+        }
+
+        private void Stop(MqttClientDisconnectType type, bool isInsideSession)
         {
             try
             {
@@ -183,7 +191,10 @@ namespace MQTTnet.Server
 
                 _willMessage = null;
 
-                _workerTask?.GetAwaiter().GetResult();
+                if (!isInsideSession)
+                {
+                    _workerTask?.GetAwaiter().GetResult();
+                }
             }
             finally
             {
@@ -329,18 +340,18 @@ namespace MQTTnet.Server
 
             if (packet is MqttDisconnectPacket)
             {
-                Stop(MqttClientDisconnectType.Clean);
+                Stop(MqttClientDisconnectType.Clean, true);
                 return;
             }
 
             if (packet is MqttConnectPacket)
             {
-                Stop(MqttClientDisconnectType.NotClean);
+                Stop(MqttClientDisconnectType.NotClean, true);
                 return;
             }
 
             _logger.Warning(null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
-            Stop(MqttClientDisconnectType.NotClean);
+            Stop(MqttClientDisconnectType.NotClean, true);
         }
 
         private void EnqueueSubscribedRetainedMessages(ICollection<TopicFilter> topicFilters)
@@ -359,7 +370,7 @@ namespace MQTTnet.Server
 
             if (subscribeResult.CloseConnection)
             {
-                Stop(MqttClientDisconnectType.NotClean);
+                Stop(MqttClientDisconnectType.NotClean, true);
                 return;
             }
 
diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs
index 117e6ef9..03888668 100644
--- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs
+++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs
@@ -262,6 +262,46 @@ namespace MQTTnet.Core.Tests
             Assert.AreEqual(1, disconnectCalled);
         }
 
+        [TestMethod]
+        public async Task MqttServer_HandleCleanDisconnect()
+        {
+            MqttNetGlobalLogger.LogMessagePublished += (_, e) =>
+            {
+                System.Diagnostics.Debug.WriteLine($"[{e.TraceMessage.Timestamp:s}] {e.TraceMessage.Source} {e.TraceMessage.Message}");
+            };
+
+            var serverAdapter = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger());
+            var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
+
+            var clientConnectedCalled = 0;
+            var clientDisconnectedCalled = 0;
+
+            s.ClientConnected += (_, __) => clientConnectedCalled++;
+            s.ClientDisconnected += (_, __) => clientDisconnectedCalled++;
+
+            var clientOptions = new MqttClientOptionsBuilder()
+                .WithTcpServer("localhost")
+                .Build();
+
+            await s.StartAsync(new MqttServerOptions());
+
+            var c1 = new MqttFactory().CreateMqttClient();
+
+            await c1.ConnectAsync(clientOptions);
+
+            await Task.Delay(100);
+
+            await c1.DisconnectAsync();
+
+            await Task.Delay(100);
+
+            await s.StopAsync();
+
+            await Task.Delay(100);
+
+            Assert.AreEqual(clientConnectedCalled, clientDisconnectedCalled);
+        }
+
         [TestMethod]
         public async Task MqttServer_RetainedMessagesFlow()
         {