Fix occurrences of sync-over-async

README.md

@@ -570,8 +570,8 @@ By default Reliable Producer/Consumer uses an `BackOffReconnectStrategy` to reco
 You can customize the behaviour implementing the `IReconnectStrategy` interface:
 
 ```csharp
-bool WhenDisconnected(string connectionInfo);
-void WhenConnected(string connectionInfo);
+ValueTask<bool> WhenDisconnected(string connectionInfo);
+ValueTask WhenConnected(string connectionInfo);
 ```
 
 If `WhenDisconnected` return is `true` Producer/Consumer will be reconnected else closed.

RabbitMQ.Stream.Client/Consumer.cs

@@ -117,11 +117,11 @@ await config.MessageHandler(this,
             throw new CreateConsumerException($"consumer could not be created code: {response.ResponseCode}");
         }
 
-        public Task<ResponseCode> Close()
+        public async Task<ResponseCode> Close()
         {
             if (client.IsClosed)
             {
-                return Task.FromResult(ResponseCode.Ok);
+                return ResponseCode.Ok;
             }
 
             var result = ResponseCode.Ok;
@@ -132,7 +132,7 @@ public Task<ResponseCode> Close()
                 // in this case we reduce the waiting time
                 // the consumer could be removed because of stream deleted 
                 // so it is not necessary to wait.
-                deleteConsumerResponseTask.Wait(TimeSpan.FromSeconds(3));
+                await deleteConsumerResponseTask.WaitAsync(TimeSpan.FromSeconds(3));
                 if (deleteConsumerResponseTask.IsCompletedSuccessfully)
                 {
                     result = deleteConsumerResponseTask.Result.ResponseCode;
@@ -146,7 +146,7 @@ public Task<ResponseCode> Close()
             var closed = client.MaybeClose($"client-close-subscriber: {subscriberId}");
             ClientExceptions.MaybeThrowException(closed.ResponseCode, $"client-close-subscriber: {subscriberId}");
             _disposed = true;
-            return Task.FromResult(result);
+            return result;
         }
 
         //

RabbitMQ.Stream.Client/HeartBeatHandler.cs

@@ -16,10 +16,18 @@ public class HeartBeatHandler
     private DateTime _lastUpdate = DateTime.Now;
     private uint _missedHeartbeat;
 
+    private readonly Func<ValueTask<bool>> _sendHeartbeatFunc;
+    private readonly Func<string, Task<CloseResponse>> _close;
+    private readonly int _heartbeat;
+
     public HeartBeatHandler(Func<ValueTask<bool>> sendHeartbeatFunc,
         Func<string, Task<CloseResponse>> close,
         int heartbeat)
     {
+        _sendHeartbeatFunc = sendHeartbeatFunc;
+        _close = close;
+        _heartbeat = heartbeat;
+
         // the heartbeat is disabled when zero
         // so all the timer won't be enabled
 
@@ -32,35 +40,42 @@ public HeartBeatHandler(Func<ValueTask<bool>> sendHeartbeatFunc,
         {
             _timer.Enabled = false;
             _timer.Interval = heartbeat * 1000;
-            _timer.Elapsed += (_, _) =>
-            {
-                var f = sendHeartbeatFunc();
-                f.AsTask().Wait(1000);
-
-                var seconds = (DateTime.Now - _lastUpdate).TotalSeconds;
-                if (seconds < heartbeat)
-                {
-                    return;
-                }
-
-                // missed the Heartbeat 
-                Interlocked.Increment(ref _missedHeartbeat);
-                LogEventSource.Log.LogWarning($"Heartbeat missed: {_missedHeartbeat}");
-                if (_missedHeartbeat <= 3)
-                {
-                    return;
-                }
-
-                // When the client does not receive the Heartbeat for three times the 
-                // client will be closed
-                LogEventSource.Log.LogError($"Too many Heartbeat missed: {_missedHeartbeat}");
-                Close();
-                close($"Too many Heartbeat missed: {_missedHeartbeat}. " +
-                      $"Client connection will be closed");
-            };
+            _timer.Elapsed += TimerElapsed;
         }
     }
 
+    private void TimerElapsed(object sender, System.Timers.ElapsedEventArgs e)
+    {
+        _ = PerformHeartBeatAsync();
+    }
+
+    private async Task PerformHeartBeatAsync()
+    {
+        var f = _sendHeartbeatFunc();
+        await f.AsTask().WaitAsync(TimeSpan.FromMilliseconds(1000));
+
+        var seconds = (DateTime.Now - _lastUpdate).TotalSeconds;
+        if (seconds < _heartbeat)
+        {
+            return;
+        }
+
+        // missed the Heartbeat 
+        Interlocked.Increment(ref _missedHeartbeat);
+        LogEventSource.Log.LogWarning($"Heartbeat missed: {_missedHeartbeat}");
+        if (_missedHeartbeat <= 3)
+        {
+            return;
+        }
+
+        // When the client does not receive the Heartbeat for three times the 
+        // client will be closed
+        LogEventSource.Log.LogError($"Too many Heartbeat missed: {_missedHeartbeat}");
+        Close();
+        await _close($"Too many Heartbeat missed: {_missedHeartbeat}. " +
+              $"Client connection will be closed");
+    }
+
     internal void UpdateHeartBeat()
     {
         _lastUpdate = DateTime.Now;

RabbitMQ.Stream.Client/Producer.cs

@@ -178,7 +178,7 @@ internal void PreValidateBatch(List<(ulong, Message)> messages)
 
         private async Task SemaphoreWait()
         {
-            if (!semaphore.Wait(0) && !client.IsClosed)
+            if (!await semaphore.WaitAsync(0) && !client.IsClosed)
             {
                 // Nope, we have maxed our In-Flight messages, let's asynchronously wait for confirms
                 if (!await semaphore.WaitAsync(TimeSpan.FromSeconds(1)).ConfigureAwait(false))
@@ -249,11 +249,11 @@ private async Task ProcessBuffer()
             }
         }
 
-        public Task<ResponseCode> Close()
+        public async Task<ResponseCode> Close()
         {
             if (client.IsClosed)
             {
-                return Task.FromResult(ResponseCode.Ok);
+                return ResponseCode.Ok;
             }
 
             var result = ResponseCode.Ok;
@@ -264,7 +264,7 @@ public Task<ResponseCode> Close()
                 // in this case we reduce the waiting time
                 // the producer could be removed because of stream deleted 
                 // so it is not necessary to wait.
-                deletePublisherResponseTask.Wait(TimeSpan.FromSeconds(3));
+                await deletePublisherResponseTask.WaitAsync(TimeSpan.FromSeconds(3));
                 if (deletePublisherResponseTask.IsCompletedSuccessfully)
                 {
                     result = deletePublisherResponseTask.Result.ResponseCode;
@@ -277,7 +277,7 @@ public Task<ResponseCode> Close()
 
             var closed = client.MaybeClose($"client-close-publisher: {publisherId}");
             ClientExceptions.MaybeThrowException(closed.ResponseCode, $"client-close-publisher: {publisherId}");
-            return Task.FromResult(result);
+            return result;
         }
 
         public static async Task<Producer> Create(ClientParameters clientParameters,

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -396,7 +396,7 @@ RabbitMQ.Stream.Client.IOffsetType.OffsetType.get -> RabbitMQ.Stream.Client.Offs
 RabbitMQ.Stream.Client.IOffsetType.Size.get -> int
 RabbitMQ.Stream.Client.IOffsetType.Write(System.Span<byte> span) -> int
 RabbitMQ.Stream.Client.IRouting
-RabbitMQ.Stream.Client.IRouting.CreateClient(RabbitMQ.Stream.Client.ClientParameters clientParameters) -> RabbitMQ.Stream.Client.IClient
+RabbitMQ.Stream.Client.IRouting.CreateClient(RabbitMQ.Stream.Client.ClientParameters clientParameters) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IClient>
 RabbitMQ.Stream.Client.IRouting.ValidateDns.get -> bool
 RabbitMQ.Stream.Client.IRouting.ValidateDns.set -> void
 RabbitMQ.Stream.Client.Keywords
@@ -633,8 +633,8 @@ RabbitMQ.Stream.Client.Reliable.ConfirmationStatus.StreamNotAvailable = 6 -> Rab
 RabbitMQ.Stream.Client.Reliable.ConfirmationStatus.UndefinedError = 200 -> RabbitMQ.Stream.Client.Reliable.ConfirmationStatus
 RabbitMQ.Stream.Client.Reliable.ConfirmationStatus.WaitForConfirmation = 0 -> RabbitMQ.Stream.Client.Reliable.ConfirmationStatus
 RabbitMQ.Stream.Client.Reliable.IReconnectStrategy
-RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenConnected(string connectionInfo) -> void
-RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenDisconnected(string connectionInfo) -> bool
+RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenConnected(string connectionInfo) -> System.Threading.Tasks.ValueTask
+RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenDisconnected(string connectionInfo) -> System.Threading.Tasks.ValueTask<bool>
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation.InsertDateTime.get -> System.DateTime
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation.InsertDateTime.init -> void
@@ -707,7 +707,7 @@ RabbitMQ.Stream.Client.ResponseCode.SubscriptionIdDoesNotExist = 4 -> RabbitMQ.S
 RabbitMQ.Stream.Client.ResponseCode.UnknownFrame = 13 -> RabbitMQ.Stream.Client.ResponseCode
 RabbitMQ.Stream.Client.ResponseCode.VirtualHostAccessFailure = 12 -> RabbitMQ.Stream.Client.ResponseCode
 RabbitMQ.Stream.Client.Routing
-RabbitMQ.Stream.Client.Routing.CreateClient(RabbitMQ.Stream.Client.ClientParameters clientParameters) -> RabbitMQ.Stream.Client.IClient
+RabbitMQ.Stream.Client.Routing.CreateClient(RabbitMQ.Stream.Client.ClientParameters clientParameters) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IClient>
 RabbitMQ.Stream.Client.Routing.Routing() -> void
 RabbitMQ.Stream.Client.Routing.ValidateDns.get -> bool
 RabbitMQ.Stream.Client.Routing.ValidateDns.set -> void

RabbitMQ.Stream.Client/Reliable/IReconnectStrategy.cs

@@ -3,7 +3,7 @@
 // Copyright (c) 2007-2020 VMware, Inc.
 
 using System;
-using System.Threading;
+using System.Threading.Tasks;
 
 namespace RabbitMQ.Stream.Client.Reliable;
 
@@ -18,13 +18,13 @@ public interface IReconnectStrategy
     /// </summary>
     /// <param name="connectionInfo">Additional connection info. Just for logging</param>
     /// <returns>if True the client will be reconnected else closed</returns>
-    bool WhenDisconnected(string connectionInfo);
+    ValueTask<bool> WhenDisconnected(string connectionInfo);
 
     /// <summary>
     /// It is raised when the TCP client is connected successfully 
     /// </summary>
     /// <param name="connectionInfo">Additional connection info. Just for logging</param>
-    void WhenConnected(string connectionInfo);
+    ValueTask WhenConnected(string connectionInfo);
 }
 
 /// <summary>
@@ -36,17 +36,18 @@ internal class BackOffReconnectStrategy : IReconnectStrategy
 {
     private int Tentatives { get; set; } = 1;
 
-    public bool WhenDisconnected(string connectionInfo)
+    public async ValueTask<bool> WhenDisconnected(string connectionInfo)
     {
         Tentatives <<= 1;
         LogEventSource.Log.LogInformation(
             $"{connectionInfo} disconnected, check if reconnection needed in {Tentatives * 100} ms.");
-        Thread.Sleep(TimeSpan.FromMilliseconds(Tentatives * 100));
+        await Task.Delay(TimeSpan.FromMilliseconds(Tentatives * 100));
         return true;
     }
 
-    public void WhenConnected(string _)
+    public ValueTask WhenConnected(string connectionInfo)
     {
         Tentatives = 1;
+        return ValueTask.CompletedTask;
     }
 }

RabbitMQ.Stream.Client/Reliable/ReliableBase.cs

@@ -30,7 +30,7 @@ protected async Task TryToReconnect(IReconnectStrategy reconnectStrategy)
         _inReconnection = true;
         try
         {
-            var reconnect = reconnectStrategy.WhenDisconnected(ToString());
+            var reconnect = await reconnectStrategy.WhenDisconnected(ToString());
             var hasToReconnect = reconnect && _needReconnect;
             var addInfo = "Client won't reconnect";
             if (hasToReconnect)
@@ -72,7 +72,7 @@ internal async Task HandleMetaDataMaybeReconnect(string stream, StreamSystem sys
     {
         // This sleep is needed. When a stream is deleted it takes sometime.
         // The StreamExists/1 could return true even the stream doesn't exist anymore.
-        Thread.Sleep(500);
+        await Task.Delay(500);
         if (await system.StreamExists(stream))
         {
             LogEventSource.Log.LogInformation(

RabbitMQ.Stream.Client/Reliable/ReliableConsumer.cs

@@ -75,8 +75,7 @@ protected override async Task GetNewReliable(bool boot)
                 },
                 MetadataHandler = update =>
                 {
-                    HandleMetaDataMaybeReconnect(update.Stream,
-                        _reliableConsumerConfig.StreamSystem).Wait();
+                    _ = HandleMetaDataMaybeReconnect(update.Stream, _reliableConsumerConfig.StreamSystem);
                 },
                 MessageHandler = async (consumer, ctx, message) =>
                 {
@@ -88,7 +87,7 @@ protected override async Task GetNewReliable(bool boot)
                     }
                 }
             });
-            _reliableConsumerConfig.ReconnectStrategy.WhenConnected(ToString());
+            await _reliableConsumerConfig.ReconnectStrategy.WhenConnected(ToString());
         }
 
         catch (Exception e)

RabbitMQ.Stream.Client/Reliable/ReliableProducer.cs

@@ -86,8 +86,9 @@ protected override async Task GetNewReliable(bool boot)
                 MaxInFlight = _reliableProducerConfig.MaxInFlight,
                 MetadataHandler = update =>
                 {
-                    HandleMetaDataMaybeReconnect(update.Stream,
-                        _reliableProducerConfig.StreamSystem).Wait();
+                    // intentionally fire & forget
+                    _ = HandleMetaDataMaybeReconnect(update.Stream,
+                        _reliableProducerConfig.StreamSystem);
                 },
                 ConnectionClosedHandler = async _ =>
                 {
@@ -110,7 +111,7 @@ protected override async Task GetNewReliable(bool boot)
                         confirmationStatus);
                 }
             });
-            _reliableProducerConfig.ReconnectStrategy.WhenConnected(ToString());
+            await _reliableProducerConfig.ReconnectStrategy.WhenConnected(ToString());
             if (boot)
             {
                 // Init the publishing id

RabbitMQ.Stream.Client/RoutingClient.cs

@@ -6,26 +6,23 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Net;
-using System.Threading;
 using System.Threading.Tasks;
 
 namespace RabbitMQ.Stream.Client
 {
     public interface IRouting
     {
-        IClient CreateClient(ClientParameters clientParameters);
+        Task<IClient> CreateClient(ClientParameters clientParameters);
         bool ValidateDns { get; set; }
     }
 
     public class Routing : IRouting
     {
         public bool ValidateDns { get; set; } = true;
 
-        public IClient CreateClient(ClientParameters clientParameters)
+        public async Task<IClient> CreateClient(ClientParameters clientParameters)
         {
-            var taskClient = Client.Create(clientParameters);
-            taskClient.Wait(TimeSpan.FromSeconds(1));
-            return taskClient.Result;
+            return await Client.Create(clientParameters);
         }
     }
 
@@ -67,14 +64,14 @@ int maxAttempts
                 // In this case we just return the node (leader for producer, random for consumer)
                 // since there is not load balancer configuration
 
-                return routing.CreateClient(clientParameters with { Endpoint = endPointNoLb });
+                return await routing.CreateClient(clientParameters with { Endpoint = endPointNoLb });
             }
 
             // here it means that there is a AddressResolver configuration
             // so there is a load-balancer or proxy we need to get the right connection
             // as first we try with the first node given from the LB
             var endPoint = clientParameters.AddressResolver.EndPoint;
-            var client = routing.CreateClient(clientParameters with { Endpoint = endPoint });
+            var client = await routing.CreateClient(clientParameters with { Endpoint = endPoint });
 
             var advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");
             var advertisedPort = GetPropertyValue(client.ConnectionProperties, "advertised_port");
@@ -85,7 +82,7 @@ int maxAttempts
                 attemptNo++;
                 await client.Close("advertised_host or advertised_port doesn't match");
 
-                client = routing.CreateClient(clientParameters with { Endpoint = endPoint });
+                client = await routing.CreateClient(clientParameters with { Endpoint = endPoint });
 
                 advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");
                 advertisedPort = GetPropertyValue(client.ConnectionProperties, "advertised_port");
@@ -95,7 +92,7 @@ int maxAttempts
                         $"Could not find broker ({broker.Host}:{broker.Port}) after {maxAttempts} attempts");
                 }
 
