* Change IConnection event ConnectionShutdown to AsyncEventHandler

.github/workflows/build-test.yaml

@@ -73,8 +73,8 @@ jobs:
           Receive-Job -Job $tx; `
           & "${{ github.workspace }}\.ci\windows\toxiproxy\toxiproxy-cli.exe" list; `
           dotnet test `
-            --environment "RABBITMQ_RABBITMQCTL_PATH=${{ steps.install-start-rabbitmq.outputs.path }}" `
             --environment 'RABBITMQ_LONG_RUNNING_TESTS=true' `
+            --environment "RABBITMQ_RABBITMQCTL_PATH=${{ steps.install-start-rabbitmq.outputs.path }}" `
             --environment 'RABBITMQ_TOXIPROXY_TESTS=true' `
             --environment 'PASSWORD=grapefruit' `
             --environment SSL_CERTS_DIR="${{ github.workspace }}\.ci\certs" `
@@ -114,7 +114,12 @@ jobs:
         id: install-start-rabbitmq
         run: ${{ github.workspace }}\.ci\windows\gha-setup.ps1
       - name: Sequential Integration Tests
-        run: dotnet test --environment "RABBITMQ_RABBITMQCTL_PATH=${{ steps.install-start-rabbitmq.outputs.path }}" "${{ github.workspace }}\projects\Test\SequentialIntegration\SequentialIntegration.csproj" --no-restore --no-build --logger 'console;verbosity=detailed'
+        run: dotnet test `
+            --environment 'RABBITMQ_LONG_RUNNING_TESTS=true' `
+            --environment "RABBITMQ_RABBITMQCTL_PATH=${{ steps.install-start-rabbitmq.outputs.path }}" `
+            --environment 'PASSWORD=grapefruit' `
+            --environment SSL_CERTS_DIR="${{ github.workspace }}\.ci\certs" `
+            "${{ github.workspace }}\projects\Test\SequentialIntegration\SequentialIntegration.csproj" --no-restore --no-build --logger 'console;verbosity=detailed'
       - name: Maybe upload RabbitMQ logs
         if: failure()
         uses: actions/upload-artifact@v4
@@ -182,8 +187,8 @@ jobs:
       - name: Integration Tests
         run: |
             dotnet test \
-                --environment "RABBITMQ_RABBITMQCTL_PATH=DOCKER:${{ steps.start-rabbitmq.outputs.id }}" \
                 --environment 'RABBITMQ_LONG_RUNNING_TESTS=true' \
+                --environment "RABBITMQ_RABBITMQCTL_PATH=DOCKER:${{ steps.start-rabbitmq.outputs.id }}" \
                 --environment 'RABBITMQ_TOXIPROXY_TESTS=true' \
                 --environment 'PASSWORD=grapefruit' \
                 --environment SSL_CERTS_DIR="${{ github.workspace }}/.ci/certs" \
@@ -222,7 +227,10 @@ jobs:
       - name: Sequential Integration Tests
         run: |
             dotnet test \
+                --environment 'RABBITMQ_LONG_RUNNING_TESTS=true' \
                 --environment "RABBITMQ_RABBITMQCTL_PATH=DOCKER:${{ steps.start-rabbitmq.outputs.id }}" \
+                --environment 'PASSWORD=grapefruit' \
+                --environment SSL_CERTS_DIR="${{ github.workspace }}/.ci/certs" \
                 "${{ github.workspace }}/projects/Test/SequentialIntegration/SequentialIntegration.csproj" --no-restore --no-build --logger 'console;verbosity=detailed'
       - name: Maybe upload RabbitMQ logs
         if: failure()

Makefile

@@ -21,13 +21,17 @@ build:
 test:
 	dotnet test $(CURDIR)/projects/Test/Unit/Unit.csproj --logger 'console;verbosity=detailed'
 	dotnet test --environment 'GITHUB_ACTIONS=true' \
-		--environment "RABBITMQ_RABBITMQCTL_PATH=DOCKER:$$(docker inspect --format='{{.Id}}' $(RABBITMQ_DOCKER_NAME))" \
 		--environment 'RABBITMQ_LONG_RUNNING_TESTS=true' \
+		--environment "RABBITMQ_RABBITMQCTL_PATH=DOCKER:$$(docker inspect --format='{{.Id}}' $(RABBITMQ_DOCKER_NAME))" \
 		--environment 'RABBITMQ_TOXIPROXY_TESTS=true' \
 		--environment 'PASSWORD=grapefruit' \
 		--environment SSL_CERTS_DIR="$(CURDIR)/.ci/certs" \
 		"$(CURDIR)/projects/Test/Integration/Integration.csproj" --logger 'console;verbosity=detailed'
