Cleaning up code.

Author: stebet

RabbitMQTest/Program.cs

@@ -34,7 +34,7 @@ static async Task Main(string[] args)
             subscriber.QueueBind("testqueue", "test", "myawesome.routing.key");
             subscriber.BasicConsume("testqueue", false, "testconsumer", asyncListener);
 
-            byte[] payload = new byte[512];
+            byte[] payload = new byte[16384];
             var batchPublish = Task.Run(async () =>
             {
                 while (messagesSent < batchesToSend * itemsPerBatch)
@@ -49,7 +49,7 @@ static async Task Main(string[] args)
                     }
                     batch.Publish();
                     messagesSent += itemsPerBatch;
-                    await publisher.WaitForConfirmsOrDieAsync();
+                    publisher.WaitForConfirmsOrDie();
                 }
             });
 

projects/client/RabbitMQ.Client/src/client/api/IModel.cs

@@ -496,8 +496,6 @@ void QueueDeclareNoWait(string queue, bool durable,
         /// </remarks>
         [AmqpMethodDoNotImplement(null)]
         bool WaitForConfirms();
-        [AmqpMethodDoNotImplement(null)]
-        Task<bool> WaitForConfirmsAsync();
 
         /// <summary>
         /// Wait until all published messages have been confirmed.
@@ -533,8 +531,6 @@ void QueueDeclareNoWait(string queue, bool durable,
         /// </remarks>
         [AmqpMethodDoNotImplement(null)]
         bool WaitForConfirms(TimeSpan timeout, out bool timedOut);
-        [AmqpMethodDoNotImplement(null)]
-        Task<(bool onlyAcksReceived, bool timedOut)> WaitForConfirmsAsync(TimeSpan timeout);
 
         /// <summary>
         /// Wait until all published messages have been confirmed.
@@ -546,8 +542,6 @@ void QueueDeclareNoWait(string queue, bool durable,
         /// </remarks>
         [AmqpMethodDoNotImplement(null)]
         void WaitForConfirmsOrDie();
-        [AmqpMethodDoNotImplement(null)]
-        Task WaitForConfirmsOrDieAsync();
 
         /// <summary>
         /// Wait until all published messages have been confirmed.
@@ -559,8 +553,6 @@ void QueueDeclareNoWait(string queue, bool durable,
         /// </remarks>
         [AmqpMethodDoNotImplement(null)]
         void WaitForConfirmsOrDie(TimeSpan timeout);
-        [AmqpMethodDoNotImplement(null)]
-        Task WaitForConfirmsOrDieAsync(TimeSpan timeout);
 
         /// <summary>
         /// Amount of time protocol  operations (e.g. <code>queue.declare</code>) are allowed to take before

projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs

@@ -273,7 +273,7 @@ public void Close(ShutdownEventArgs reason, bool abort)
         {
             var k = new ShutdownContinuation();
             ModelShutdown += k.OnConnectionShutdown;
-            _connectionClosingCancellation.Cancel();
+            _connectionClosingCancellation.Cancel(false);
 
             try
             {
@@ -284,7 +284,11 @@ public void Close(ShutdownEventArgs reason, bool abort)
                 }
                 k.Wait(TimeSpan.FromMilliseconds(10000));
                 ConsumerDispatcher.Shutdown(this);
-                _multipleConfirmCleanup.Wait();
+                try
+                {
+                    _multipleConfirmCleanup.Wait();
+                }
+                catch (AggregateException) { }
             }
             catch (AlreadyClosedException)
             {
@@ -1628,23 +1632,16 @@ public async Task CleanupUnconfirmedTagsAsync()
         {
             while (!_connectionClosingCancellation.IsCancellationRequested)
             {
-                try
+                await _multipleConfirmLock.WaitAsync(_connectionClosingCancellation.Token).ConfigureAwait(false);
+                foreach (ulong key in m_unconfirmedSet.Keys)
                 {
-                    await _multipleConfirmLock.WaitAsync(_connectionClosingCancellation.Token).ConfigureAwait(false);
-                    foreach (ulong key in m_unconfirmedSet.Keys)
+                    if (key <= _highestDeliveryTag)
                     {
-                        if (key <= _highestDeliveryTag)
-                        {
-                            m_unconfirmedSet.TryRemove(key, out _);
-                        }
+                        m_unconfirmedSet.TryRemove(key, out _);
                     }
-
-                    TriggerAllOutstandingCompleted();
-                }
-                catch (TaskCanceledException)
-                {
-                    // Swallow the task cancel exception since the model is being closed.
                 }
+
+                TriggerAllOutstandingCompleted();
             }
         }