Refactor Confirmation Pipe and Heatbeat start (#141)

* Move the confirmation pipe after the Producer initialization to avoid
pendig thread in case the producer init fails

* Move the heartbeat start after the login to avoid pending threads
in case the client fails to connect for some reason.

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

RabbitMQ.Stream.Client/Client.cs

@@ -85,7 +85,7 @@ public int Write(Span<byte> span)
 
     public class Client : IClient
     {
-        private bool isClosed = false;
+        private bool isClosed = true;
 
         private readonly TimeSpan defaultTimeout = TimeSpan.FromSeconds(10);
 
@@ -165,6 +165,11 @@ private Client(ClientParameters parameters)
             IsClosed = false;
         }
 
+        private void StartHeartBeat()
+        {
+            _heartBeatHandler.Start();
+        }
+
         public delegate Task ConnectionCloseHandler(string reason);
 
         public event ConnectionCloseHandler ConnectionClosed;
@@ -228,6 +233,8 @@ await client.Publish(new TuneRequest(0,
             client.ConnectionProperties = open.ConnectionProperties;
 
             client.correlationId = 100;
+            // start heart beat only when the client is connected
+            client.StartHeartBeat();
             return client;
         }
 

RabbitMQ.Stream.Client/HeartBeatHandler.cs

@@ -30,7 +30,7 @@ public HeartBeatHandler(Func<ValueTask<bool>> sendHeartbeatFunc,
         // }
         if (heartbeat > 0)
         {
-            _timer.Enabled = true;
+            _timer.Enabled = false;
             _timer.Interval = heartbeat * 1000;
             _timer.Elapsed += (_, _) =>
             {
@@ -72,6 +72,11 @@ internal void Close()
         _timer.Close();
     }
 
+    internal void Start()
+    {
+        _timer.Enabled = true;
+    }
+
     internal bool IsActive()
     {
         return _timer.Enabled;

RabbitMQ.Stream.Client/Reliable/ReliableProducer.cs

@@ -61,7 +61,6 @@ private ReliableProducer(ReliableProducerConfig reliableProducerConfig)
             reliableProducerConfig.ConfirmationHandler,
             reliableProducerConfig.TimeoutMessageAfter
         );
-        _confirmationPipe.Start();
     }
 
     public static async Task<ReliableProducer> CreateReliableProducer(ReliableProducerConfig reliableProducerConfig)
@@ -114,6 +113,9 @@ protected override async Task GetNewReliable(bool boot)
                 // Init the publishing id
                 Interlocked.Exchange(ref _publishingId,
                     await _producer.GetLastPublishingId());
+
+                // confirmation Pipe can start only if the producer is ready
+                _confirmationPipe.Start();
             }
         }
         catch (Exception e)

Tests/UnitTests.cs

@@ -322,7 +322,7 @@ public void HeartBeatRaiseClose()
                     return null;
                 },
                 1);
-
+            hBeatHandler.Start();
             var r = testPassed.Task.Wait(6_000);
             Assert.True(r);
             Assert.False(hBeatHandler.IsActive());