add the OnConnectionClosed

function on the Producer class

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

Author: Gsantomaggio

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -88,6 +88,8 @@ RabbitMQ.Stream.Client.Reliable.DeduplicatingProducerConfig
 RabbitMQ.Stream.Client.Reliable.DeduplicatingProducerConfig.DeduplicatingProducerConfig(RabbitMQ.Stream.Client.StreamSystem streamSystem, string stream, string reference) -> void
 RabbitMQ.Stream.Client.Reliable.ProducerConfig.Filter.get -> RabbitMQ.Stream.Client.ProducerFilter
 RabbitMQ.Stream.Client.Reliable.ProducerConfig.Filter.set -> void
+RabbitMQ.Stream.Client.Reliable.ProducerConfig.OnConnectionClosed.get -> System.Func<string, System.Threading.Tasks.Task>
+RabbitMQ.Stream.Client.Reliable.ProducerConfig.OnConnectionClosed.set -> void
 RabbitMQ.Stream.Client.Reliable.ProducerConfig.Reference.set -> void
 RabbitMQ.Stream.Client.Reliable.SuperStreamConfig.RoutingStrategyType.get -> RabbitMQ.Stream.Client.RoutingStrategyType
 RabbitMQ.Stream.Client.Reliable.SuperStreamConfig.RoutingStrategyType.set -> void

RabbitMQ.Stream.Client/Reliable/Producer.cs

@@ -104,6 +104,8 @@ public TimeSpan TimeoutMessageAfter
         }
     }
 
+    public Func<string, Task> OnConnectionClosed { get; set; } = null;
+
     public ProducerConfig(StreamSystem streamSystem, string stream) : base(streamSystem, stream)
     {
     }

RabbitMQ.Stream.Client/Reliable/ProducerFactory.cs

@@ -62,6 +62,11 @@ private async Task<IProducer> SuperStreamProducer()
 
     private async Task<IProducer> StandardProducer()
     {
+        var onConnectionClosed = _producerConfig.OnConnectionClosed ?? (async _ =>
+        {
+            await TryToReconnect(_producerConfig.ReconnectStrategy).ConfigureAwait(false);
+        });
+
         return await _producerConfig.StreamSystem.CreateRawProducer(new RawProducerConfig(_producerConfig.Stream)
         {
             ClientProvidedName = _producerConfig.ClientProvidedName,
@@ -80,10 +85,7 @@ private async Task<IProducer> StandardProducer()
                         _producerConfig.StreamSystem).WaitAsync(CancellationToken.None);
                 });
             },
-            ConnectionClosedHandler = async _ =>
-            {
-                await TryToReconnect(_producerConfig.ReconnectStrategy).ConfigureAwait(false);
-            },
+            ConnectionClosedHandler = onConnectionClosed,
             ConfirmHandler = confirmation =>
             {
                 var confirmationStatus = confirmation.Code switch