Auto-reconnect publishers and consumers

RabbitMQ.AMQP.Client/ILifeCycle.cs

@@ -9,7 +9,6 @@ public enum State
     Closed,
 }
 
-
 public class Error(string? errorCode, string? description)
 {
     public string? Description { get; } = description;
@@ -21,15 +20,13 @@ public override string ToString()
     }
 }
 
-
-
 public delegate void LifeCycleCallBack(object sender, State previousState, State currentState, Error? failureCause);
 
 public interface ILifeCycle
 {
     Task CloseAsync();
-    
+
     public State State { get; }
-    
-    event LifeCycleCallBack ChangeState;
+
+    public event LifeCycleCallBack ChangeState;
 }

RabbitMQ.AMQP.Client/Impl/AbstractLifeCycle.cs

@@ -40,3 +40,5 @@ protected void OnNewStatus(State newState, Error? error)
 
     public event LifeCycleCallBack? ChangeState;
 }
+
+

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -46,6 +46,55 @@ public class AmqpConnection : AbstractLifeCycle, IConnection
     private readonly AmqpManagement _management;
     private readonly RecordingTopologyListener _recordingTopologyListener = new();
 
+    private void ChangeEntitiesStatus(State state, Error? error)
+    {
+        ChangePublishersStatus(state, error);
+        ChangeConsumersStatus(state, error);
+        _management.ChangeStatus(state, error);
+    }
+
+    private void ChangePublishersStatus(State state, Error? error)
+    {
+        foreach (var publisher1 in Publishers.Values)
+        {
+            var publisher = (AmqpPublisher)publisher1;
+            publisher.ChangeStatus(state, error);
+        }
+    }
+
+    private void ChangeConsumersStatus(State state, Error? error)
+    {
+        foreach (var consumer1 in Consumers.Values)
+        {
+            var consumer = (AmqpConsumer)consumer1;
+            consumer.ChangeStatus(state, error);
+        }
+    }
+
+
+    private async Task ReconnectEntities()
+    {
+        await ReconnectPublishers().ConfigureAwait(false);
+        await ReconnectConsumers().ConfigureAwait(false);
+    }
+
+    private async Task ReconnectPublishers()
+    {
+        foreach (var publisher1 in Publishers.Values)
+        {
+            var publisher = (AmqpPublisher)publisher1;
+            await publisher.Reconnect().ConfigureAwait(false);
+        }
+    }
+
+    private async Task ReconnectConsumers()
+    {
+        foreach (var consumer1 in Consumers.Values)
+        {
+            var consumer = (AmqpConsumer)consumer1;
+            await consumer.Reconnect().ConfigureAwait(false);
+        }
+    }
 
     private readonly ConnectionSettings _connectionSettings;
     internal readonly AmqpSessionManagement _nativePubSubSessions;
@@ -138,7 +187,6 @@ protected override Task OpenAsync()
 
     private void EnsureConnection()
     {
-        // await _semaphore.WaitAsync();
         try
         {
             if (_nativeConnection is { IsClosed: false })
@@ -180,14 +228,6 @@ [new Symbol("connection_name")] = _connectionSettings.ConnectionName(),
             Trace.WriteLine(TraceLevel.Error, $"Error trying to connect. Info: {ToString()}, error: {e}");
             throw new ConnectionException($"Error trying to connect. Info: {ToString()}, error: {e}");
         }
-
-
-        finally
-        {
-            // _semaphore.Release();
-        }
-
-        // return Task.CompletedTask;
     }
 
     /// <summary>
