* Rename _disposedValue to _disposed

* Check for `_disposed` before releasing semaphore

Author: lukebakken

RabbitMQ.AMQP.Client/Impl/AbstractLifeCycle.cs

@@ -6,7 +6,12 @@ public class AmqpNotOpenException(string message) : Exception(message);
 
 public abstract class AbstractLifeCycle : ILifeCycle
 {
-    private bool _disposedValue;
+    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()
     {
@@ -44,10 +49,6 @@ protected void ThrowIfClosed()
         }
     }
 
-    // wait until the close operation is completed
-    protected readonly TaskCompletionSource<bool> ConnectionCloseTaskCompletionSource =
-        new(TaskCreationOptions.RunContinuationsAsynchronously);
-
     protected void OnNewStatus(State newState, Error? error)
     {
         if (State == newState)
@@ -62,7 +63,7 @@ protected void OnNewStatus(State newState, Error? error)
 
     protected virtual void Dispose(bool disposing)
     {
-        if (!_disposedValue)
+        if (false == _disposed)
         {
             if (disposing)
             {
@@ -71,7 +72,7 @@ protected virtual void Dispose(bool disposing)
 
             // TODO: free unmanaged resources (unmanaged objects) and override finalizer
             // TODO: set large fields to null
-            _disposedValue = true;
+            _disposed = true;
         }
     }
 

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -128,7 +128,7 @@ await _nativeConnection.CloseAsync()
             _semaphoreClose.Release();
         }
 
-        await ConnectionCloseTaskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(10))
+        await _connectionCloseTaskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(10))
             .ConfigureAwait(false);
 
         OnNewStatus(State.Closed, null);
@@ -305,6 +305,11 @@ private ClosedCallback BuildClosedCallback()
     {
         return async (sender, error) =>
         {
+            if (_disposed)
+            {
+                return;
+            }
+
             await _semaphoreClose.WaitAsync()
                 .ConfigureAwait(false);
             try
@@ -318,7 +323,7 @@ void DoClose(Error? argError = null)
                     Trace.WriteLine(TraceLevel.Verbose, $"{ToString()} is closed");
                     OnNewStatus(State.Closed, err);
                     ChangeEntitiesStatus(State.Closed, err);
-                    ConnectionCloseTaskCompletionSource.SetResult(true);
+                    _connectionCloseTaskCompletionSource.SetResult(true);
                 }
 
                 if (error is null)
@@ -424,12 +429,16 @@ await _recordingTopologyListener.Accept(visitor)
                 // TODO set states to Closed? Error?
                 // This will be skipped if reconnection succeeds, but if there
                 // is an exception, it's important that this be called.
-                ConnectionCloseTaskCompletionSource.SetResult(true);
+                _connectionCloseTaskCompletionSource.SetResult(true);
                 throw;
             }
             finally
             {
-                _semaphoreClose.Release();
+                // TODO it is odd to have to add this check
+                if (false == _disposed)
+                {
+                    _semaphoreClose.Release();
+                }
             }
         };
     }

RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs

@@ -129,7 +129,7 @@ await EnsureReceiverLinkAsync()
 
             OnNewStatus(State.Closed, Utils.ConvertError(error));
             // Note: TrySetResult *must* be used here
-            ConnectionCloseTaskCompletionSource.TrySetResult(true);
+            _connectionCloseTaskCompletionSource.TrySetResult(true);
         };
 
         await base.OpenAsync()
@@ -442,7 +442,7 @@ public override async Task CloseAsync()
             await _managementSession.CloseAsync(closeSpan)
                 .ConfigureAwait(false);
 
-            await ConnectionCloseTaskCompletionSource.Task.WaitAsync(closeSpan)
+            await _connectionCloseTaskCompletionSource.Task.WaitAsync(closeSpan)
                 .ConfigureAwait(false);
 
             _managementSession = null;

RabbitMQ.AMQP.Client/Impl/AmqpPublisher.cs

@@ -1,5 +1,4 @@
-using System.Diagnostics;
-using Amqp;
+using Amqp;
 using Amqp.Framing;
 using Trace = Amqp.Trace;
 using TraceLevel = Amqp.TraceLevel;
@@ -98,7 +97,8 @@ public async Task<PublishResult> PublishAsync(IMessage message, CancellationToke
 
         try
         {
-            Debug.Assert(false == _senderLink.IsClosed);
+            // TODO anything special to do if the _senderLink is closed?
+            // Debug.Assert(false == _senderLink.IsClosed);
             Message nativeMessage = ((AmqpMessage)message).NativeMessage;
             await _senderLink.SendAsync(nativeMessage)
                 .ConfigureAwait(false);

RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt

@@ -171,6 +171,7 @@ RabbitMQ.AMQP.Client.Impl.AbstractLifeCycle.Dispose() -> void
 RabbitMQ.AMQP.Client.Impl.AbstractLifeCycle.OnNewStatus(RabbitMQ.AMQP.Client.State newState, RabbitMQ.AMQP.Client.Error? error) -> void
 RabbitMQ.AMQP.Client.Impl.AbstractLifeCycle.State.get -> RabbitMQ.AMQP.Client.State
 RabbitMQ.AMQP.Client.Impl.AbstractLifeCycle.ThrowIfClosed() -> void
+RabbitMQ.AMQP.Client.Impl.AbstractLifeCycle._disposed -> bool
 RabbitMQ.AMQP.Client.Impl.AbstractReconnectLifeCycle
 RabbitMQ.AMQP.Client.Impl.AbstractReconnectLifeCycle.AbstractReconnectLifeCycle() -> void
 RabbitMQ.AMQP.Client.Impl.AddressBuilder
@@ -567,7 +568,7 @@ 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>!
+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!