Consumer connected only to the followers

RabbitMQ.Stream.Client/RoutingClient.cs

@@ -67,23 +67,27 @@ internal static async Task<IClient> LookupConnection(
                 // In this case we just return the node (leader for producer, random for consumer)
                 // since there is not load balancer configuration
 
-                return await routing.CreateClient(clientParameters with
-                {
-                    Endpoint = endPointNoLb,
-                    ClientProvidedName = clientParameters.ClientProvidedName
-                }, broker, logger)
+                return await routing
+                    .CreateClient(
+                        clientParameters with
+                        {
+                            Endpoint = endPointNoLb,
+                            ClientProvidedName = clientParameters.ClientProvidedName
+                        }, broker, logger)
                     .ConfigureAwait(false);
             }
 
             // 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 = await routing.CreateClient(clientParameters with
-            {
-                Endpoint = endPoint,
-                ClientProvidedName = clientParameters.ClientProvidedName
-            }, broker, logger)
+            var client = await routing
+                .CreateClient(
+                    clientParameters with
+                    {
+                        Endpoint = endPoint,
+                        ClientProvidedName = clientParameters.ClientProvidedName
+                    }, broker, logger)
                 .ConfigureAwait(false);
 
             var advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");
@@ -95,11 +99,13 @@ internal static async Task<IClient> LookupConnection(
                 attemptNo++;
                 await client.Close("advertised_host or advertised_port doesn't match").ConfigureAwait(false);
 
-                client = await routing.CreateClient(clientParameters with
-                {
-                    Endpoint = endPoint,
-                    ClientProvidedName = clientParameters.ClientProvidedName
-                }, broker, logger)
+                client = await routing
+                    .CreateClient(
+                        clientParameters with
+                        {
+                            Endpoint = endPoint,
+                            ClientProvidedName = clientParameters.ClientProvidedName
+                        }, broker, logger)
                     .ConfigureAwait(false);
 
                 advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");
@@ -180,8 +186,14 @@ await LookupConnection(clientParameters, metaDataInfo.Leader, MaxAttempts(metaDa
         public static async Task<IClient> LookupRandomConnection(ClientParameters clientParameters,
             StreamInfo metaDataInfo, ConnectionsPool pool, ILogger logger = null)
         {
-            var brokers = new List<Broker>() { metaDataInfo.Leader };
+            var brokers = new List<Broker>() { };
+            if (metaDataInfo.Replicas is { Count: <= 0 })
+            {
+                brokers.Add(metaDataInfo.Leader);
+            }
+
             brokers.AddRange(metaDataInfo.Replicas);
+
             var exceptions = new List<Exception>();
             var br = brokers.OrderBy(x => Random.Shared.Next()).ToList();
 

Tests/UnitTests.cs

@@ -107,7 +107,7 @@ public Task<IClient> CreateClient(ClientParameters clientParameters, Broker brok
         public bool ValidateDns { get; set; } = false;
     }
 
-    public class ReplicaRouting : IRouting
+    public class LeaderRouting : IRouting
     {
         public Task<IClient> CreateClient(ClientParameters clientParameters, Broker broker, ILogger logger = null)
         {
@@ -125,6 +125,25 @@ public Task<IClient> CreateClient(ClientParameters clientParameters, Broker brok
         public bool ValidateDns { get; set; } = false;
     }
 
+    public class ReplicaseRouting : IRouting
+    {
+        public Task<IClient> CreateClient(ClientParameters clientParameters, Broker broker, ILogger logger = null)
+        {
+            var fake = new FakeClient(clientParameters)
+            {
+                ConnectionProperties = new Dictionary<string, string>()
+                {
+
+                    ["advertised_port"] = "5553",
+                    ["advertised_host"] = "replica2"
+                }
+            };
+            return Task.FromResult<IClient>(fake);
+        }
+
+        public bool ValidateDns { get; set; } = false;
+    }
+
     // This class is only for unit tests
     public class UnitTests
     {
@@ -214,14 +233,34 @@ public void RandomReplicaLeader()
             var metaDataInfo = new StreamInfo("stream", ResponseCode.Ok, new Broker("leader", 5552),
                 new List<Broker>());
             var client =
-                RoutingHelper<ReplicaRouting>.LookupRandomConnection(clientParameters, metaDataInfo,
+                RoutingHelper<LeaderRouting>.LookupRandomConnection(clientParameters, metaDataInfo,
                     new ConnectionsPool(1, 1));
             Assert.Equal("5552", client.Result.ConnectionProperties["advertised_port"]);
             var res = (client.Result.ConnectionProperties["advertised_host"] == "leader" ||
                        client.Result.ConnectionProperties["advertised_host"] == "replica");
             Assert.True(res);
         }
 
+        [Fact]
+        public void RandomOnlyReplicaIfThereAre()
+        {
+            // this test is not completed yet should add also some replicas
+            var addressResolver = new AddressResolver(new IPEndPoint(IPAddress.Parse("192.168.10.99"), 5552));
+            var clientParameters = new ClientParameters() { AddressResolver = addressResolver, };
+            var metaDataInfo = new StreamInfo("stream",
+                ResponseCode.Ok, new Broker("leader", 5552),
+                new List<Broker>()
+                {
+                    new Broker("replica2", 5553),
+                });
+            var client =
+                RoutingHelper<ReplicaseRouting>.LookupRandomConnection(clientParameters, metaDataInfo,
+                    new ConnectionsPool(1, 1));
+            Assert.Equal("5553", client.Result.ConnectionProperties["advertised_port"]);
+            var res = (client.Result.ConnectionProperties["advertised_host"] == "replica2");
+            Assert.True(res);
+        }
+
         [Fact]
         public void CompressUnCompressShouldHaveTheSize()
         {

docs/ReliableClient/Program.cs

@@ -8,13 +8,13 @@
 var rClient = RClient.Start(new RClient.Config()
 {
     ProducersPerConnection = 2, 
-    ConsumersPerConnection = 2,
-    Host = "Node0",
+    ConsumersPerConnection = 100,
+    Host = "node1",
     Port = 5553,
     LoadBalancer = true,
-    SuperStream = true,
-    Streams = 1,
-    Producers = 1,
+    SuperStream = false,
+    Streams = 10,
+    Producers = 4,
     MessagesPerProducer = 50_000_000,
     Consumers = 4
     // Username = "test",

docs/ReliableClient/RClient.cs

@@ -91,6 +91,11 @@ public static async Task Start(Config config)
                     AddressResolver = resolver,
                     UserName = config.Username,
                     Password = config.Password,
+                    ConnectionPoolConfig = new ConnectionPoolConfig()
+                    {
+                        ProducersPerConnection = config.ProducersPerConnection,
+                        ConsumersPerConnection = config.ConsumersPerConnection,
+                    },
                     Endpoints = new List<EndPoint>() {resolver.EndPoint}
                 };
             }