@@ -205,8 +245,12 @@ private ClosedCallback MaybeRecoverConnection()
 
             try
             {
+                // close all the sessions, if the connection is closed the sessions are not valid anymore
+                _nativePubSubSessions.ClearSessions();
+
                 if (error != null)
                 {
+                    //  we assume here that the connection is closed unexpectedly, since the error is not null
                     Trace.WriteLine(TraceLevel.Warning, $"connection is closed unexpectedly. " +
                                                         $"Info: {ToString()}");
 
@@ -216,11 +260,16 @@ private ClosedCallback MaybeRecoverConnection()
                     if (!_connectionSettings.RecoveryConfiguration.IsActivate())
                     {
                         OnNewStatus(State.Closed, Utils.ConvertError(error));
+                        ChangeEntitiesStatus(State.Closed, Utils.ConvertError(error));
                         return;
                     }
 
-                    // TODO: Block the publishers and consumers
+                    // change the status for the connection and all the entities
+                    // to reconnecting and all the events are fired
                     OnNewStatus(State.Reconnecting, Utils.ConvertError(error));
+                    ChangeEntitiesStatus(State.Reconnecting, Utils.ConvertError(error));
+
+
                     await Task.Run(async () =>
                     {
                         bool connected = false;
@@ -266,6 +315,10 @@ await Task.Delay(TimeSpan.FromMilliseconds(next))
                             OnNewStatus(State.Closed,
                                 new Error(ConnectionNotRecoveredCode,
                                     $"{ConnectionNotRecoveredMessage}, recover status: {_connectionSettings.RecoveryConfiguration}"));
+
+                            ChangeEntitiesStatus(State.Closed, new Error(ConnectionNotRecoveredCode,
+                                $"{ConnectionNotRecoveredMessage}, recover status: {_connectionSettings.RecoveryConfiguration}"));
+
                             return;
                         }
 
@@ -278,9 +331,10 @@ await _recordingTopologyListener.Accept(visitor)
                         }
 
                         OnNewStatus(State.Open, null);
-                    }).ConfigureAwait(false);
-
+                        // after the connection is recovered we have to reconnect all the publishers and consumers
 
+                        await ReconnectEntities().ConfigureAwait(false);
+                    }).ConfigureAwait(false);
                     return;
                 }
 
@@ -320,6 +374,7 @@ await _semaphoreClose.WaitAsync()
             await CloseAllConsumers().ConfigureAwait(false);
 
             _recordingTopologyListener.Clear();
+            _nativePubSubSessions.ClearSessions();
 
             if (State == State.Closed)
             {

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -54,6 +54,8 @@ protected sealed override Task OpenAsync()
 
     private void ProcessMessages()
     {
+        // TODO: Check the performance during the download messages
+        // The publisher is faster than the consumer
         _receiverLink?.Start(_initialCredits,
             (link, message) =>
             {
@@ -88,6 +90,30 @@ public override async Task CloseAsync()
             return;
         }
 
+        OnNewStatus(State.Closing, null);
         await (_receiverLink.CloseAsync()).ConfigureAwait(false);
+        _receiverLink = null;
+        OnNewStatus(State.Closed, null);
+        _connection.Consumers.TryRemove(Id, out _);
+    }
+
+
+    internal void ChangeStatus(State newState, Error? error)
+    {
+        OnNewStatus(newState, error);
+    }
+
+    internal async Task Reconnect()
+    {
+        int randomWait = Random.Shared.Next(200, 800);
+        Trace.WriteLine(TraceLevel.Information, $"Consumer: {ToString()} is reconnecting in {randomWait} ms");
+        await Task.Delay(randomWait).ConfigureAwait(false);
+
+        if (_receiverLink != null)
+        {
+            await _receiverLink.DetachAsync().ConfigureAwait(false)!;
+        }
+
+        await OpenAsync().ConfigureAwait(false);
     }
 }

RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs

@@ -388,6 +388,12 @@ public override string ToString()
 
         return info;
     }
+
+
+    internal void ChangeStatus(State newState, Error? error)
+    {
+        OnNewStatus(newState, error);
+    }
 }
 
 public class InvalidCodeException(string message) : Exception(message);

RabbitMQ.AMQP.Client/Impl/AmqpPublisher.cs

@@ -41,6 +41,7 @@ protected sealed override Task OpenAsync()
                 throw new PublisherException("Failed to create sender link. Link state is not attached, error: " +
                     _senderLink.Error?.ToString() ?? "Unknown error");
             }
+
             return base.OpenAsync();
         }
         catch (Exception e)
@@ -52,7 +53,7 @@ protected sealed override Task OpenAsync()
     private string Id { get; } = Guid.NewGuid().ToString();
 
 
