fix connection name

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

RabbitMQ.Stream.Client/RoutingClient.cs

@@ -67,15 +67,23 @@ 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 }, 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 }, 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");
@@ -87,7 +95,11 @@ 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 }, broker, logger)
+                client = await routing.CreateClient(clientParameters with
+                {
+                    Endpoint = endPoint,
+                    ClientProvidedName = clientParameters.ClientProvidedName
+                }, broker, logger)
                     .ConfigureAwait(false);
 
                 advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");

docs/ReliableClient/Program.cs

@@ -10,13 +10,13 @@
     ProducersPerConnection = 2, 
     ConsumersPerConnection = 2,
     Host = "Node0",
-    Port = 5562,
-    // LoadBalancer = true,
+    Port = 5553,
+    LoadBalancer = true,
     SuperStream = true,
-    Streams = 1,
-    Producers = 2,
+    Streams = 3,
+    Producers = 4,
     MessagesPerProducer = 50_000_000,
-    Consumers = 3
+    Consumers = 4
     // Username = "test",
     // Password = "test"
 });

docs/ReliableClient/RClient.cs

@@ -48,6 +48,7 @@ public static async Task Start(Config config)
         var loggerFactory = serviceCollection.BuildServiceProvider()
             .GetService<ILoggerFactory>();
 
+
         if (loggerFactory != null)
         {
             var lp = loggerFactory.CreateLogger<Producer>();
@@ -84,7 +85,7 @@ public static async Task Start(Config config)
 
             if (config.LoadBalancer)
             {
-                var resolver = new AddressResolver(new IPEndPoint(IPAddress.Parse(config.Host), config.Port));
+                var resolver = new AddressResolver(ep);
                 streamConf = new StreamSystemConfig()
                 {
                     AddressResolver = resolver,
@@ -114,7 +115,6 @@ public static async Task Start(Config config)
             var totalError = 0;
             var totalConsumed = 0;
             var totalSent = 0;
-            var totalSentToSuperStream = 0;
             var isRunning = true;
 
             _ = Task.Run(() =>
@@ -124,12 +124,11 @@ public static async Task Start(Config config)
                     Console.WriteLine(
                         $"When: {DateTime.Now}, " +
                         $"Tr {System.Diagnostics.Process.GetCurrentProcess().Threads.Count}, " +
+                        $"Sent: {totalSent:#,##0.00}, " +
                         $"Conf: {totalConfirmed:#,##0.00}, " +
                         $"Error: {totalError:#,##0.00}, " +
                         $"Total: {(totalConfirmed + totalError):#,##0.00}, " +
                         $"Consumed: {totalConsumed:#,##0.00}, " +
-                        $"Sent: {totalSent:#,##0.00}, " +
-                        $"Sent To SuperStream: {totalSentToSuperStream:#,##0.00}, " +
                         $"Sent per stream: {totalSent / streamsList.Count}");
                     Thread.Sleep(5000);
                 }
@@ -180,6 +179,11 @@ await system.CreateStream(new StreamSpec(stream) {MaxLengthBytes = 30_000_000_00
                         await Consumer.Create(conf, lc).ConfigureAwait(false));
                 }
 
+                async Task MaybeSend(Producer producer, Message message, ManualResetEvent publishEvent)
+                {
+                    publishEvent.WaitOne();
+                    await producer.Send(message).ConfigureAwait(false);
+                }
 
                 for (var z = 0; z < config.Producers; z++)
                 {
@@ -215,12 +219,12 @@ await system.CreateStream(new StreamSpec(stream) {MaxLengthBytes = 30_000_000_00
                             var streamInfo = status.Partition is not null
                                 ? $" Partition {status.Partition} of super stream: {status.Stream}"
                                 : $"Stream: {status.Stream}";
-                            
+
                             lp.LogInformation("Producer: {Id} - status changed from {From} to {To}. {Info}",
-                                status.Identifier, 
+                                status.Identifier,
                                 status.From,
                                 status.To, streamInfo);
-                            
+
                             if (status.To == ReliableEntityStatus.Open)
                             {
                                 publishEvent.Set();
@@ -238,14 +242,13 @@ await system.CreateStream(new StreamSpec(stream) {MaxLengthBytes = 30_000_000_00
 
                         for (var i = 0; i < config.MessagesPerProducer; i++)
                         {
-                            publishEvent.WaitOne();
                             if (!unconfirmedMessages.IsEmpty)
                             {
                                 var msgs = unconfirmedMessages.ToArray();
                                 unconfirmedMessages.Clear();
                                 foreach (var msg in msgs)
                                 {
-                                    await producer.Send(msg).ConfigureAwait(false);
+                                    await MaybeSend(producer, msg, publishEvent).ConfigureAwait(false);
                                     Interlocked.Increment(ref totalSent);
                                 }
                             }
@@ -254,8 +257,7 @@ await system.CreateStream(new StreamSpec(stream) {MaxLengthBytes = 30_000_000_00
                             {
                                 Properties = new Properties() {MessageId = $"hello{i}"}
                             };
-                            await producer.Send(message).ConfigureAwait(false);
-                            await Task.Delay(10).ConfigureAwait(false);
+                            await MaybeSend(producer, message, publishEvent).ConfigureAwait(false);
                             Interlocked.Increment(ref totalSent);
                         }
                     });
@@ -270,6 +272,9 @@ await system.CreateStream(new StreamSpec(stream) {MaxLengthBytes = 30_000_000_00
             producersList.ForEach(async p => await p.Close().ConfigureAwait(false));
             Console.WriteLine("closing the consumers ..... ");
             consumersList.ForEach(async c => await c.Close().ConfigureAwait(false));
+
+            Console.WriteLine("Press any key to close all");
+            Console.ReadKey();
         }
 
         Console.WriteLine("Closed all the consumers and producers");