-	dotnet test --environment "RABBITMQ_RABBITMQCTL_PATH=DOCKER:$$(docker inspect --format='{{.Id}}' $(RABBITMQ_DOCKER_NAME))" $(CURDIR)/projects/Test/SequentialIntegration/SequentialIntegration.csproj --logger 'console;verbosity=detailed'
+	dotnet test --environment 'RABBITMQ_LONG_RUNNING_TESTS=true' \
+		--environment "RABBITMQ_RABBITMQCTL_PATH=DOCKER:$$(docker inspect --format='{{.Id}}' $(RABBITMQ_DOCKER_NAME))" \
+		--environment 'PASSWORD=grapefruit' \
+		--environment SSL_CERTS_DIR="$(CURDIR)/.ci/certs" \
+		$(CURDIR)/projects/Test/SequentialIntegration/SequentialIntegration.csproj --logger 'console;verbosity=detailed'
 
 # Note:
 # You must have the expected OAuth2 environment set up for this target

projects/RabbitMQ.Client/PublicAPI.Unshipped.txt

@@ -505,7 +505,7 @@ RabbitMQ.Client.IConnection.ClientProvidedName.get -> string
 RabbitMQ.Client.IConnection.CloseReason.get -> RabbitMQ.Client.ShutdownEventArgs
 RabbitMQ.Client.IConnection.ConnectionBlocked -> System.EventHandler<RabbitMQ.Client.Events.ConnectionBlockedEventArgs>
 RabbitMQ.Client.IConnection.ConnectionRecoveryError -> System.EventHandler<RabbitMQ.Client.Events.ConnectionRecoveryErrorEventArgs>
-RabbitMQ.Client.IConnection.ConnectionShutdown -> System.EventHandler<RabbitMQ.Client.ShutdownEventArgs>
+RabbitMQ.Client.IConnection.ConnectionShutdownAsync -> RabbitMQ.Client.Events.AsyncEventHandler<RabbitMQ.Client.ShutdownEventArgs>
 RabbitMQ.Client.IConnection.ConnectionUnblocked -> System.EventHandler<System.EventArgs>
 RabbitMQ.Client.IConnection.ConsumerTagChangeAfterRecovery -> System.EventHandler<RabbitMQ.Client.Events.ConsumerTagChangedAfterRecoveryEventArgs>
 RabbitMQ.Client.IConnection.Endpoint.get -> RabbitMQ.Client.AmqpTcpEndpoint

projects/RabbitMQ.Client/client/api/IConnection.cs

@@ -155,7 +155,7 @@ public interface IConnection : INetworkConnection, IDisposable
         /// event handler is added to this event, the event handler
         /// will be fired immediately.
         /// </remarks>
-        event EventHandler<ShutdownEventArgs> ConnectionShutdown;
+        event AsyncEventHandler<ShutdownEventArgs> ConnectionShutdownAsync;
 
         /// <summary>
         /// Raised when the connection completes recovery.

projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recovery.cs

@@ -46,7 +46,7 @@ internal sealed partial class AutorecoveringConnection
         private Task? _recoveryTask;
         private readonly CancellationTokenSource _recoveryCancellationTokenSource = new CancellationTokenSource();
 
-        private void HandleConnectionShutdown(object _, ShutdownEventArgs args)
+        private Task HandleConnectionShutdownAsync(object _, ShutdownEventArgs args)
         {
             if (ShouldTriggerConnectionRecovery(args))
             {
@@ -80,6 +80,8 @@ static bool ShouldTriggerConnectionRecovery(ShutdownEventArgs args)
 
                 return false;
             }
+
+            return Task.CompletedTask;
         }
 
         private async Task RecoverConnectionAsync()

projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs

@@ -88,7 +88,7 @@ void onException(Exception exception, string context) =>
             _consumerTagChangeAfterRecoveryWrapper = new EventingWrapper<ConsumerTagChangedAfterRecoveryEventArgs>("OnConsumerRecovery", onException);
             _queueNameChangedAfterRecoveryWrapper = new EventingWrapper<QueueNameChangedAfterRecoveryEventArgs>("OnQueueRecovery", onException);
 
-            ConnectionShutdown += HandleConnectionShutdown;
+            ConnectionShutdownAsync += HandleConnectionShutdownAsync;
         }
 
         public event EventHandler<EventArgs> RecoverySucceeded
@@ -117,10 +117,10 @@ public event EventHandler<ConnectionBlockedEventArgs> ConnectionBlocked
             remove => InnerConnection.ConnectionBlocked -= value;
         }
 
