* Add TCS for server-originated channel closure.

Author: lukebakken

projects/Applications/GH-1749/Program.cs

@@ -34,6 +34,7 @@
 using System.Runtime.ExceptionServices;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
+using RabbitMQ.Client.Exceptions;
 
 namespace GH_1749
 {
@@ -74,7 +75,10 @@ static async Task Main(string[] args)
             ConnectionFactory connectionFactory = new()
             {
                 HostName = hostname,
-                AutomaticRecoveryEnabled = true,
+                AutomaticRecoveryEnabled = false,
+                TopologyRecoveryEnabled = false,
+                RequestedConnectionTimeout = TimeSpan.FromSeconds(600),
+                RequestedHeartbeat = TimeSpan.FromSeconds(600),
                 UserName = "guest",
                 Password = "guest",
                 ClientProvidedName = ConnectionClientProvidedName
@@ -86,7 +90,6 @@ static async Task Main(string[] args)
             connection.RecoverySucceededAsync += (object sender, AsyncEventArgs ea) =>
             {
                 Console.WriteLine("{0} [INFO] saw RecoverySucceededAsync, event: {1}", Now, ea);
-                _ = CloseConnectionAsync();
                 return Task.CompletedTask;
             };
 
@@ -113,33 +116,15 @@ static async Task Main(string[] args)
             channel.CallbackExceptionAsync += Channel_CallbackExceptionAsync;
             channel.ChannelShutdownAsync += Channel_ChannelShutdownAsync;
 
-            QueueDeclareOk queue = await channel.QueueDeclareAsync();
-
-            var consumer = new GH1749Consumer(channel);
-            await channel.BasicConsumeAsync(queue.QueueName, true, consumer);
-
-            _ = CloseConnectionAsync();
-
-            Console.WriteLine("{0} [INFO] consumer is running", Util.Now);
-            Console.ReadLine();
-        }
-
-        static async Task CloseConnectionAsync()
-        {
-            if (s_util is null)
-            {
-                throw new NullReferenceException("s_util");
-            }
-
             try
             {
-                Console.WriteLine("{0} [INFO] start closing connection: {1}", Now, ConnectionClientProvidedName);
-                await s_util.CloseConnectionAsync(ConnectionClientProvidedName);
-                Console.WriteLine("{0} [INFO] done closing connection: {1}", Now, ConnectionClientProvidedName);
+                await channel.QueueDeclarePassiveAsync(Guid.NewGuid().ToString());
             }
-            catch (Exception ex)
+            catch (OperationInterruptedException)
             {
-                Console.Error.WriteLine("{0} [ERROR] error while closing connection: {1}", Now, ex);
+                await channel.DisposeAsync();
+                // rabbitmq-dotnet-client-1749
+                // await Task.Delay(2000);
             }
         }
 
@@ -156,6 +141,8 @@ private static Task Channel_ChannelShutdownAsync(object sender, ShutdownEventArg
         {
             Console.WriteLine("{0} [INFO] saw ChannelShutdownAsync, event: {1}", Now, ea);
             return Task.CompletedTask;
+            // rabbitmq-dotnet-client-1749
+            // return Task.Delay(1000);
         }
 
         private static void CurrentDomain_FirstChanceException(object? sender, FirstChanceExceptionEventArgs e)

projects/RabbitMQ.Client/Impl/Channel.cs

@@ -57,13 +57,14 @@ internal partial class Channel : IChannel, IRecoverable
         private readonly RpcContinuationQueue _continuationQueue = new RpcContinuationQueue();
 
         private ShutdownEventArgs? _closeReason;
-        public ShutdownEventArgs? CloseReason => Volatile.Read(ref _closeReason);
+        private TaskCompletionSource<bool>? _serverOriginatedChannelCloseTcs;
 
         internal readonly IConsumerDispatcher ConsumerDispatcher;
 
         private bool _disposedValue;
         private bool _isDisposing;
-        private readonly object _isDisposingLock = new();
+
+        private readonly object _locker = new();
 
         public Channel(ISession session, CreateChannelOptions createChannelOptions)
         {
@@ -86,6 +87,8 @@ public Channel(ISession session, CreateChannelOptions createChannelOptions)
 
         internal TimeSpan HandshakeContinuationTimeout { get; set; } = TimeSpan.FromSeconds(10);
 
+        public ShutdownEventArgs? CloseReason => Volatile.Read(ref _closeReason);
+
         public TimeSpan ContinuationTimeout { get; set; }
 
         public event AsyncEventHandler<BasicAckEventArgs> BasicAcksAsync
@@ -548,7 +551,7 @@ protected virtual void Dispose(bool disposing)
 
             if (disposing)
             {
-                lock (_isDisposingLock)
+                lock (_locker)
                 {
                     if (_isDisposing)
                     {
@@ -564,13 +567,20 @@ protected virtual void Dispose(bool disposing)
                         this.AbortAsync().GetAwaiter().GetResult();
                     }
 
-                    ConsumerDispatcher.Dispose();
-                    _rpcSemaphore.Dispose();
-                    _confirmSemaphore.Dispose();
-                    _outstandingPublisherConfirmationsRateLimiter?.Dispose();
+                    MaybeWaitForServerOriginatedClose();
                 }
                 finally
                 {
+                    try
+                    {
+                        ConsumerDispatcher.Dispose();
+                        _rpcSemaphore.Dispose();
+                        _confirmSemaphore.Dispose();
+                        _outstandingPublisherConfirmationsRateLimiter?.Dispose();
+                    }
+                    catch
+                    {
+                    }
                     _disposedValue = true;
                     _isDisposing = false;
                 }
@@ -586,7 +596,7 @@ protected virtual async ValueTask DisposeAsyncCore(bool disposing)
 
             if (disposing)
             {
-                lock (_isDisposingLock)
+                lock (_locker)
                 {
                     if (_isDisposing)
                     {
@@ -603,18 +613,26 @@ await this.AbortAsync()
                             .ConfigureAwait(false);
                     }
 
-                    ConsumerDispatcher.Dispose();
-                    _rpcSemaphore.Dispose();
-                    _confirmSemaphore.Dispose();
-
                     if (_outstandingPublisherConfirmationsRateLimiter is not null)
                     {
                         await _outstandingPublisherConfirmationsRateLimiter.DisposeAsync()
                             .ConfigureAwait(false);
                     }
+
+                    await MaybeWaitForServerOriginatedCloseAsync()
+                        .ConfigureAwait(false);
                 }
                 finally
                 {
+                    try
+                    {
+                        ConsumerDispatcher.Dispose();
+                        _rpcSemaphore.Dispose();
+                        _confirmSemaphore.Dispose();
+                    }
+                    catch
+                    {
+                    }
                     _disposedValue = true;
                     _isDisposing = false;
                 }
@@ -714,25 +732,39 @@ protected virtual ulong AdjustDeliveryTag(ulong deliveryTag)
 
         protected async Task<bool> HandleChannelCloseAsync(IncomingCommand cmd, CancellationToken cancellationToken)
         {
-            // TODO add check for Disposing / Disposed
-            var channelClose = new ChannelClose(cmd.MethodSpan);
-            SetCloseReason(new ShutdownEventArgs(ShutdownInitiator.Peer,
-                channelClose._replyCode,
-                channelClose._replyText,
-                channelClose._classId,
-                channelClose._methodId));
-
-            await Session.CloseAsync(_closeReason, notify: false)
-                .ConfigureAwait(false);
+            lock (_locker)
+            {
+                _serverOriginatedChannelCloseTcs ??= new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            }
 
-            var method = new ChannelCloseOk();
-            await ModelSendAsync(in method, cancellationToken)
-                .ConfigureAwait(false);
+            try
+            {
+                // TODO add check for Disposing / Disposed
+                var channelClose = new ChannelClose(cmd.MethodSpan);
+                SetCloseReason(new ShutdownEventArgs(ShutdownInitiator.Peer,
+                    channelClose._replyCode,
+                    channelClose._replyText,
+                    channelClose._classId,
+                    channelClose._methodId));
 
-            await Session.NotifyAsync(cancellationToken)
-                .ConfigureAwait(false);
+                await Session.CloseAsync(_closeReason, notify: false)
+                    .ConfigureAwait(false);
 
-            return true;
+                var method = new ChannelCloseOk();
+                await ModelSendAsync(in method, cancellationToken)
+                    .ConfigureAwait(false);
+
+                await Session.NotifyAsync(cancellationToken)
+                    .ConfigureAwait(false);
+
+                _serverOriginatedChannelCloseTcs.TrySetResult(true);
+                return true;
+            }
+            catch (Exception ex)
+            {
+                _serverOriginatedChannelCloseTcs.TrySetException(ex);
+                throw;
+            }
         }
 
         protected async Task<bool> HandleChannelCloseOkAsync(IncomingCommand cmd, CancellationToken cancellationToken)
@@ -1654,5 +1686,34 @@ private Task<bool> DispatchCommandAsync(IncomingCommand cmd, CancellationToken c
                     }
             }
         }
+
+        private void MaybeWaitForServerOriginatedClose()
+        {
+            if (_serverOriginatedChannelCloseTcs is not null)
+            {
+                try
+                {
+                    _serverOriginatedChannelCloseTcs.Task.Wait(TimeSpan.FromSeconds(5));
+                }
+                catch
+                {
+                }
+            }
+        }
+
+        private async Task MaybeWaitForServerOriginatedCloseAsync()
+        {
+            if (_serverOriginatedChannelCloseTcs is not null)
+            {
+                try
+                {
+                    await _serverOriginatedChannelCloseTcs.Task.WaitAsync(TimeSpan.FromSeconds(5))
+                        .ConfigureAwait(false);
+                }
+                catch
+                {
+                }
+            }
+        }
     }
 }