fix: ConnectionMonitor is stopped by cancellation token during disconnect

Author: herrmann-m

Source/HiveMQtt/Client/Connection/ConnectionManager.cs

@@ -129,7 +129,7 @@ internal async Task<bool> ConnectAsync()
         this.ConnectionWriterTask = this.ConnectionWriterAsync(this.cancellationTokenSource.Token);
         this.ConnectionReaderTask = this.ConnectionReaderAsync(this.cancellationTokenSource.Token);
         this.ReceivedPacketsHandlerTask = this.ReceivedPacketsHandlerAsync(this.cancellationTokenSource.Token);
-        this.ConnectionMonitorThread = this.LaunchConnectionMonitorThread();
+        this.ConnectionMonitorThread = this.LaunchConnectionMonitorThreadAsync(this.cancellationTokenSource.Token);
 
         return true;
     }
@@ -184,7 +184,7 @@ internal async Task CancelBackgroundTasksAsync()
             Logger.Trace("ReceivedPacketsHandlerTask did not complete in time");
         }
 
-        if (this.ConnectionMonitorThread is not null && !this.ConnectionMonitorThread.IsAlive)
+        if (this.ConnectionMonitorThread is not null && this.ConnectionMonitorThread.IsCompleted)
         {
             this.ConnectionMonitorThread = null;
         }

Source/HiveMQtt/Client/Connection/ConnectionManagerTasks.cs

@@ -21,7 +21,7 @@ public partial class ConnectionManager
 
     internal Task? ReceivedPacketsHandlerTask { get; set; }
 
-    internal Thread? ConnectionMonitorThread { get; set; }
+    internal Task? ConnectionMonitorThread { get; set; }
 
     /// <summary>
     /// Health check method to assure that tasks haven't faulted unexpectedly.
@@ -43,18 +43,14 @@ private void RunTaskHealthCheck(Task? task, string taskName)
         }
     }
 
-    private Thread LaunchConnectionMonitorThread()
-    {
-        var thread = new Thread(this.ConnectionMonitor);
-        thread.Start();
-        return thread;
-    }
+    private Task LaunchConnectionMonitorThreadAsync(CancellationToken cancellationToken) =>
+        Task.Run(() => this.ConnectionMonitorAsync(cancellationToken), cancellationToken);
 
     /// <summary>
     /// Asynchronous background task that monitors the connection state and sends PingReq packets when
     /// necessary.
     /// </summary>
-    private void ConnectionMonitor()
+    private async Task ConnectionMonitorAsync(CancellationToken cancellationToken)
     {
         Logger.Trace($"{this.Client.Options.ClientId}-(CM)- Starting...{this.State}");
         if (this.Client.Options.KeepAlive == 0)
@@ -65,7 +61,7 @@ private void ConnectionMonitor()
         var keepAlivePeriod = this.Client.Options.KeepAlive;
         this.lastCommunicationTimer.Start();
 
-        while (true)
+        while (!cancellationToken.IsCancellationRequested)
         {
             try
             {
@@ -97,7 +93,11 @@ private void ConnectionMonitor()
                 this.RunTaskHealthCheck(this.ReceivedPacketsHandlerTask, "ReceivedPacketsHandler");
 
                 // Sleep cycle
-                Thread.Sleep(2000);
+                await Task.Delay(2000, cancellationToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                Logger.Debug($"{this.Client.Options.ClientId}-(CM)- Stopped by cancellation token");
             }
             catch (Exception ex)
             {

Tests/HiveMQtt.Test/HiveMQClient/ClientTest.cs

@@ -81,7 +81,7 @@ public async Task ClientStateAsync()
         Assert.Equal(TaskStatus.WaitingForActivation, client.Connection.ConnectionWriterTask.Status);
         Assert.Equal(TaskStatus.WaitingForActivation, client.Connection.ConnectionReaderTask.Status);
         Assert.Equal(TaskStatus.WaitingForActivation, client.Connection.ReceivedPacketsHandlerTask.Status);
-        Assert.True(client.Connection.ConnectionMonitorThread.IsAlive);
+        Assert.False(client.Connection.ConnectionMonitorThread.IsCompleted);
 
         // Queues
         Assert.NotNull(client.Connection.SendQueue);
@@ -118,7 +118,7 @@ public async Task ClientStateAsync()
         Assert.Equal(TaskStatus.WaitingForActivation, client.Connection.ConnectionWriterTask.Status);
         Assert.Equal(TaskStatus.WaitingForActivation, client.Connection.ConnectionReaderTask.Status);
         Assert.Equal(TaskStatus.WaitingForActivation, client.Connection.ReceivedPacketsHandlerTask.Status);
-        Assert.True(client.Connection.ConnectionMonitorThread.IsAlive);
+        Assert.False(client.Connection.ConnectionMonitorThread.IsCompleted);
 
         // Queues
         Assert.NotNull(client.Connection.SendQueue);
@@ -136,7 +136,10 @@ public async Task ClientStateAsync()
         Assert.Null(client.Connection.ConnectionWriterTask);
         Assert.Null(client.Connection.ConnectionReaderTask);
         Assert.Null(client.Connection.ReceivedPacketsHandlerTask);
-        Assert.True(client.Connection.ConnectionMonitorThread.IsAlive);
+
+        // The task should be completed and null at this point since with every new call to ConnectAsync
+        // a new task is started and during DisconnectAsync this task should be stopped and removed
+        Assert.Null(client.Connection.ConnectionMonitorThread);
 
         // Queues
         Assert.NotNull(client.Connection.SendQueue);
@@ -145,4 +148,23 @@ public async Task ClientStateAsync()
         // State
         Assert.Equal(ConnectState.Disconnected, client.Connection.State);
     }
+
+    [Fact]
+    public async Task AfterDisconnect_ConnectionMonitorThread_ShouldBe_StoppedAsync()
+    {
+        using var client = new HiveMQClient();
+        await client.ConnectAsync().ConfigureAwait(false);
+
+        // Hold the reference to the task since it's removed in the client
+        // after DisconnectAsync
+        var monitorThread = client.Connection.ConnectionMonitorThread;
+
+        Assert.True(monitorThread is not null && !monitorThread.IsCompleted);
+
+        await client.DisconnectAsync().ConfigureAwait(false);
+
+        // Task should be completed at this point. During DisconnectAsync the cancellation token
+        // should be cancelled and the task should stop
+        Assert.True(monitorThread is not null && monitorThread.IsCompleted);
+    }
 }