Replace Delay with Connection Event Signal (#272)

Author: pglombardo

Source/HiveMQtt/Client/Connection/ConnectionManager.cs

@@ -16,6 +16,8 @@
 namespace HiveMQtt.Client.Connection;
 
 using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
 using HiveMQtt.Client.Internal;
 using HiveMQtt.Client.Transport;
 using HiveMQtt.MQTT5;
@@ -68,6 +70,9 @@ public partial class ConnectionManager : IDisposable
     // This is used to know if and when we need to send a MQTT PingReq
     private readonly Stopwatch lastCommunicationTimer = new();
 
+    // Event-like signal to indicate the connection reached Connected state
+    private TaskCompletionSource<bool> connectedSignal = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
     /// <summary>
     /// Initializes a new instance of the <see cref="ConnectionManager"/> class.
     /// </summary>
@@ -79,6 +84,7 @@ public ConnectionManager(HiveMQClient client)
         this.IPubTransactionQueue = new BoundedDictionaryX<int, List<ControlPacket>>(this.Client.Options.ClientReceiveMaximum);
         this.OPubTransactionQueue = new BoundedDictionaryX<int, List<ControlPacket>>(65535);
         this.State = ConnectState.Disconnected;
+        this.ResetConnectedSignal();
 
         // Connect the appropriate transport
         if (this.Client.Options.Host.StartsWith("ws://", StringComparison.OrdinalIgnoreCase) ||
@@ -100,6 +106,18 @@ public ConnectionManager(HiveMQClient client)
         Logger.Trace("    -(RPH)- == ReceivedPacketsHandler");
     }
 
+    internal Task WaitUntilConnectedAsync(CancellationToken cancellationToken) => this.connectedSignal.Task.WaitAsync(cancellationToken);
+
+    internal void SignalConnected()
+    {
+        if (!this.connectedSignal.Task.IsCompleted)
+        {
+            this.connectedSignal.TrySetResult(true);
+        }
+    }
+
+    internal void ResetConnectedSignal() => this.connectedSignal = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+
     internal async Task<bool> ConnectAsync()
     {
         // Connect the appropriate transport

Source/HiveMQtt/Client/Connection/ConnectionManagerHandlers.cs

@@ -42,6 +42,7 @@ internal void HandleIncomingConnAckPacket(ConnAckPacket connAckPacket)
         }
 
         this.ConnectionProperties = connAckPacket.Properties;
+
         this.Client.OnConnAckReceivedEventLauncher(connAckPacket);
     }
 
@@ -390,6 +391,9 @@ internal async Task<bool> HandleDisconnectionAsync(bool clean = true)
         // Cancel all background tasks and close the socket
         this.State = ConnectState.Disconnected;
 
+        // Reset the connection-ready signal for the next connect cycle
+        this.ResetConnectedSignal();
+
         // Cancel all background tasks
         await this.CancelBackgroundTasksAsync().ConfigureAwait(false);
 

Source/HiveMQtt/Client/Connection/ConnectionManagerTasks.cs

@@ -119,11 +119,11 @@ private Task ConnectionPublishWriterAsync(CancellationToken cancellationToken) =
             {
                 try
                 {
-                    while (this.State != ConnectState.Connected)
+                    // Await connection readiness without polling to avoid arbitrary delay
+                    if (this.State != ConnectState.Connected)
                     {
                         Logger.Trace($"{this.Client.Options.ClientId}-(PW)- Not connected.  Waiting for connect...");
-                        await Task.Delay(500).ConfigureAwait(false);
-                        continue;
+                        await this.WaitUntilConnectedAsync(cancellationToken).ConfigureAwait(false);
                     }
 
                     var writeSuccess = true;

Source/HiveMQtt/Client/HiveMQClient.cs

@@ -152,6 +152,9 @@ public async Task<ConnectResult> ConnectAsync(ConnectOptions? connectOptions = n
         if (connAck.ReasonCode == ConnAckReasonCode.Success)
         {
             this.Connection.State = ConnectState.Connected;
+
+            // Ensure connection-ready signal is set for any writers awaiting readiness
+            this.Connection.SignalConnected();
         }
         else
         {