-        public event EventHandler<ShutdownEventArgs> ConnectionShutdown
+        public event AsyncEventHandler<ShutdownEventArgs> ConnectionShutdownAsync
         {
-            add => InnerConnection.ConnectionShutdown += value;
-            remove => InnerConnection.ConnectionShutdown -= value;
+            add => InnerConnection.ConnectionShutdownAsync += value;
+            remove => InnerConnection.ConnectionShutdownAsync -= value;
         }
 
         public event EventHandler<EventArgs> ConnectionUnblocked

projects/RabbitMQ.Client/client/impl/ChannelBase.cs

@@ -84,6 +84,7 @@ protected ChannelBase(ConnectionConfig config, ISession session)
 
             Action<Exception, string> onException = (exception, context) =>
                 OnCallbackException(CallbackExceptionEventArgs.Build(exception, context));
+
             _basicAcksWrapper = new EventingWrapper<BasicAckEventArgs>("OnBasicAck", onException);
             _basicNacksWrapper = new EventingWrapper<BasicNackEventArgs>("OnBasicNack", onException);
             _basicReturnWrapper = new EventingWrapper<BasicReturnEventArgs>("OnBasicReturn", onException);
@@ -93,7 +94,7 @@ protected ChannelBase(ConnectionConfig config, ISession session)
             _channelShutdownWrapper = new EventingWrapper<ShutdownEventArgs>("OnChannelShutdown", onException);
             _recoveryWrapper = new EventingWrapper<EventArgs>("OnChannelRecovery", onException);
             session.CommandReceived = HandleCommandAsync;
-            session.SessionShutdown += OnSessionShutdown;
+            session.SessionShutdownAsync += OnSessionShutdownAsync;
             Session = session;
         }
 
@@ -407,12 +408,13 @@ await ModelSendAsync(method, k.CancellationToken)
             }
         }
 
-        internal void FinishClose()
+        internal async Task FinishCloseAsync()
         {
             ShutdownEventArgs reason = CloseReason;
             if (reason != null)
             {
-                Session.Close(reason);
+                await Session.CloseAsync(reason)
+                    .ConfigureAwait(false);
             }
 
             m_connectionStartCell?.TrySetResult(null);
@@ -488,7 +490,7 @@ private void OnChannelShutdown(ShutdownEventArgs reason)
                     if (_confirmsTaskCompletionSources?.Count > 0)
                     {
                         var exception = new AlreadyClosedException(reason);
-                        foreach (var confirmsTaskCompletionSource in _confirmsTaskCompletionSources)
+                        foreach (TaskCompletionSource<bool> confirmsTaskCompletionSource in _confirmsTaskCompletionSources)
                         {
                             confirmsTaskCompletionSource.TrySetException(exception);
                         }
@@ -505,14 +507,12 @@ private void OnChannelShutdown(ShutdownEventArgs reason)
             _flowControlBlock.Set();
         }
 
-        // TODO async
-        private void OnSessionShutdown(object sender, ShutdownEventArgs reason)
+        private Task OnSessionShutdownAsync(object sender, ShutdownEventArgs reason)
         {
             ConsumerDispatcher.Quiesce();
             SetCloseReason(reason);
             OnChannelShutdown(reason);
-            // TODO async
-            ConsumerDispatcher.Shutdown(reason);
+            return ConsumerDispatcher.ShutdownAsync(reason);
         }
 
         internal bool SetCloseReason(ShutdownEventArgs reason)
@@ -723,7 +723,8 @@ protected async Task<bool> HandleChannelCloseAsync(IncomingCommand cmd, Cancella
                     channelClose._classId,
                     channelClose._methodId));
 
-                Session.Close(CloseReason, false);
+                await Session.CloseAsync(CloseReason, false)
+                    .ConfigureAwait(false);
 
                 var method = new ChannelCloseOk();
                 await ModelSendAsync(method, cancellationToken)
@@ -734,7 +735,8 @@ await ModelSendAsync(method, cancellationToken)
             finally
             {
                 cmd.ReturnBuffers();
-                Session.Notify();
+                await Session.NotifyAsync()
+                    .ConfigureAwait(false);
             }
         }
 
