Remove code duplication to handle the pool (#356)

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

Author: Gsantomaggio

RabbitMQ.Stream.Client/ConnectionsPool.cs

@@ -145,14 +145,13 @@ internal async Task<IClient> GetOrCreateClient(string brokerInfo, Func<Task<ICli
             // do we have a connection for this brokerInfo and with free slots for producer or consumer?
             // it does not matter which connection is available 
             // the important is to have a connection available for the brokerInfo
-            var count = Connections.Values.Count(x => x.BrokerInfo == brokerInfo && x.Available);
+            var connectionItems = Connections.Values.Where(x => x.BrokerInfo == brokerInfo && x.Available).ToList();
 
-            if (count > 0)
+            if (connectionItems.Any())
             {
                 // ok we have a connection available for this brokerInfo
                 // let's get the first one
-                // TODO: we can improve this by getting the connection with the less active items
-                var connectionItem = Connections.Values.First(x => x.BrokerInfo == brokerInfo && x.Available);
+                var connectionItem = connectionItems.OrderBy(x => x.EntitiesCount).First();
                 connectionItem.LastUsed = DateTime.UtcNow;
 
                 if (connectionItem.Client is not { IsClosed: true })
@@ -162,11 +161,11 @@ internal async Task<IClient> GetOrCreateClient(string brokerInfo, Func<Task<ICli
                 // let's remove it from the pool
                 Connections.TryRemove(connectionItem.Client.ClientId, out _);
                 // let's create a new one
-                connectionItem = new ConnectionItem(brokerInfo, _idsPerConnection,
+                var newConnectionItem = new ConnectionItem(brokerInfo, _idsPerConnection,
                     await createClient().ConfigureAwait(false));
                 Connections.TryAdd(connectionItem.Client.ClientId, connectionItem);
 
-                return connectionItem.Client;
+                return newConnectionItem.Client;
             }
 
             if (_maxConnections > 0 && Connections.Count >= _maxConnections)

RabbitMQ.Stream.Client/RawSuperStreamProducer.cs

@@ -311,16 +311,16 @@ public Task<ResponseCode> Close()
     //<summary>
     /// Returns lower from the LastPublishingId for all the producers
     // </summary> 
-    public Task<ulong> GetLastPublishingId()
+    public async Task<ulong> GetLastPublishingId()
     {
         foreach (var stream in _streamInfos.Keys.ToList())
         {
-            MaybeAddAndGetProducer(stream).Wait();
+            await MaybeAddAndGetProducer(stream).ConfigureAwait(false);
         }
 
         var v = _producers.Values.Min(p => p.GetLastPublishingId().Result);
 
-        return Task.FromResult(v);
+        return v;
     }
 
     public bool IsOpen()

docs/ReliableClient/Program.cs

@@ -12,11 +12,11 @@
     Host = "node0",
     Port = 5562,
     LoadBalancer = false,
-    SuperStream = false,
+    SuperStream = true,
     Streams = 3,
-    Producers = 10,
+    Producers = 2,
     MessagesPerProducer = 10_000_000,
-    Consumers = 10,
+    Consumers = 5,
     // Username = "test",
     // Password = "test"
 });

docs/ReliableClient/RClient.cs

@@ -43,7 +43,7 @@ public static async Task Start(Config config)
                 options.TimestampFormat = "[HH:mm:ss] ";
                 options.ColorBehavior = LoggerColorBehavior.Default;
             })
-            .AddFilter(level => level >= LogLevel.Information)
+            .AddFilter(level => level >= LogLevel.Debug)
         );
         var loggerFactory = serviceCollection.BuildServiceProvider()
             .GetService<ILoggerFactory>();
@@ -261,7 +261,7 @@ async Task MaybeSend(Producer producer, Message message, ManualResetEvent publis
                                 Properties = new Properties() {MessageId = $"hello{i}"}
                             };
                             await MaybeSend(producer, message, publishEvent).ConfigureAwait(false);
-                            await Task.Delay(20).ConfigureAwait(false);
+                            await Task.Delay(320).ConfigureAwait(false);
                             Interlocked.Increment(ref totalSent);
                         }
                     });