Implement DnsAddressResolver and ResolveAsync for DNS Load Balanced clusters

.github/release.yml

@@ -21,6 +21,9 @@ changelog:
     - title: 📚 Documentation
       labels:
         - documentation
+    - title: 🪫 Deprecation
+      labels:
+        - deprecation
     - title: 🔄 Dependency Updates
       labels:
         - dependencies

RabbitMQ.Stream.Client/AddressResolver.cs

@@ -2,7 +2,9 @@
 // 2.0, and the Mozilla Public License, version 2.0.
 // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 
+using System;
 using System.Net;
+using System.Threading.Tasks;
 
 namespace RabbitMQ.Stream.Client
 {
@@ -16,6 +18,8 @@ public AddressResolver(EndPoint endPoint)
 
         public EndPoint EndPoint { get; set; }
         public bool Enabled { get; set; }
-        public EndPoint Resolve(string address, int host) => EndPoint;
+        [Obsolete("Deprecated. Use ResolveAsync instead.")]
+        public EndPoint Resolve(string address, int port) => EndPoint;
+        public async Task<EndPoint> ResolveAsync(string address, int port) => EndPoint;
     }
 }

RabbitMQ.Stream.Client/AddressResolverDynamic.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Net;
+using System.Threading.Tasks;
 
 namespace RabbitMQ.Stream.Client;
 
@@ -18,5 +19,9 @@ public AddressResolverDynamic(Func<string, int, EndPoint> resolveFunction)
     }
 
     public bool Enabled { get; set; }
-    public EndPoint Resolve(string address, int host) => _resolveFunction(address, host);
+    [Obsolete("Deprecated. Use ResolveAsync instead.")]
+    public EndPoint Resolve(string address, int port) => _resolveFunction(address, port);
+#pragma warning disable CS0618
+    public Task<EndPoint> ResolveAsync(string address, int port) => Task.FromResult(Resolve(address, port));
+#pragma warning restore CS0618
 }

RabbitMQ.Stream.Client/DnsAddressResolver.cs

@@ -0,0 +1,31 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+// Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+
+using System;
+using System.Net;
+using System.Threading.Tasks;
+
+namespace RabbitMQ.Stream.Client
+{
+    public class DnsAddressResolver : IAddressResolver
+    {
+        public DnsAddressResolver(DnsEndPoint endPoint)
+        {
+            EndPoint = endPoint;
+            Enabled = true;
+        }
+
+        public EndPoint EndPoint { get; set; }
+        public bool Enabled { get; set; }
+        [Obsolete("Deprecated. Use ResolveAsync instead.")]
+        public EndPoint Resolve(string address, int port) => ResolveAsync(address, port).GetAwaiter().GetResult();
+        public async Task<EndPoint> ResolveAsync(string address, int port)
+        {
+            var entries = await Dns.GetHostEntryAsync(((DnsEndPoint)EndPoint).Host).ConfigureAwait(false);
+            var addressList = entries.AddressList;
+            var targetIp = addressList[Random.Shared.Next(addressList.Length)];
+            return new IPEndPoint(targetIp, port);
+        }
+    }
+}

RabbitMQ.Stream.Client/IAddressResolver.cs

@@ -3,11 +3,12 @@
 // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 
 using System.Net;
