* Renaming methods that return Task or Task<T> to end with Async suffic

* Re-ordering methods in `AmqpConnection` so that they are in public, internal, private order
* Enable semaphores in `AmqpConnection` and add `IDisposable`

Author: lukebakken

RabbitMQ.AMQP.Client/Impl/AbstractLifeCycle.cs

@@ -61,7 +61,7 @@ internal void ChangeStatus(State newState, Error? error)
         OnNewStatus(newState, error);
     }
 
-    internal async Task Reconnect()
+    internal async Task ReconnectAsync()
     {
         try
         {
@@ -84,7 +84,7 @@ internal async Task Reconnect()
             await Task.Delay(delay).ConfigureAwait(false);
             if (_backOffDelayPolicy.IsActive())
             {
-                await Reconnect().ConfigureAwait(false);
+                await ReconnectAsync().ConfigureAwait(false);
             }
         }
     }

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -11,81 +11,30 @@ namespace RabbitMQ.AMQP.Client.Impl;
 /// AmqpConnection is the concrete implementation of <see cref="IConnection"/>
 /// It is a wrapper around the AMQP.Net Lite <see cref="Connection"/> class
 /// </summary>
-public class AmqpConnection : AbstractLifeCycle, IConnection
+public class AmqpConnection : AbstractLifeCycle, IConnection, IDisposable
 {
     private const string ConnectionNotRecoveredCode = "CONNECTION_NOT_RECOVERED";
     private const string ConnectionNotRecoveredMessage = "Connection not recovered";
-    private readonly SemaphoreSlim _semaphoreClose = new(1, 1); // TODO this needs to be Disposed
+
+    private readonly SemaphoreSlim _semaphoreClose = new(1, 1);
+    private readonly SemaphoreSlim _semaphoreOpen = new(1, 1);
 
     // The native AMQP.Net Lite connection
     private Connection? _nativeConnection;
 
     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 (IPublisher publisher1 in Publishers.Values)
-        {
-            var publisher = (AmqpPublisher)publisher1;
-            publisher.ChangeStatus(state, error);
-        }
-    }
-
-    private void ChangeConsumersStatus(State state, Error? error)
-    {
-        foreach (IConsumer 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 (IPublisher publisher1 in Publishers.Values)
-        {
-            var publisher = (AmqpPublisher)publisher1;
-            await publisher.Reconnect().ConfigureAwait(false);
-        }
-    }
-
-    private async Task ReconnectConsumers()
-    {
-        foreach (IConsumer consumer1 in Consumers.Values)
-        {
-            var consumer = (AmqpConsumer)consumer1;
-            await consumer.Reconnect().ConfigureAwait(false);
-        }
-    }
-
     private readonly IConnectionSettings _connectionSettings;
     internal readonly AmqpSessionManagement _nativePubSubSessions;
 
-    // TODO: Implement the semaphore to avoid multiple connections
-    // private readonly SemaphoreSlim _semaphore = new(1, 1);
-
     /// <summary>
     /// Publishers contains all the publishers created by the connection.
     /// Each connection can have multiple publishers.
     /// They key is the publisher Id ( a Guid)  
     /// See <see cref="AmqpPublisher"/>
     /// </summary>
     internal ConcurrentDictionary<string, IPublisher> Publishers { get; } = new();
-
     internal ConcurrentDictionary<string, IConsumer> Consumers { get; } = new();
 
     public ReadOnlyCollection<IPublisher> GetPublishers()
@@ -116,6 +65,87 @@ await connection.OpenAsync()
         return connection;
     }
 
