Validate the pool is not null

and the values are valid.
Add ProducerInfo to better understand the logs.
Move dispose function to the abstract entity to remove code duplication.
Better logs in case of errors.

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

Author: Gsantomaggio

RabbitMQ.Stream.Client/StreamSystem.cs

@@ -14,6 +14,24 @@ namespace RabbitMQ.Stream.Client
 {
     public record StreamSystemConfig : INamedEntity
     {
+        internal void Validate()
+        {
+            if (ConnectionPoolConfig is null)
+            {
+                throw new ArgumentException("ConnectionPoolConfig can't be null");
+            }
+
+            if (ConnectionPoolConfig.MaxConsumersConnections < 0)
+            {
+                throw new ArgumentException("MaxConsumersConnections can't be negative");
+            }
+
+            if (ConnectionPoolConfig.MaxProducersConnections < 0)
+            {
+                throw new ArgumentException("MaxProducersConnections can't be negative");
+            }
+        }
+
         public string UserName { get; set; } = "guest";
         public string Password { get; set; } = "guest";
         public string VirtualHost { get; set; } = "/";
@@ -24,7 +42,7 @@ public record StreamSystemConfig : INamedEntity
         /// </summary>
         public SslOption Ssl { get; set; } = new();
 
-        public IList<EndPoint> Endpoints { get; set; } = new List<EndPoint> { new IPEndPoint(IPAddress.Loopback, 5552) };
+        public IList<EndPoint> Endpoints { get; set; } = new List<EndPoint> {new IPEndPoint(IPAddress.Loopback, 5552)};
 
         public AddressResolver AddressResolver { get; set; }
         public string ClientProvidedName { get; set; } = "dotnet-stream-locator";
@@ -59,6 +77,7 @@ private StreamSystem(ClientParameters clientParameters, Client client,
 
         public static async Task<StreamSystem> Create(StreamSystemConfig config, ILogger<StreamSystem> logger = null)
         {
+            config.Validate();
             var clientParams = new ClientParameters
             {
                 UserName = config.UserName,
@@ -76,7 +95,7 @@ public static async Task<StreamSystem> Create(StreamSystemConfig config, ILogger
             {
                 try
                 {
-                    var client = await Client.Create(clientParams with { Endpoint = endPoint }, logger)
+                    var client = await Client.Create(clientParams with {Endpoint = endPoint}, logger)
                         .ConfigureAwait(false);
                     if (!client.IsClosed)
                     {
@@ -183,7 +202,7 @@ public async Task<IProducer> CreateRawSuperStreamProducer(
 
             var r = RawSuperStreamProducer.Create(rawSuperStreamProducerConfig,
                 streamInfos,
-                _clientParameters with { ClientProvidedName = rawSuperStreamProducerConfig.ClientProvidedName },
+                _clientParameters with {ClientProvidedName = rawSuperStreamProducerConfig.ClientProvidedName},
                 logger);
             _logger?.LogDebug("Raw Producer: {ProducerReference} created for SuperStream: {SuperStream}",
                 rawSuperStreamProducerConfig.Reference,
@@ -234,7 +253,7 @@ public async Task<IConsumer> CreateSuperStreamConsumer(
 
             var s = RawSuperStreamConsumer.Create(rawSuperStreamConsumerConfig,
                 streamInfos,
-                _clientParameters with { ClientProvidedName = rawSuperStreamConsumerConfig.ClientProvidedName },
+                _clientParameters with {ClientProvidedName = rawSuperStreamConsumerConfig.ClientProvidedName},
                 logger);
             _logger?.LogDebug("Consumer: {Reference} created for SuperStream: {SuperStream}",
                 rawSuperStreamConsumerConfig.Reference, rawSuperStreamConsumerConfig.SuperStream);
@@ -263,7 +282,7 @@ public async Task<IProducer> CreateRawProducer(RawProducerConfig rawProducerConf
                 await _semClientProvidedName.WaitAsync().ConfigureAwait(false);
                 rawProducerConfig.Pool = PoolProducers;
                 var p = await RawProducer.Create(
-                    _clientParameters with { ClientProvidedName = rawProducerConfig.ClientProvidedName },
+                    _clientParameters with {ClientProvidedName = rawProducerConfig.ClientProvidedName},
                     rawProducerConfig, metaStreamInfo, logger).ConfigureAwait(false);
                 _logger?.LogDebug("Raw Producer: {Reference} created for Stream: {Stream}",
                     rawProducerConfig.Reference, rawProducerConfig.Stream);
@@ -288,7 +307,7 @@ private async Task<StreamInfo> StreamInfo(string streamName)
                 var clientParametersEndpoint = _clientParameters.Endpoints[0];
                 switch (clientParametersEndpoint)
                 {
-                    case DnsEndPoint { Host: "localhost" } dnsEndPoint:
+                    case DnsEndPoint {Host: "localhost"} dnsEndPoint:
                         forceLocalHost = true;
                         localPort = dnsEndPoint.Port;
                         break;
@@ -305,12 +324,12 @@ private async Task<StreamInfo> StreamInfo(string streamName)
                 // craft the metadata response to force using localhost
                 var leader = new Broker("localhost", (uint)localPort);
                 metaStreamInfo = new StreamInfo(streamName, ResponseCode.Ok, leader,
-                    new List<Broker>(1) { leader });
+                    new List<Broker>(1) {leader});
             }
             else
             {
                 await MayBeReconnectLocator().ConfigureAwait(false);
-                var meta = await _client.QueryMetadata(new[] { streamName }).ConfigureAwait(false);
+                var meta = await _client.QueryMetadata(new[] {streamName}).ConfigureAwait(false);
                 metaStreamInfo = meta.StreamInfos[streamName];
             }
 
@@ -411,7 +430,7 @@ public async Task<IConsumer> CreateRawConsumer(RawConsumerConfig rawConsumerConf
             {
                 await _semClientProvidedName.WaitAsync().ConfigureAwait(false);
                 rawConsumerConfig.Pool = PoolConsumers;
-                var s = _clientParameters with { ClientProvidedName = rawConsumerConfig.ClientProvidedName };
+                var s = _clientParameters with {ClientProvidedName = rawConsumerConfig.ClientProvidedName};
                 var c = await RawConsumer.Create(s,
                     rawConsumerConfig, metaStreamInfo, logger).ConfigureAwait(false);
                 _logger?.LogDebug("Raw Consumer: {Reference} created for Stream: {Stream}",