@@ -746,7 +748,8 @@ protected async Task<bool> HandleChannelCloseOkAsync(IncomingCommand cmd, Cancel
                  * Note:
                  * This call _must_ come before completing the async continuation
                  */
-                FinishClose();
+                await FinishCloseAsync()
+                    .ConfigureAwait(false);
 
                 if (_continuationQueue.TryPeek<ChannelCloseAsyncRpcContinuation>(out var k))
                 {
@@ -815,7 +818,8 @@ protected async Task<bool> HandleConnectionCloseAsync(IncomingCommand cmd, Cance
                 var reason = new ShutdownEventArgs(ShutdownInitiator.Peer, method._replyCode, method._replyText, method._classId, method._methodId);
                 try
                 {
-                    Session.Connection.ClosedViaPeer(reason);
+                    await Session.Connection.ClosedViaPeerAsync(reason)
+                        .ConfigureAwait(false);
 
                     var replyMethod = new ConnectionCloseOk();
                     await ModelSendAsync(replyMethod, cancellationToken)

projects/RabbitMQ.Client/client/impl/Connection.Heartbeat.cs

@@ -109,7 +109,8 @@ private async void HeartbeatReadTimerCallback(object? state)
                     {
                         var eose = new EndOfStreamException($"Heartbeat missing with heartbeat == {_heartbeat} seconds");
                         LogCloseError(eose.Message, eose);
-                        HandleMainLoopException(new ShutdownEventArgs(ShutdownInitiator.Library, 0, "End of stream", eose));
+                        await HandleMainLoopExceptionAsync(new ShutdownEventArgs(ShutdownInitiator.Library, 0, "End of stream", eose))
+                            .ConfigureAwait(false);
                         shouldTerminate = true;
                     }
                 }
@@ -158,6 +159,13 @@ await WriteAsync(Client.Impl.Framing.Heartbeat.GetHeartbeatFrame(), _mainLoopCts
                     _heartbeatWriteTimer?.Change((int)_heartbeatWriteTimeSpan.TotalMilliseconds, Timeout.Infinite);
                 }
             }
+            catch (OperationCanceledException ocex)
+            {
+                if (ocex.CancellationToken != _mainLoopCts.Token)
+                {
+                    throw;
+                }
+            }
             catch (ObjectDisposedException)
             {
                 // timer is already disposed,

projects/RabbitMQ.Client/client/impl/Connection.Receive.cs

@@ -59,7 +59,8 @@ await ReceiveLoopAsync(mainLoopToken)
                 var ea = new ShutdownEventArgs(ShutdownInitiator.Library,
                     0, "End of stream",
                     exception: eose);
-                HandleMainLoopException(ea);
+                await HandleMainLoopExceptionAsync(ea)
+                    .ConfigureAwait(false);
             }
             catch (HardProtocolException hpe)
             {
@@ -75,15 +76,29 @@ await HardProtocolExceptionHandlerAsync(hpe, mainLoopToken)
                 var ea = new ShutdownEventArgs(ShutdownInitiator.Library,
                     Constants.InternalError, fileLoadException.Message,
                     exception: fileLoadException);
-                HandleMainLoopException(ea);
+                await HandleMainLoopExceptionAsync(ea)
+                    .ConfigureAwait(false);
+            }
+            catch (OperationCanceledException ocex)
+            {
+                if (ocex.CancellationToken != mainLoopToken)
+                {
+                    var ea = new ShutdownEventArgs(ShutdownInitiator.Library,
+                        Constants.InternalError,
+                        $"Unexpected Exception: {ocex.Message}",
+                        exception: ocex);
+                    await HandleMainLoopExceptionAsync(ea)
+                        .ConfigureAwait(false);
+                }
             }
             catch (Exception ex)
             {
                 var ea = new ShutdownEventArgs(ShutdownInitiator.Library,
                     Constants.InternalError,
                     $"Unexpected Exception: {ex.Message}",
                     exception: ex);
-                HandleMainLoopException(ea);
+                await HandleMainLoopExceptionAsync(ea)
+                    .ConfigureAwait(false);
             }
 
             using var cts = new CancellationTokenSource(InternalConstants.DefaultConnectionCloseTimeout);
@@ -175,26 +190,29 @@ private void MaybeTerminateMainloopAndStopHeartbeatTimers(bool cancelMainLoop =
             MaybeStopHeartbeatTimers();
         }
 
-        private void HandleMainLoopException(ShutdownEventArgs reason)
+        private Task HandleMainLoopExceptionAsync(ShutdownEventArgs reason)
         {
             string message = reason.GetLogMessage();
             if (false == SetCloseReason(reason))
             {
                 LogCloseError($"Unexpected Main Loop Exception while closing: {message}", reason.Exception);
-                return;
+                return Task.CompletedTask;
             }
 
             _channel0.MaybeSetConnectionStartException(reason.Exception);
 
-            OnShutdown(reason);
             LogCloseError($"Unexpected connection closure: {message}", reason.Exception);
+
+            return OnShutdownAsync(reason);
         }
 
         private async Task HardProtocolExceptionHandlerAsync(HardProtocolException hpe, CancellationToken cancellationToken)
         {
             if (SetCloseReason(hpe.ShutdownReason))
             {
-                OnShutdown(hpe.ShutdownReason);
+                await OnShutdownAsync(hpe.ShutdownReason)
+                    .ConfigureAwait(false);
+
                 await _session0.SetSessionClosingAsync(false)
                     .ConfigureAwait(false);
                 try