Increase the timeout for confirmation to 10s

3s was too aggressive.

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

Author: Gsantomaggio

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -109,6 +109,11 @@ RabbitMQ.Stream.Client.Reliable.Producer.Info.get -> RabbitMQ.Stream.Client.Prod
 RabbitMQ.Stream.Client.Reliable.ProducerConfig.Filter.get -> RabbitMQ.Stream.Client.ProducerFilter
 RabbitMQ.Stream.Client.Reliable.ProducerConfig.Filter.set -> void
 RabbitMQ.Stream.Client.Reliable.ProducerConfig.Reference.set -> void
+RabbitMQ.Stream.Client.Reliable.ReconnectingArgs
+RabbitMQ.Stream.Client.Reliable.ReconnectingArgs.IsReconnecting.get -> bool
+RabbitMQ.Stream.Client.Reliable.ReconnectingArgs.ReconnectingArgs(bool isReconnecting) -> void
+RabbitMQ.Stream.Client.Reliable.ReliableConfig.Reconnecting -> RabbitMQ.Stream.Client.Reliable.ReliableConfig.ReconnectingEventHandler
+RabbitMQ.Stream.Client.Reliable.ReliableConfig.ReconnectingEventHandler
 RabbitMQ.Stream.Client.Reliable.SuperStreamConfig.RoutingStrategyType.get -> RabbitMQ.Stream.Client.RoutingStrategyType
 RabbitMQ.Stream.Client.Reliable.SuperStreamConfig.RoutingStrategyType.set -> void
 RabbitMQ.Stream.Client.RouteNotFoundException
@@ -146,3 +151,4 @@ RabbitMQ.Stream.Client.UnsupportedOperationException.UnsupportedOperationExcepti
 static RabbitMQ.Stream.Client.Connection.Create(System.Net.EndPoint endpoint, System.Func<System.Memory<byte>, System.Threading.Tasks.Task> commandCallback, System.Func<string, System.Threading.Tasks.Task> closedCallBack, RabbitMQ.Stream.Client.SslOption sslOption, Microsoft.Extensions.Logging.ILogger logger) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Connection>
 static RabbitMQ.Stream.Client.Message.From(ref System.Buffers.ReadOnlySequence<byte> seq, uint len) -> RabbitMQ.Stream.Client.Message
 static RabbitMQ.Stream.Client.Reliable.DeduplicatingProducer.Create(RabbitMQ.Stream.Client.Reliable.DeduplicatingProducerConfig producerConfig, Microsoft.Extensions.Logging.ILogger<RabbitMQ.Stream.Client.Reliable.Producer> logger = null) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Reliable.DeduplicatingProducer>
+virtual RabbitMQ.Stream.Client.Reliable.ReliableBase.OnReconnecting(RabbitMQ.Stream.Client.Reliable.ReconnectingArgs e) -> void

RabbitMQ.Stream.Client/Reliable/Producer.cs

@@ -22,7 +22,7 @@ public record SuperStreamConfig
 
 public record ProducerConfig : ReliableConfig
 {
-    private readonly TimeSpan _timeoutMessageAfter = TimeSpan.FromSeconds(3);
+    private readonly TimeSpan _timeoutMessageAfter = TimeSpan.FromSeconds(10);
 
     /// <summary>
     /// Reference used for deduplication.

RabbitMQ.Stream.Client/Reliable/ReliableBase.cs

@@ -10,12 +10,32 @@
 
 namespace RabbitMQ.Stream.Client.Reliable;
 
+public class ReconnectingArgs : EventArgs
+{
+    public ReconnectingArgs(bool isReconnecting)
+    {
+        IsReconnecting = isReconnecting;
+    }
+
+    public bool IsReconnecting { get; }
+}
+
 public record ReliableConfig
 {
     public IReconnectStrategy ReconnectStrategy { get; set; }
     public StreamSystem StreamSystem { get; }
     public string Stream { get; }
 
+
+    public delegate void ReconnectingEventHandler(object sender, ReconnectingArgs e);
+
+    public event ReconnectingEventHandler Reconnecting;
+
+    internal virtual void OnReconnecting(ReconnectingArgs e)
+    {
+        Reconnecting?.Invoke(this, e);
+    }
+
     protected ReliableConfig(StreamSystem streamSystem, string stream)
     {
         if (string.IsNullOrWhiteSpace(stream))
@@ -41,6 +61,15 @@ public abstract class ReliableBase
     protected bool _isOpen;
     protected bool _inReconnection;
 
+    internal delegate void ReconnectingEventHandler(object sender, ReconnectingArgs e);
+
+    internal event ReconnectingEventHandler Reconnecting;
+
+    protected virtual void OnReconnecting(ReconnectingArgs e)
+    {
+        Reconnecting?.Invoke(this, e);
+    }
+
     protected abstract ILogger BaseLogger { get; }
 
     internal async Task Init(IReconnectStrategy reconnectStrategy)
@@ -105,6 +134,7 @@ private async Task Init(bool boot, IReconnectStrategy reconnectStrategy)
     protected async Task TryToReconnect(IReconnectStrategy reconnectStrategy)
     {
         _inReconnection = true;
+        OnReconnecting(new ReconnectingArgs(true));
         try
         {
             switch (await reconnectStrategy.WhenDisconnected(ToString()).ConfigureAwait(false) && _isOpen)
@@ -122,6 +152,7 @@ protected async Task TryToReconnect(IReconnectStrategy reconnectStrategy)
         finally
         {
             _inReconnection = false;
+            OnReconnecting(new ReconnectingArgs(false));
         }
     }
 
@@ -143,7 +174,8 @@ internal async Task HandleMetaDataMaybeReconnect(string stream, StreamSystem sys
         await Task.Delay(500).ConfigureAwait(false);
         if (await system.StreamExists(stream).ConfigureAwait(false))
         {
-            BaseLogger.LogInformation("Meta data update stream: {StreamIdentifier}. The stream still exists. Client: {Identity}",
+            BaseLogger.LogInformation(
+                "Meta data update stream: {StreamIdentifier}. The stream still exists. Client: {Identity}",
                 stream,
                 ToString()
             );

docs/Documentation/Documentation.csproj

@@ -6,7 +6,8 @@
     <MinVerSkip>true</MinVerSkip>
     <Nullable>enable</Nullable>
     <OutputType>Exe</OutputType>
-    <TargetFramework>net6.0</TargetFramework>
+    <TargetFramework>net7.0</TargetFramework>
+    <TargetFrameworks />
   </PropertyGroup>
 
   <ItemGroup>