-                Thread.Sleep(TimeSpan.FromMilliseconds(200));
+                await Task.Delay(TimeSpan.FromMilliseconds(200));
             }
 
             return client;
@@ -142,8 +139,7 @@ public static async Task<IClient> LookupRandomConnection(ClientParameters client
         {
             var brokers = new List<Broker>() { metaDataInfo.Leader };
             brokers.AddRange(metaDataInfo.Replicas);
-            var rnd = new Random();
-            brokers.Sort((_, _) => rnd.Next(-1, 1));
+            brokers.Sort((_, _) => Random.Shared.Next(-1, 1));
             var exceptions = new List<Exception>();
             foreach (var broker in brokers)
             {

Tests/ReliableTests.cs

@@ -448,14 +448,15 @@ public MyReconnection(ITestOutputHelper testOutputHelper)
             _testOutputHelper = testOutputHelper;
         }
 
-        public bool WhenDisconnected(string info)
+        ValueTask<bool> IReconnectStrategy.WhenDisconnected(string info)
         {
             _testOutputHelper.WriteLine($"MyReconnection WhenDisconnected {info}");
-            return false;
+            return ValueTask.FromResult(false);
         }
 
-        public void WhenConnected(string _)
+        ValueTask IReconnectStrategy.WhenConnected(string connectionInfo)
         {
+            return ValueTask.CompletedTask;
         }
     }