Enrich SocketException when opening a connection (#124)

* Enrich socketExceptions; report aggregate exceptions when connecting to consumer nodes

* Change thrown exception type

* Use the Exception.Data dictionary to add host/ip and port info

Co-authored-by: Ricard Siliuk <[email protected]>
Co-authored-by: Luke Bakken <[email protected]>

Author: 3 people

RabbitMQ.Stream.Client/Connection.cs

@@ -52,7 +52,7 @@ private Connection(Socket socket, Func<Memory<byte>, Task> callback,
             Task.Run(ProcessIncomingFrames);
         }
 
-        public static async Task<Connection> Create(EndPoint ipEndpoint, Func<Memory<byte>, Task> commandCallback,
+        public static async Task<Connection> Create(EndPoint endpoint, Func<Memory<byte>, Task> commandCallback,
             Func<string, Task> closedCallBack, SslOption sslOption)
         {
             var socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
@@ -61,7 +61,27 @@ public static async Task<Connection> Create(EndPoint ipEndpoint, Func<Memory<byt
             socket.SendBufferSize *= 10;
             socket.ReceiveBufferSize *= 10;
 
-            await socket.ConnectAsync(ipEndpoint);
+            try
+            {
+                await socket.ConnectAsync(endpoint);
+            }
+            catch (SocketException ex)
+            {
+                switch (endpoint)
+                {
+                    case IPEndPoint ipEndPoint:
+                        ex.Data.Add("Address", ipEndPoint.Address);
+                        ex.Data.Add("Port", ipEndPoint.Port);
+                        throw;
+                    case DnsEndPoint dnsEndPoint:
+                        ex.Data.Add("Host", dnsEndPoint.Host);
+                        ex.Data.Add("Port", dnsEndPoint.Port);
+                        throw;
+                    default:
+                        throw;
+                }
+            }
+
             return new Connection(socket, commandCallback, closedCallBack, sslOption);
         }
 

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -853,7 +853,7 @@ static RabbitMQ.Stream.Client.AMQP.Properties.Parse(System.Buffers.ReadOnlySeque
 static RabbitMQ.Stream.Client.Client.Create(RabbitMQ.Stream.Client.ClientParameters parameters) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Client>
 static RabbitMQ.Stream.Client.CompressionHelper.Compress(System.Collections.Generic.List<RabbitMQ.Stream.Client.Message> messages, RabbitMQ.Stream.Client.CompressionType compressionType) -> RabbitMQ.Stream.Client.ICompressionCodec
 static RabbitMQ.Stream.Client.CompressionHelper.UnCompress(RabbitMQ.Stream.Client.CompressionType compressionType, System.Buffers.ReadOnlySequence<byte> source, uint dataLen, uint unCompressedDataSize) -> System.Buffers.ReadOnlySequence<byte>
-static RabbitMQ.Stream.Client.Connection.Create(System.Net.EndPoint ipEndpoint, System.Func<System.Memory<byte>, System.Threading.Tasks.Task> commandCallback, System.Func<string, System.Threading.Tasks.Task> closedCallBack, RabbitMQ.Stream.Client.SslOption sslOption) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Connection>
+static RabbitMQ.Stream.Client.Connection.Create(System.Net.EndPoint endpoint, System.Func<System.Memory<byte>, System.Threading.Tasks.Task> commandCallback, System.Func<string, System.Threading.Tasks.Task> closedCallBack, RabbitMQ.Stream.Client.SslOption sslOption) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Connection>
 static RabbitMQ.Stream.Client.Consumer.Create(RabbitMQ.Stream.Client.ClientParameters clientParameters, RabbitMQ.Stream.Client.ConsumerConfig config, RabbitMQ.Stream.Client.StreamInfo metaStreamInfo) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Consumer>
 static RabbitMQ.Stream.Client.LeaderLocator.ClientLocal.get -> RabbitMQ.Stream.Client.LeaderLocator
 static RabbitMQ.Stream.Client.LeaderLocator.LeastLeaders.get -> RabbitMQ.Stream.Client.LeaderLocator

RabbitMQ.Stream.Client/RoutingClient.cs

@@ -69,16 +69,6 @@ int maxAttempts
             endPoint = clientParameters.AddressResolver.EndPoint;
             var client = routing.CreateClient(clientParameters with { Endpoint = endPoint });
 
-            string GetPropertyValue(IDictionary<string, string> connectionProperties, string key)
-            {
-                if (!connectionProperties.TryGetValue(key, out var value))
-                {
-                    throw new RoutingClientException($"can't lookup {key}");
-                }
-
-                return value;
-            }
-
             var advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");
             var advertisedPort = GetPropertyValue(client.ConnectionProperties, "advertised_port");
 
@@ -118,6 +108,16 @@ private static int MaxAttempts(StreamInfo metaDataInfo)
                 2); // Pow just to be sure that the LoadBalancer will ping all the nodes
         }
 
+        private static string GetPropertyValue(IDictionary<string, string> connectionProperties, string key)
+        {
+            if (!connectionProperties.TryGetValue(key, out var value))
+            {
+                throw new RoutingClientException($"can't lookup {key}");
+            }
+
+            return value;
+        }
+
         /// <summary>
         /// Gets the leader connection. The producer must connect to the leader. 
         /// </summary>
@@ -135,16 +135,22 @@ public static async Task<IClient> LookupRandomConnection(ClientParameters client
         {
             var brokers = new List<Broker>() { metaDataInfo.Leader };
             brokers.AddRange(metaDataInfo.Replicas);
-
-            // pick one random node from leader and replicas.
-            // we need to use the LookupConnection since not all the nodes can have 
-            // replicas. For example if AddressResolver is configured the client could be
-            // connected to a node without replicas.
             var rnd = new Random();
-            var brokerId = rnd.Next(0, brokers.Count);
-            var broker = brokers[brokerId];
+            brokers.Sort((_, _) => rnd.Next(-1, 1));
+            var exceptions = new List<Exception>();
+            foreach (var broker in brokers)
+            {
+                try
+                {
+                    return await LookupConnection(clientParameters, broker, MaxAttempts(metaDataInfo));
+                }
+                catch (Exception ex)
+                {
+                    exceptions.Add(ex);
+                }
+            }
 
-            return await LookupConnection(clientParameters, broker, MaxAttempts(metaDataInfo));
+            throw new AggregateException("None of the brokers could be reached", exceptions);
         }
     }
 

Tests/UnitTests.cs

@@ -119,17 +119,30 @@ public IClient CreateClient(ClientParameters clientParameters)
     public class UnitTests
     {
         [Fact]
-        public void AddressResolverShouldRaiseAnExceptionIfAdvIsNull()
+        public async Task GiveProperExceptionWhenUnableToConnect()
+        {
+            var clientParameters = new ClientParameters
+            {
+                // if using `Routing` and not a purpose-built impl of `IRouting`,
+                // it's enough to specify an endpoint that won't be reached.
+                // Can be replaced with `IRouting` impl that throws on create.
+                Endpoint = new DnsEndPoint("localhost", 3939)
+            };
+            var metaDataInfo = new StreamInfo("stream", ResponseCode.Ok, new Broker("localhost", 3939), new List<Broker>());
+            await Assert.ThrowsAsync<AggregateException>(() => RoutingHelper<Routing>.LookupRandomConnection(clientParameters, metaDataInfo));
+        }
+
+        [Fact]
+        public async Task AddressResolverShouldRaiseAnExceptionIfAdvIsNull()
         {
             var addressResolver = new AddressResolver(new IPEndPoint(IPAddress.Loopback, 5552));
             var clientParameters = new ClientParameters()
             {
                 AddressResolver = addressResolver,
             };
-            var metaDataInfo = new StreamInfo("stream", ResponseCode.Ok, new Broker("leader", 5552),
-                null);
+            var metaDataInfo = new StreamInfo("stream", ResponseCode.Ok, new Broker("leader", 5552), new List<Broker>());
             // run more than one time just to be sure to use all the IP with random
-            Assert.ThrowsAsync<RoutingClientException>(() =>
+            await Assert.ThrowsAsync<RoutingClientException>(() =>
                 RoutingHelper<MissingFieldsRouting>.LookupLeaderConnection(clientParameters, metaDataInfo));
         }
 
@@ -153,7 +166,7 @@ public void AddressResolverLoadBalancerSimulate()
         }
 
         [Fact]
-        public void RoutingHelperShouldThrowIfLoadBalancerIsMisconfigured()
+        public async Task RoutingHelperShouldThrowIfLoadBalancerIsMisconfigured()
         {
             var addressResolver = new AddressResolver(new IPEndPoint(IPAddress.Parse("192.168.10.99"), 5552));
             var clientParameters = new ClientParameters()
@@ -163,7 +176,7 @@ public void RoutingHelperShouldThrowIfLoadBalancerIsMisconfigured()
             var metaDataInfo = new StreamInfo("stream", ResponseCode.Ok, new Broker("node2", 5552),
                 new List<Broker>() { new Broker("replica", 5552) });
 
-            Assert.ThrowsAsync<RoutingClientException>(
+            await Assert.ThrowsAsync<RoutingClientException>(
                 () => RoutingHelper<MisconfiguredLoadBalancerRouting>.LookupLeaderConnection(clientParameters, metaDataInfo));
         }