Make channel implement IAsyncDisposable

Author: danielmarbach

projects/RabbitMQ.Client/IChannel.cs

@@ -42,10 +42,10 @@ namespace RabbitMQ.Client
     /// functionality offered by versions 0-8, 0-8qpid, 0-9 and 0-9-1 of AMQP.
     /// </summary>
     /// <remarks>
-    /// Extends the <see cref="IDisposable"/> interface, so that the "using"
+    /// Extends the <see cref="IDisposable"/> interface and the <see cref="IAsyncDisposable"/> interface, so that the "using"
     /// statement can be used to scope the lifetime of a channel when appropriate.
     /// </remarks>
-    public interface IChannel : IDisposable
+    public interface IChannel : IAsyncDisposable, IDisposable
     {
         /// <summary>
         /// Channel number, unique per connections.

projects/RabbitMQ.Client/Impl/AutorecoveringChannel.cs

@@ -273,6 +273,23 @@ public void Dispose()
             _disposed = true;
         }
 
+        public async ValueTask DisposeAsync()
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            if (IsOpen)
+            {
+                await this.AbortAsync()
+                    .ConfigureAwait(false);
+            }
+
+            _recordedConsumerTags.Clear();
+            _disposed = true;
+        }
+
         public ValueTask<ulong> GetNextPublishSequenceNumberAsync(CancellationToken cancellationToken = default) => InnerChannel.GetNextPublishSequenceNumberAsync(cancellationToken);
 
         public ValueTask BasicAckAsync(ulong deliveryTag, bool multiple, CancellationToken cancellationToken)

projects/RabbitMQ.Client/Impl/ChannelBase.cs

@@ -563,11 +563,31 @@ protected virtual void Dispose(bool disposing)
                 }
 
                 ConsumerDispatcher.Dispose();
-                _rpcSemaphore.Dispose();
+                _rpcSemaphore?.Dispose();
                 _confirmSemaphore?.Dispose();
             }
         }
 
+        public async ValueTask DisposeAsync()
+        {
+            await DisposeAsyncCore()
+                .ConfigureAwait(false);
+
+            Dispose(false);
+        }
+
+        protected virtual async ValueTask DisposeAsyncCore()
+        {
+            if (IsOpen)
+            {
+                await this.AbortAsync().ConfigureAwait(false);
+            }
+
+            ConsumerDispatcher.Dispose();
+            _rpcSemaphore.Dispose();
+            _confirmSemaphore?.Dispose();
+        }
+
         public Task ConnectionTuneOkAsync(ushort channelMax, uint frameMax, ushort heartbeat, CancellationToken cancellationToken)
         {
             var method = new ConnectionTuneOk(channelMax, frameMax, heartbeat);