+    public IManagement Management()
+    {
+        return _management;
+    }
+
+    public IConsumerBuilder ConsumerBuilder()
+    {
+        return new AmqpConsumerBuilder(this);
+    }
+
+    public override async Task OpenAsync()
+    {
+        await OpenConnectionAsync()
+            .ConfigureAwait(false);
+        await base.OpenAsync()
+            .ConfigureAwait(false);
+    }
+
+    public IPublisherBuilder PublisherBuilder()
+    {
+        ThrowIfClosed();
+        var publisherBuilder = new AmqpPublisherBuilder(this);
+        return publisherBuilder;
+    }
+
+    public override async Task CloseAsync()
+    {
+        await _semaphoreClose.WaitAsync()
+            .ConfigureAwait(false);
+        try
+        {
+            await CloseAllPublishers().ConfigureAwait(false);
+            await CloseAllConsumers().ConfigureAwait(false);
+
+            _recordingTopologyListener.Clear();
+            _nativePubSubSessions.ClearSessions();
+
+            if (State == State.Closed)
+            {
+                return;
+            }
+
+            OnNewStatus(State.Closing, null);
+
+            await _management.CloseAsync()
+                .ConfigureAwait(false);
+
+            if (_nativeConnection is { IsClosed: false })
+            {
+                await _nativeConnection.CloseAsync()
+                    .ConfigureAwait(false);
+            }
+        }
+        finally
+        {
+            _semaphoreClose.Release();
+        }
+
+        await ConnectionCloseTaskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(10))
+            .ConfigureAwait(false);
+
+        OnNewStatus(State.Closed, null);
+    }
+
+    public void Dispose()
+    {
+        // TODO probably more should happen in this method
+        _semaphoreOpen.Dispose();
+        _semaphoreClose.Dispose();
+    }
+
+    public override string ToString()
+    {
+        string info = $"AmqpConnection{{ConnectionSettings='{_connectionSettings}', Status='{State.ToString()}'}}";
+        return info;
+    }
+
+    internal Connection? NativeConnection()
+    {
+        return _nativeConnection;
+    }
 
     /// <summary>
     /// Closes all the publishers. It is called when the connection is closed.
@@ -150,28 +180,11 @@ private AmqpConnection(IConnectionSettings connectionSettings)
             new AmqpManagement(new AmqpManagementParameters(this).TopologyListener(_recordingTopologyListener));
     }
 
-    public IManagement Management()
-    {
-        return _management;
-    }
-
-    public IConsumerBuilder ConsumerBuilder()
+    // TODO cancellation token
+    private async Task OpenConnectionAsync()
     {
-        return new AmqpConsumerBuilder(this);
-    }
-
-    public override async Task OpenAsync()
-    {
-        await EnsureConnection()
+        await _semaphoreOpen.WaitAsync()
             .ConfigureAwait(false);
-        await base.OpenAsync()
-            .ConfigureAwait(false);
-    }
-
-    private async Task EnsureConnection()
-    {
-        // TODO: do this!
-        // await _semaphore.WaitAsync();
         try
         {
             if (_nativeConnection is { IsClosed: false })
@@ -252,27 +265,34 @@ await _management.OpenAsync()
 
             _nativeConnection.Closed += MaybeRecoverConnection();
         }
-
         catch (AmqpException e)
         {
             Trace.WriteLine(TraceLevel.Error, $"Error trying to connect. Info: {ToString()}, error: {e}");
             throw new ConnectionException($"Error trying to connect. Info: {ToString()}, error: {e}");
         }
+        finally
+        {
+            _semaphoreOpen.Release();
+        }
     }
 
     /// <summary>
     /// Event handler for the connection closed event.
     /// In case the error is null means that the connection is closed by the user.
     /// The recover mechanism is activated only if the error is not null.
     /// The connection maybe recovered if the recovery configuration is active.
