Pass main loop token forward

Author: danielmarbach

projects/RabbitMQ.Client/IChannel.cs

@@ -265,6 +265,17 @@ Task CloseAsync(ushort replyCode, string replyText, bool abort,
         /// <returns></returns>
         Task CloseAsync(ShutdownEventArgs reason, bool abort);
 
+        /// <summary>
+        /// Asynchronously close this session.
+        /// </summary>
+        /// <param name="reason">The <see cref="ShutdownEventArgs"/> instance containing the close data.</param>
+        /// <param name="abort">Whether or not the close is an abort (ignoring certain exceptions).</param>
+        /// <param name="cancellationToken">CancellationToken for this operation.</param>
+        /// <returns></returns>
+        [Obsolete("7.2.0 - cancellationToken is ignored")]
+        Task CloseAsync(ShutdownEventArgs reason, bool abort,
+            CancellationToken cancellationToken = default);
+
         /// <summary>Asynchronously declare an exchange.</summary>
         /// <param name="exchange">The name of the exchange.</param>
         /// <param name="type">The type of the exchange.</param>

projects/RabbitMQ.Client/IConnectionExtensions.cs

@@ -88,6 +88,21 @@ public static Task CloseAsync(this IConnection connection, ushort reasonCode, st
                 CancellationToken.None);
         }
 
+        /// <summary>
+        /// Asynchronously abort this connection and all its channels.
+        /// </summary>
+        /// <remarks>
+        /// Note that all active channels and sessions will be closed if this method is called.
+        /// In comparison to normal <see cref="CloseAsync(IConnection, CancellationToken)"/> method, <see cref="AbortAsync(IConnection, CancellationToken)"/> will not throw
+        /// <see cref="IOException"/> during closing connection.
+        ///This method waits infinitely for the in-progress close operation to complete.
+        /// </remarks>
+        public static Task AbortAsync(this IConnection connection)
+        {
+            return connection.CloseAsync(Constants.ReplySuccess, "Connection close forced",
+                InternalConstants.DefaultConnectionAbortTimeout, true, default);
+        }
+
         /// <summary>
         /// Asynchronously abort this connection and all its channels.
         /// </summary>