-    public void PausePublishing()
+    private void PausePublishing()
     {
         _pausePublishing.Reset();
     }
@@ -150,11 +151,7 @@ public override async Task CloseAsync()
         {
             return;
         }
-
         OnNewStatus(State.Closing, null);
-
-        _connection.Publishers.TryRemove(Id, out _);
-
         try
         {
             if (_senderLink != null)
@@ -172,10 +169,26 @@ await _senderLink.CloseAsync()
         }
 
         OnNewStatus(State.Closed, null);
+        _connection.Publishers.TryRemove(Id, out _);
+    }
+
+
+    internal void ChangeStatus(State newState, Error? error)
+    {
+        OnNewStatus(newState, error);
     }
 
-    public void ResumePublishing()
+    internal async Task Reconnect()
     {
-        MaybeResumePublishing();
+        int randomWait = Random.Shared.Next(200, 800);
+        Trace.WriteLine(TraceLevel.Information, $"Publisher: {ToString()} is reconnecting in {randomWait} ms");
+        await Task.Delay(randomWait).ConfigureAwait(false);
+
+        if (_senderLink != null)
+        {
+            await _senderLink.DetachAsync().ConfigureAwait(false)!;
+        }
+
+        await OpenAsync().ConfigureAwait(false);
     }
 }

RabbitMQ.AMQP.Client/Impl/AmqpSessionManagement.cs

@@ -18,4 +18,9 @@ public Session GetOrCreateSession()
         Sessions.Add(session);
         return session;
     }
+
+    public void ClearSessions()
+    {
+        Sessions.Clear();
+    }
 }

RabbitMQ.AMQP.Client/Impl/DeliveryContext.cs

@@ -6,16 +6,31 @@ public class DeliveryContext(IReceiverLink link, Message message) : IContext
 {
     public void Accept()
     {
+        if (link.IsClosed)
+        {
+            throw new ConsumerException("Link is closed");
+        }
+
         link.Accept(message);
     }
 
     public void Discard()
     {
+        if (link.IsClosed)
+        {
+            throw new ConsumerException("Link is closed");
+        }
+
         link.Reject(message);
     }
 
     public void Requeue()
     {
+        if (!link.IsClosed)
+        {
+            throw new ConsumerException("Link is closed");
+        }
+
         link.Release(message);
     }
 }

Tests/ConnectionRecoverTests.cs → Tests/ConnectionRecoveryTests.cs

@@ -38,9 +38,9 @@ public void Reset()
     public bool IsActive() => true;
 }
 
-public class ConnectionRecoverTests(ITestOutputHelper testOutputHelper)
+public class ConnectionRecoveryTests(ITestOutputHelper testOutputHelper)
 {
-    public ITestOutputHelper TestOutputHelper { get; } = testOutputHelper;
+    private ITestOutputHelper TestOutputHelper { get; } = testOutputHelper;
 
     /// <summary>
     /// The normal close the status should be correct and error null
@@ -53,8 +53,8 @@ public class ConnectionRecoverTests(ITestOutputHelper testOutputHelper)
     [InlineData(false)]
     public async Task NormalCloseTheStatusShouldBeCorrectAndErrorNull(bool activeRecovery)
     {
-        var connectionName = Guid.NewGuid().ToString();
-        var connection = await AmqpConnection.CreateAsync(
+        string connectionName = Guid.NewGuid().ToString();
+        IConnection connection = await AmqpConnection.CreateAsync(
             ConnectionSettingBuilder.Create().ConnectionName(connectionName).RecoveryConfiguration(
                 RecoveryConfiguration.Create().Activated(activeRecovery).Topology(false)).Build());
 

Tests/ManagementTests.cs

@@ -97,7 +97,8 @@ public async Task RaiseInvalidCodeException()
             {
                 Properties = new Properties()
                 {
-                    CorrelationId = messageId, Subject = "506", // 506 is not a valid code
+                    CorrelationId = messageId,
+                    Subject = "506", // 506 is not a valid code
                 }
             });
         });