Make reconnection async

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

RabbitMQ.Stream.Client/Reliable/ReliableBase.cs

@@ -82,6 +82,7 @@ internal async Task HandleMetaDataMaybeReconnect(string stream, StreamSystem sys
                 $" Client: {ToString()}");
             // Here we just close the producer connection
             // the func TryToReconnect/0 will be called. 
+
             await CloseReliable();
         }
         else
@@ -100,6 +101,13 @@ internal static bool IsAKnownException(Exception exception)
         return exception is (SocketException or TimeoutException or LeaderNotFoundException);
     }
 
+    internal void LogException(Exception exception)
+    {
+        LogEventSource.Log.LogError(IsAKnownException(exception)
+            ? $"Trying to reconnect {ToString()} due of: {exception.Message}"
+            : $"Error during initialization {ToString()} due of: {exception.Message}");
+    }
+
     protected abstract Task CloseReliable();
     public abstract Task Close();
 

RabbitMQ.Stream.Client/Reliable/ReliableConsumer.cs

@@ -76,8 +76,11 @@ protected override async Task GetNewReliable(bool boot)
                 },
                 MetadataHandler = update =>
                 {
-                    HandleMetaDataMaybeReconnect(update.Stream,
-                        _reliableConsumerConfig.StreamSystem).Wait();
+                    Task.Run(() =>
+                    {
+                        HandleMetaDataMaybeReconnect(update.Stream,
+                            _reliableConsumerConfig.StreamSystem).WaitAsync(CancellationToken.None);
+                    });
                 },
                 MessageHandler = async (consumer, ctx, message) =>
                 {
@@ -95,6 +98,7 @@ protected override async Task GetNewReliable(bool boot)
         catch (Exception e)
         {
             reconnect = IsAKnownException(e);
+            LogException(e);
             if (!reconnect)
             {
                 throw;

RabbitMQ.Stream.Client/Reliable/ReliableProducer.cs

@@ -90,8 +90,11 @@ protected override async Task GetNewReliable(bool boot)
                 MaxInFlight = _reliableProducerConfig.MaxInFlight,
                 MetadataHandler = update =>
                 {
-                    HandleMetaDataMaybeReconnect(update.Stream,
-                        _reliableProducerConfig.StreamSystem).Wait();
+                    Task.Run(() =>
+                    {
+                        HandleMetaDataMaybeReconnect(update.Stream,
+                            _reliableProducerConfig.StreamSystem).WaitAsync(CancellationToken.None);
+                    });
                 },
                 ConnectionClosedHandler = async _ =>
                 {
@@ -130,6 +133,7 @@ protected override async Task GetNewReliable(bool boot)
         catch (Exception e)
         {
             reconnect = IsAKnownException(e);
+            LogException(e);
             if (!reconnect)
             {
                 throw;