@@ -99,8 +114,27 @@ public static Task CloseAsync(this IConnection connection, ushort reasonCode, st
         /// </remarks>
         public static Task AbortAsync(this IConnection connection, CancellationToken cancellationToken = default)
         {
-            return connection.CloseAsync(Constants.ReplySuccess, "Connection close forced", InternalConstants.DefaultConnectionAbortTimeout, true,
-                cancellationToken);
+            return connection.CloseAsync(Constants.ReplySuccess, "Connection close forced",
+                InternalConstants.DefaultConnectionAbortTimeout, true, cancellationToken);
+        }
+
+        /// <summary>
+        /// Asynchronously abort this connection and all its channels.
+        /// </summary>
+        /// <remarks>
+        /// The method behaves in the same way as <see cref="AbortAsync(IConnection, CancellationToken)"/>, with the only
+        /// difference that the connection is closed with the given connection close code and message.
+        /// <para>
+        /// The close code (See under "Reply Codes" in the AMQP 0-9-1 specification)
+        /// </para>
+        /// <para>
+        /// A message indicating the reason for closing the connection
+        /// </para>
+        /// </remarks>
+        public static Task AbortAsync(this IConnection connection, ushort reasonCode, string reasonText)
+        {
+            return connection.CloseAsync(reasonCode, reasonText,
+                InternalConstants.DefaultConnectionAbortTimeout, true, default);
         }
 
         /// <summary>
@@ -118,8 +152,8 @@ public static Task AbortAsync(this IConnection connection, CancellationToken can
         /// </remarks>
         public static Task AbortAsync(this IConnection connection, ushort reasonCode, string reasonText, CancellationToken cancellationToken = default)
         {
-            return connection.CloseAsync(reasonCode, reasonText, InternalConstants.DefaultConnectionAbortTimeout, true,
-                cancellationToken);
+            return connection.CloseAsync(reasonCode, reasonText,
+                InternalConstants.DefaultConnectionAbortTimeout, true, cancellationToken);
         }
 
         /// <summary>

projects/RabbitMQ.Client/Impl/AutorecoveringChannel.cs

@@ -234,6 +234,12 @@ await _connection.DeleteRecordedChannelAsync(this,
             }
         }
 
+        public Task CloseAsync(ShutdownEventArgs args, bool abort,
+            CancellationToken cancellationToken)
+        {
+            return CloseAsync(args, abort);
+        }
+
         public async Task CloseAsync(ShutdownEventArgs args, bool abort)
         {
             ThrowIfDisposed();

projects/RabbitMQ.Client/Impl/Channel.cs

@@ -205,12 +205,18 @@ public Task CloseAsync(ushort replyCode, string replyText, bool abort,
             return CloseAsync(args, abort);
         }
 
+        public Task CloseAsync(ShutdownEventArgs args, bool abort,
+            CancellationToken cancellationToken)
+        {
+            return CloseAsync(args, abort);
+        }
+
         public async Task CloseAsync(ShutdownEventArgs args, bool abort)
         {
             CancellationToken cancellationToken = args.CancellationToken;
 
             bool enqueued = false;
-            // We should really try to clsoe the connection and therefore we don't allow this to be canceled by the user
+            // We should really try to close the channel and therefore we don't allow this to be canceled by the user
             var k = new ChannelCloseAsyncRpcContinuation(ContinuationTimeout, IsOpen ? CancellationToken.None : cancellationToken);
 
             await _rpcSemaphore.WaitAsync(k.CancellationToken)

projects/RabbitMQ.Client/Impl/Connection.Heartbeat.cs

@@ -109,7 +109,7 @@ private async void HeartbeatReadTimerCallback(object? state)
                     {
                         var eose = new EndOfStreamException($"Heartbeat missing with heartbeat == {_heartbeat} seconds");
                         LogCloseError(eose.Message, eose);
-                        await HandleMainLoopExceptionAsync(new ShutdownEventArgs(ShutdownInitiator.Library, 0, "End of stream", eose))
+                        await HandleMainLoopExceptionAsync(new ShutdownEventArgs(ShutdownInitiator.Library, 0, "End of stream", eose, _mainLoopCts.Token))
                             .ConfigureAwait(false);
                         shouldTerminate = true;
                     }

projects/RabbitMQ.Client/PublicAPI.Unshipped.txt

@@ -5,11 +5,14 @@ RabbitMQ.Client.Exceptions.PublishReturnException.PublishReturnException(ulong p
 RabbitMQ.Client.Exceptions.PublishReturnException.ReplyCode.get -> ushort
 RabbitMQ.Client.Exceptions.PublishReturnException.ReplyText.get -> string!
 RabbitMQ.Client.Exceptions.PublishReturnException.RoutingKey.get -> string!
+RabbitMQ.Client.IChannel.CloseAsync(RabbitMQ.Client.Events.ShutdownEventArgs! reason, bool abort, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
 RabbitMQ.Client.RabbitMQTracingOptions
 RabbitMQ.Client.RabbitMQTracingOptions.RabbitMQTracingOptions() -> void
 RabbitMQ.Client.RabbitMQTracingOptions.UsePublisherAsParent.get -> bool
 RabbitMQ.Client.RabbitMQTracingOptions.UsePublisherAsParent.set -> void
 RabbitMQ.Client.RabbitMQTracingOptions.UseRoutingKeyAsOperationName.get -> bool
 RabbitMQ.Client.RabbitMQTracingOptions.UseRoutingKeyAsOperationName.set -> void
+static RabbitMQ.Client.IConnectionExtensions.AbortAsync(this RabbitMQ.Client.IConnection! connection) -> System.Threading.Tasks.Task!
+static RabbitMQ.Client.IConnectionExtensions.AbortAsync(this RabbitMQ.Client.IConnection! connection, ushort reasonCode, string! reasonText) -> System.Threading.Tasks.Task!
 static RabbitMQ.Client.RabbitMQActivitySource.TracingOptions.get -> RabbitMQ.Client.RabbitMQTracingOptions!
 static RabbitMQ.Client.RabbitMQActivitySource.TracingOptions.set -> void