+    ///
+    /// TODO this method could be improved.
+    /// MaybeRecoverConnection should set a connection state to RECOVERING
+    /// and then kick off a task dedicated to recovery
     /// </summary>
     /// <returns></returns>
     private ClosedCallback MaybeRecoverConnection()
     {
         return async (sender, error) =>
         {
-            await _semaphoreClose.WaitAsync().ConfigureAwait(false);
-
+            await _semaphoreClose.WaitAsync()
+                .ConfigureAwait(false);
             try
             {
                 // close all the sessions, if the connection is closed the sessions are not valid anymore
@@ -299,7 +319,6 @@ private ClosedCallback MaybeRecoverConnection()
                     OnNewStatus(State.Reconnecting, Utils.ConvertError(error));
                     ChangeEntitiesStatus(State.Reconnecting, Utils.ConvertError(error));
 
-
                     await Task.Run(async () =>
                     {
                         bool connected = false;
@@ -328,7 +347,7 @@ await Task.Run(async () =>
                                 await Task.Delay(TimeSpan.FromMilliseconds(next))
                                     .ConfigureAwait(false);
 
-                                await EnsureConnection()
+                                await OpenConnectionAsync()
                                     .ConfigureAwait(false);
                                 connected = true;
                             }
@@ -370,7 +389,7 @@ await _recordingTopologyListener.Accept(visitor)
 
                         try
                         {
-                            await ReconnectEntities().ConfigureAwait(false);
+                            await ReconnectEntitiesAsync().ConfigureAwait(false);
                         }
                         catch (Exception e)
                         {
@@ -394,70 +413,68 @@ await _recordingTopologyListener.Accept(visitor)
         };
     }
 
-    internal Connection? NativeConnection()
+    private void ChangeEntitiesStatus(State state, Error? error)
     {
-        return _nativeConnection;
+        ChangePublishersStatus(state, error);
+        ChangeConsumersStatus(state, error);
+        _management.ChangeStatus(state, error);
     }
 
-    public IPublisherBuilder PublisherBuilder()
+    private void ChangePublishersStatus(State state, Error? error)
     {
-        ThrowIfClosed();
-        var publisherBuilder = new AmqpPublisherBuilder(this);
-        return publisherBuilder;
+        foreach (IPublisher publisher1 in Publishers.Values)
+        {
+            var publisher = (AmqpPublisher)publisher1;
+            publisher.ChangeStatus(state, error);
+        }
     }
 
-    public override async Task CloseAsync()
+    private void ChangeConsumersStatus(State state, Error? error)
     {
-        await _semaphoreClose.WaitAsync()
-            .ConfigureAwait(false);
-        try
-        {
-            await CloseAllPublishers().ConfigureAwait(false);
-            await CloseAllConsumers().ConfigureAwait(false);
-
-            _recordingTopologyListener.Clear();
-            _nativePubSubSessions.ClearSessions();
-
-            if (State == State.Closed)
-            {
-                return;
-            }
-
-            OnNewStatus(State.Closing, null);
-
-            await _management.CloseAsync()
-                .ConfigureAwait(false);
-
-            if (_nativeConnection is { IsClosed: false })
-            {
-                await _nativeConnection.CloseAsync()
-                    .ConfigureAwait(false);
-            }
-        }
-        finally
+        foreach (IConsumer consumer1 in Consumers.Values)
         {
-            _semaphoreClose.Release();
+            var consumer = (AmqpConsumer)consumer1;
+            consumer.ChangeStatus(state, error);
         }
+    }
 
-        await ConnectionCloseTaskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(10))
+    private async Task ReconnectEntitiesAsync()
+    {
+        await ReconnectPublishersAsync()
             .ConfigureAwait(false);
+        await ReconnectConsumersAsync()
+            .ConfigureAwait(false);
+    }
 
-        OnNewStatus(State.Closed, null);
+    private async Task ReconnectPublishersAsync()
+    {
+        // TODO this could be done in parallel
+        foreach (IPublisher publisher1 in Publishers.Values)
+        {
+            var publisher = (AmqpPublisher)publisher1;
+            await publisher.ReconnectAsync()
+                .ConfigureAwait(false);
+        }
     }
 
-    public override string ToString()
+    private async Task ReconnectConsumersAsync()
     {
-        string info = $"AmqpConnection{{ConnectionSettings='{_connectionSettings}', Status='{State.ToString()}'}}";
-        return info;
+        // TODO this could be done in parallel
+        foreach (IConsumer consumer1 in Consumers.Values)
+        {
+            var consumer = (AmqpConsumer)consumer1;
+            await consumer.ReconnectAsync().ConfigureAwait(false);
+        }
     }
 }
 
 internal class Visitor(AmqpManagement management) : IVisitor
 {
     private AmqpManagement Management { get; } = management;
 
-    public async Task VisitQueues(IEnumerable<QueueSpec> queueSpec)
+    public async Task VisitQueuesAsync(IEnumerable<QueueSpec> queueSpec)
     {
+        // TODO this could be done in parallel
         foreach (QueueSpec spec in queueSpec)
         {
             Trace.WriteLine(TraceLevel.Information, $"Recovering queue {spec.Name}");