+using System.Threading.Tasks;
 
 namespace RabbitMQ.Stream.Client;
 
 public interface IAddressResolver
 {
     public bool Enabled { get; }
-    public EndPoint Resolve(string address, int host);
+    public Task<EndPoint> ResolveAsync(string address, int port);
 }

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -1,3 +1,11 @@
+RabbitMQ.Stream.Client.DnsAddressResolver
+RabbitMQ.Stream.Client.DnsAddressResolver.DnsAddressResolver(System.Net.DnsEndPoint endPoint) -> void
+RabbitMQ.Stream.Client.DnsAddressResolver.Enabled.get -> bool
+RabbitMQ.Stream.Client.DnsAddressResolver.Enabled.set -> void
+RabbitMQ.Stream.Client.DnsAddressResolver.EndPoint.get -> System.Net.EndPoint
+RabbitMQ.Stream.Client.DnsAddressResolver.EndPoint.set -> void
+RabbitMQ.Stream.Client.DnsAddressResolver.Resolve(string address, int port) -> System.Net.EndPoint
+RabbitMQ.Stream.Client.DnsAddressResolver.ResolveAsync(string address, int port) -> System.Threading.Tasks.Task<System.Net.EndPoint>
 abstract RabbitMQ.Stream.Client.AbstractEntity.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
 abstract RabbitMQ.Stream.Client.AbstractEntity.DeleteEntityFromTheServer(bool ignoreIfAlreadyDeleted = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
 abstract RabbitMQ.Stream.Client.AbstractEntity.DumpEntityConfiguration() -> string
@@ -29,12 +37,14 @@ RabbitMQ.Stream.Client.AddressResolver.Enabled.get -> bool
 RabbitMQ.Stream.Client.AddressResolver.Enabled.set -> void
 RabbitMQ.Stream.Client.AddressResolver.EndPoint.get -> System.Net.EndPoint
 RabbitMQ.Stream.Client.AddressResolver.EndPoint.set -> void
-RabbitMQ.Stream.Client.AddressResolver.Resolve(string address, int host) -> System.Net.EndPoint
+RabbitMQ.Stream.Client.AddressResolver.Resolve(string address, int port) -> System.Net.EndPoint
+RabbitMQ.Stream.Client.AddressResolver.ResolveAsync(string address, int port) -> System.Threading.Tasks.Task<System.Net.EndPoint>
 RabbitMQ.Stream.Client.AddressResolverDynamic
 RabbitMQ.Stream.Client.AddressResolverDynamic.AddressResolverDynamic(System.Func<string, int, System.Net.EndPoint> resolveFunction) -> void
 RabbitMQ.Stream.Client.AddressResolverDynamic.Enabled.get -> bool
 RabbitMQ.Stream.Client.AddressResolverDynamic.Enabled.set -> void
-RabbitMQ.Stream.Client.AddressResolverDynamic.Resolve(string address, int host) -> System.Net.EndPoint
+RabbitMQ.Stream.Client.AddressResolverDynamic.Resolve(string address, int port) -> System.Net.EndPoint
+RabbitMQ.Stream.Client.AddressResolverDynamic.ResolveAsync(string address, int port) -> System.Threading.Tasks.Task<System.Net.EndPoint>
 RabbitMQ.Stream.Client.AlreadyClosedException
 RabbitMQ.Stream.Client.AlreadyClosedException.AlreadyClosedException(string s) -> void
 RabbitMQ.Stream.Client.AuthMechanism
@@ -169,7 +179,7 @@ RabbitMQ.Stream.Client.HashRoutingMurmurStrategy.Route(RabbitMQ.Stream.Client.Me
 RabbitMQ.Stream.Client.HeartBeatHandler.HeartBeatHandler(System.Func<System.Threading.Tasks.ValueTask<bool>> sendHeartbeatFunc, System.Func<string, string, System.Threading.Tasks.Task<RabbitMQ.Stream.Client.CloseResponse>> close, int heartbeat, Microsoft.Extensions.Logging.ILogger<RabbitMQ.Stream.Client.HeartBeatHandler> logger = null) -> void
 RabbitMQ.Stream.Client.IAddressResolver
 RabbitMQ.Stream.Client.IAddressResolver.Enabled.get -> bool
-RabbitMQ.Stream.Client.IAddressResolver.Resolve(string address, int host) -> System.Net.EndPoint
+RabbitMQ.Stream.Client.IAddressResolver.ResolveAsync(string address, int port) -> System.Threading.Tasks.Task<System.Net.EndPoint>
 RabbitMQ.Stream.Client.IClient.ClientId.get -> string
 RabbitMQ.Stream.Client.IClient.ClientId.init -> void
 RabbitMQ.Stream.Client.IClient.Consumers.get -> System.Collections.Generic.IDictionary<byte, (string, RabbitMQ.Stream.Client.ConsumerEvents)>

RabbitMQ.Stream.Client/RoutingClient.cs

@@ -79,7 +79,7 @@ clientParameters with
             // 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.Resolve(broker.Host, (int)broker.Port);
+            var endPoint = await clientParameters.AddressResolver.ResolveAsync(broker.Host, (int)broker.Port).ConfigureAwait(false);
             var client = await routing
                 .CreateClient(
                     clientParameters with
@@ -94,8 +94,10 @@ clientParameters with
 
             var attemptNo = 0;
             while (broker.Host != advertisedHost || broker.Port != uint.Parse(advertisedPort))
+
             {
-                logger?.LogDebug(
+                endPoint = await clientParameters.AddressResolver.ResolveAsync(broker.Host, (int)broker.Port).ConfigureAwait(false);
+                logger?.LogInformation(
                     "advertised_host or advertised_port doesn't match. Expected: {ExpectedHost}:{ExpectedPort}, " +
                     "Actual: {AdvertisedHost}:{AdvertisedPort}. Attempt number: {AttemptNo}/{MaxAttempts}",
                     broker.Host, broker.Port, advertisedHost, advertisedPort, attemptNo, maxAttempts);