* Ensure that dequeued RPC continuations are always disposed. Found by @DenisMayorko

Author: lukebakken

projects/Applications/CreateChannel/Program.cs

@@ -85,8 +85,6 @@ public static async Task Main()
             Console.WriteLine($"{channelsOpened,5}");
             Console.WriteLine();
             Console.WriteLine($"Took {watch.Elapsed.TotalMilliseconds} ms");
-
-            Console.ReadLine();
         }
     }
 }

projects/RabbitMQ.Client/Impl/Channel.cs

@@ -744,11 +744,13 @@ protected async Task<bool> HandleChannelCloseOkAsync(IncomingCommand cmd, Cancel
             await FinishCloseAsync(cancellationToken)
                 .ConfigureAwait(false);
 
-            if (_continuationQueue.TryPeek<ChannelCloseAsyncRpcContinuation>(out ChannelCloseAsyncRpcContinuation? k))
+            if (_continuationQueue.TryPeek(out ChannelCloseAsyncRpcContinuation? k))
             {
-                _continuationQueue.Next();
-                await k.HandleCommandAsync(cmd)
-                    .ConfigureAwait(false);
+                using (IRpcContinuation c = _continuationQueue.Next())
+                {
+                    await k.HandleCommandAsync(cmd)
+                        .ConfigureAwait(false);
+                }
             }
 
             return true;
@@ -818,10 +820,12 @@ await ModelSendAsync(in replyMethod, cancellationToken)
 
         protected async Task<bool> HandleConnectionSecureAsync(IncomingCommand cmd, CancellationToken cancellationToken)
         {
-            var k = (ConnectionSecureOrTuneAsyncRpcContinuation)_continuationQueue.Next();
-            await k.HandleCommandAsync(new IncomingCommand())
-                .ConfigureAwait(false); // release the continuation.
-            return true;
+            using (var k = (ConnectionSecureOrTuneAsyncRpcContinuation)_continuationQueue.Next())
+            {
+                await k.HandleCommandAsync(new IncomingCommand())
+                    .ConfigureAwait(false); // release the continuation.
+                return true;
+            }
         }
 
         protected async Task<bool> HandleConnectionStartAsync(IncomingCommand cmd, CancellationToken cancellationToken)

projects/RabbitMQ.Client/Impl/RpcContinuationQueue.cs

@@ -97,7 +97,10 @@ public void Enqueue(IRpcContinuation k)
         ///</remarks>
         public void HandleChannelShutdown(ShutdownEventArgs reason)
         {
-            Next().HandleChannelShutdown(reason);
+            using (IRpcContinuation c = Next())
+            {
+                c.HandleChannelShutdown(reason);
+            }
         }
 
         ///<summary>Retrieve the next waiting continuation.</summary>