* Move closed TCS into AmqpManagement and AmqpConnection.

Author: lukebakken

RabbitMQ.AMQP.Client/Impl/AbstractLifeCycle.cs

@@ -12,11 +12,6 @@ public abstract class AbstractLifeCycle : ILifeCycle
 {
     protected bool _disposed;
 
-    // TODO this should not be part of AbstractLifeCycle
-    // wait until the close operation is completed
-    protected readonly TaskCompletionSource<bool> _connectionCloseTaskCompletionSource =
-        new(TaskCreationOptions.RunContinuationsAsynchronously);
-
     public virtual Task OpenAsync()
     {
         OnNewStatus(State.Open, null);

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -42,6 +42,9 @@ public class AmqpConnection : AbstractLifeCycle, IConnection
     internal ConcurrentDictionary<Guid, IPublisher> Publishers { get; } = new();
     internal ConcurrentDictionary<Guid, IConsumer> Consumers { get; } = new();
 
+    private readonly TaskCompletionSource _connectionClosedTcs =
+        new(TaskCreationOptions.RunContinuationsAsynchronously);
+
     public ReadOnlyCollection<IPublisher> GetPublishers()
     {
         return Publishers.Values.ToList().AsReadOnly();
@@ -131,7 +134,8 @@ await _nativeConnection.CloseAsync()
             _semaphoreClose.Release();
         }
 
-        await _connectionCloseTaskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(10))
+        // TODO configurable timeout?
+        await _connectionClosedTcs.Task.WaitAsync(TimeSpan.FromSeconds(10))
             .ConfigureAwait(false);
 
         OnNewStatus(State.Closed, null);
@@ -344,7 +348,7 @@ void DoClose(Error? argError = null)
                     Trace.WriteLine(TraceLevel.Verbose, $"{ToString()} is closed");
                     OnNewStatus(State.Closed, err);
                     ChangeEntitiesStatus(State.Closed, err);
-                    _connectionCloseTaskCompletionSource.SetResult(true);
+                    _connectionClosedTcs.SetResult();
                 }
 
                 if (error is null)
@@ -449,10 +453,10 @@ await ReconnectEntitiesAsync()
             }
             catch
             {
-                // TODO set states to Closed? Error?
+                // TODO set states to Closed? Error? Log exception? Set exception on TCS?
                 // This will be skipped if reconnection succeeds, but if there
                 // is an exception, it's important that this be called.
-                _connectionCloseTaskCompletionSource.SetResult(true);
+                _connectionClosedTcs.SetResult();
                 throw;
             }
             finally

RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs

@@ -43,6 +43,9 @@ public class AmqpManagement : AbstractLifeCycle, IManagement, IManagementTopolog
     internal const string Delete = "DELETE";
     private const string ReplyTo = "$me";
 
+    protected readonly TaskCompletionSource _managementSessionClosedTcs =
+        new(TaskCreationOptions.RunContinuationsAsynchronously);
+
     internal AmqpManagement(AmqpManagementParameters amqpManagementParameters)
     {
         _amqpManagementParameters = amqpManagementParameters;
@@ -165,7 +168,7 @@ private void OnManagementSessionClosed(IAmqpObject sender, Amqp.Framing.Error er
         OnNewStatus(State.Closed, Utils.ConvertError(error));
 
         // Note: TrySetResult *must* be used here
-        _connectionCloseTaskCompletionSource.TrySetResult(true);
+        _managementSessionClosedTcs.TrySetResult();
     }
 
     private async Task ProcessResponses()
@@ -472,7 +475,7 @@ public override async Task CloseAsync()
             await _managementSession.CloseAsync(closeSpan)
                 .ConfigureAwait(false);
 
-            await _connectionCloseTaskCompletionSource.Task.WaitAsync(closeSpan)
+            await _managementSessionClosedTcs.Task.WaitAsync(closeSpan)
                 .ConfigureAwait(false);
 
             _managementSession = null;

RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt

@@ -555,7 +555,6 @@ RabbitMQ.AMQP.Client.StreamOffsetSpecification
 RabbitMQ.AMQP.Client.StreamOffsetSpecification.First = 0 -> RabbitMQ.AMQP.Client.StreamOffsetSpecification
 RabbitMQ.AMQP.Client.StreamOffsetSpecification.Last = 1 -> RabbitMQ.AMQP.Client.StreamOffsetSpecification
 RabbitMQ.AMQP.Client.StreamOffsetSpecification.Next = 2 -> RabbitMQ.AMQP.Client.StreamOffsetSpecification
-readonly RabbitMQ.AMQP.Client.Impl.AbstractLifeCycle._connectionCloseTaskCompletionSource -> System.Threading.Tasks.TaskCompletionSource<bool>!
 static RabbitMQ.AMQP.Client.ByteCapacity.B(long bytes) -> RabbitMQ.AMQP.Client.ByteCapacity!
 static RabbitMQ.AMQP.Client.ByteCapacity.From(string! value) -> RabbitMQ.AMQP.Client.ByteCapacity!
 static RabbitMQ.AMQP.Client.ByteCapacity.Gb(long gigabytes) -> RabbitMQ.AMQP.Client.ByteCapacity!