Implement async for simple AMQP methods

Related to:
* #1345
* #1308
* #970
* #843

Implement QueueDeleteAsync, ExchangeDeclareAsync and ExchangeDeleteAsync. Refactoring to come.

Fix public API

Move rpc continuations to their own files

Add continuation timeouts to new AsyncRpcContinuations classes.

Add ExchangeBindAsync to interface

Fixup compilation after rebase

Add QueueDeleteOk to API

Add ExchangeBindAsync

Add `[Collection("IntegrationFixture")]` to test class that needs it, move test to an integration fixture class

dotnet format fix

Add exchange-exchange binding to test

Add QueueBindAsync and add to tests.

Add ExchangeUnbindAsync method

Add BasicRejectAsync

Add BasicCancelAsync

Add BasicConsumeAsync

Added BasicAckAsync and BasicQosAsync

Author: lukebakken

projects/OAuth2Test/TestOAuth2.cs

@@ -148,7 +148,7 @@ private async Task Publish(IChannel publisher)
         private async ValueTask<IChannel> declareConsumer()
         {
             IChannel subscriber = _connection.CreateChannel();
-            await subscriber.QueueDeclareAsync("testqueue", true, false, false, arguments: null);
+            await subscriber.QueueDeclareAsync(queue: "testqueue", passive: false, true, false, false, arguments: null);
             subscriber.QueueBind("testqueue", Exchange, "hello");
             return subscriber;
         }

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

@@ -151,37 +151,61 @@ public interface IChannel : IDisposable
         /// </remarks>
         event EventHandler<ShutdownEventArgs> ChannelShutdown;
 
-        /// <summary>
-        /// Acknowledge one or more delivered message(s).
-        /// </summary>
+        /// <summary>Acknknowledges one or more messages.</summary>
+        /// <param name="deliveryTag">The delivery tag.</param>
+        /// <param name="multiple">Ack all messages up to the delivery tag if set to <c>true</c>.</param>
         void BasicAck(ulong deliveryTag, bool multiple);
 
-        /// <summary>
-        /// Delete a Basic content-class consumer.
-        /// </summary>
+        /// <summary>Asynchronously acknknowledges one or more messages.</summary>
+        /// <param name="deliveryTag">The delivery tag.</param>
+        /// <param name="multiple">Ack all messages up to the delivery tag if set to <c>true</c>.</param>
+        ValueTask BasicAckAsync(ulong deliveryTag, bool multiple);
+
+        /// <summary>Cancel a Basic content-class consumer.</summary>
+        /// <param name="consumerTag">The consumer tag.</param>
         void BasicCancel(string consumerTag);
 
+        /// <summary>Asynchronously cancel a Basic content-class consumer.</summary>
+        /// <param name="consumerTag">The consumer tag.</param>
+        ValueTask BasicCancelAsync(string consumerTag);
+
         /// <summary>
         /// Same as BasicCancel but sets nowait to true and returns void (as there
         /// will be no response from the server).
         /// </summary>
+        /// <param name="consumerTag">The consumer tag.</param>
         void BasicCancelNoWait(string consumerTag);
 
         /// <summary>Start a Basic content-class consumer.</summary>
-        string BasicConsume(
-            string queue,
-            bool autoAck,
-            string consumerTag,
-            bool noLocal,
-            bool exclusive,
-            IDictionary<string, object> arguments,
-            IBasicConsumer consumer);
+        /// <param name="queue">The queue.</param>
+        /// <param name="autoAck">If set to <c>true</c>, automatically ack messages.</param>
+        /// <param name="consumerTag">The consumer tag.</param>
+        /// <param name="noLocal">If set to <c>true</c>, this consumer will not receive messages published by the same connection.</param>
+        /// <param name="exclusive">If set to <c>true</c>, the consumer is exclusive.</param>
+        /// <param name="arguments">Consumer arguments.</param>
+        /// <param name="consumer">The consumer, an instance of <see cref="IBasicConsumer"/></param>
+        /// <returns></returns>
+        string BasicConsume(string queue, bool autoAck, string consumerTag, bool noLocal, bool exclusive, IDictionary<string, object> arguments, IBasicConsumer consumer);
+
+        /// <summary>Asynchronously start a Basic content-class consumer.</summary>
+        /// <param name="queue">The queue.</param>
+        /// <param name="autoAck">If set to <c>true</c>, automatically ack messages.</param>
+        /// <param name="consumerTag">The consumer tag.</param>
+        /// <param name="noLocal">If set to <c>true</c>, this consumer will not receive messages published by the same connection.</param>
+        /// <param name="exclusive">If set to <c>true</c>, the consumer is exclusive.</param>
+        /// <param name="arguments">Consumer arguments.</param>
+        /// <param name="consumer">The consumer, an instance of <see cref="IBasicConsumer"/></param>
+        /// <returns></returns>
+        ValueTask<string> BasicConsumeAsync(string queue, bool autoAck, string consumerTag, bool noLocal, bool exclusive, IDictionary<string, object> arguments, IBasicConsumer consumer);
 
         /// <summary>
         /// Retrieve an individual message, if
         /// one is available; returns null if the server answers that
-        /// no messages are currently available. See also <see cref="IChannel.BasicAck"/>.
+        /// no messages are currently available. See also <see cref="IChannel.BasicAck" />.
         /// </summary>
+        /// <param name="queue">The queue.</param>
+        /// <param name="autoAck">If set to <c>true</c>, automatically ack the message.</param>
+        /// <returns><see cref="BasicGetResult"/></returns>
         BasicGetResult BasicGet(string queue, bool autoAck);
 
         /// <summary>Reject one or more delivered message(s).</summary>
@@ -238,23 +262,41 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         /// <summary>
         /// Configures QoS parameters of the Basic content-class.
         /// </summary>
+        /// <param name="prefetchSize">Size of the prefetch in bytes.</param>
+        /// <param name="prefetchCount">The prefetch count.</param>
+        /// <param name="global">If set to <c>true</c>, use global prefetch.
+        /// See the <seealso href="https://www.rabbitmq.com/consumer-prefetch.html#overview">Consumer Prefetch documentation</seealso>.</param>
         void BasicQos(uint prefetchSize, ushort prefetchCount, bool global);
 
+        /// <summary>
+        /// Configures QoS parameters of the Basic content-class.
+        /// </summary>
+        /// <param name="prefetchSize">Size of the prefetch in bytes.</param>
+        /// <param name="prefetchCount">The prefetch count.</param>
+        /// <param name="global">If set to <c>true</c>, use global prefetch.
+        /// See the <seealso href="https://www.rabbitmq.com/consumer-prefetch.html#overview">Consumer Prefetch documentation</seealso>.</param>
+        ValueTask BasicQosAsync(uint prefetchSize, ushort prefetchCount, bool global);
+
         /// <summary>
         /// Indicates that a consumer has recovered.
         /// Deprecated. Should not be used.
         /// </summary>
+        [Obsolete]
         void BasicRecover(bool requeue);
 
         /// <summary>
         /// Indicates that a consumer has recovered.
         /// Deprecated. Should not be used.
         /// </summary>
+        [Obsolete]
         void BasicRecoverAsync(bool requeue);
 
         /// <summary> Reject a delivered message.</summary>
         void BasicReject(ulong deliveryTag, bool requeue);
 
+        /// <summary> Reject a delivered message.</summary>
+        ValueTask BasicRejectAsync(ulong deliveryTag, bool requeue);
+
         /// <summary>Close this session.</summary>
         /// <param name="replyCode">The reply code to send for closing (See under "Reply Codes" in the AMQP specification).</param>
         /// <param name="replyText">The reply text to send for closing.</param>
@@ -276,6 +318,16 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         /// </remarks>
         void ExchangeBind(string destination, string source, string routingKey, IDictionary<string, object> arguments);
 
+        /// <summary>
+        /// Asynchronously binds an exchange to an exchange.
+        /// </summary>
+        /// <remarks>
+        ///   <para>
+        ///     Routing key must be shorter than 255 bytes.
+        ///   </para>
+        /// </remarks>
+        ValueTask ExchangeBindAsync(string destination, string source, string routingKey, IDictionary<string, object> arguments);
+
         /// <summary>
         /// Like ExchangeBind but sets nowait to true.
         /// </summary>
@@ -289,10 +341,17 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         /// <summary>Declare an exchange.</summary>
         /// <remarks>
         /// The exchange is declared non-passive and non-internal.
-        /// The "nowait" option is not exercised.
+        /// The "nowait" option is not used.
         /// </remarks>
         void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments);
 
+        /// <summary>Asynchronously declare an exchange.</summary>
+        /// <remarks>
+        /// The exchange is declared non-internal.
+        /// The "nowait" option is not used.
+        /// </remarks>
+        ValueTask ExchangeDeclareAsync(string exchange, string type, bool passive, bool durable, bool autoDelete, IDictionary<string, object> arguments);
+
         /// <summary>
         /// Same as ExchangeDeclare but sets nowait to true and returns void (as there
         /// will be no response from the server).
@@ -315,6 +374,19 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         /// </summary>
         void ExchangeDelete(string exchange, bool ifUnused);
 
+        /*
+         * TODO LRB rabbitmq/rabbitmq-dotnet-client#1347
+        /// <summary>
+        /// Asynchronously delete an exchange.
+        /// </summary>
+        ValueTask ExchangeDeleteAsync(string exchange, bool ifUnused);
+        */
+
+        /// <summary>
+        /// Asynchronously delete an exchange.
+        /// </summary>
+        ValueTask ExchangeDeleteAsync(string exchange, bool ifUnused);
+
         /// <summary>
         /// Like ExchangeDelete but sets nowait to true.
         /// </summary>
@@ -328,6 +400,14 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         /// </remarks>
         void ExchangeUnbind(string destination, string source, string routingKey, IDictionary<string, object> arguments);
 
+        /// <summary>
+        /// Asynchronously unbind an exchange from an exchange.
+        /// </summary>
+        /// <remarks>
+        /// Routing key must be shorter than 255 bytes.
+        /// </remarks>
+        ValueTask ExchangeUnbindAsync(string destination, string source, string routingKey, IDictionary<string, object> arguments);
+
         /// <summary>
         /// Like ExchangeUnbind but sets nowait to true.
         /// </summary>
@@ -348,6 +428,16 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         /// </remarks>
         void QueueBind(string queue, string exchange, string routingKey, IDictionary<string, object> arguments);
 
+        /// <summary>
+        /// Asynchronously bind a queue to an exchange.
+        /// </summary>
+        /// <remarks>
+        ///   <para>
+        ///     Routing key must be shorter than 255 bytes.
+        ///   </para>
+        /// </remarks>
+        ValueTask QueueBindAsync(string queue, string exchange, string routingKey, IDictionary<string, object> arguments);
+
         /// <summary>Same as QueueBind but sets nowait parameter to true.</summary>
         /// <remarks>
         ///   <para>
@@ -370,11 +460,12 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         /// Asynchronously declares a queue. See the <a href="https://www.rabbitmq.com/queues.html">Queues guide</a> to learn more.
         /// </summary>
         /// <param name="queue">The name of the queue. Pass an empty string to make the server generate a name.</param>
+        /// <param name="passive">Set to <code>true</code> to passively declare the queue (i.e. check for its existence)</param>
         /// <param name="durable">Should this queue will survive a broker restart?</param>
         /// <param name="exclusive">Should this queue use be limited to its declaring connection? Such a queue will be deleted when its declaring connection closes.</param>
         /// <param name="autoDelete">Should this queue be auto-deleted when its last consumer (if any) unsubscribes?</param>
         /// <param name="arguments">Optional; additional queue arguments, e.g. "x-queue-type"</param>
-        ValueTask<QueueDeclareOk> QueueDeclareAsync(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments);
+        ValueTask<QueueDeclareOk> QueueDeclareAsync(string queue, bool passive, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments);
 
         /// <summary>
         /// Declares a queue. See the <a href="https://www.rabbitmq.com/queues.html">Queues guide</a> to learn more.
@@ -411,17 +502,30 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         uint ConsumerCount(string queue);
 
         /// <summary>
-        /// Delete a queue.
+        /// Deletes a queue. See the <a href="https://www.rabbitmq.com/queues.html">Queues guide</a> to learn more.
+        /// </summary>
+        /// <param name="queue">The name of the queue.</param>
+        /// <param name="ifUnused">Only delete the queue if it is unused.</param>
+        /// <param name="ifEmpty">Only delete the queue if it is empty.</param>
+        /// <returns>Returns the number of messages purged during deletion.</returns>
+        uint QueueDelete(string queue, bool ifUnused, bool ifEmpty);
+
+        /// <summary>
+        /// Asynchronously deletes a queue. See the <a href="https://www.rabbitmq.com/queues.html">Queues guide</a> to learn more.
         /// </summary>
         /// <remarks>
         ///Returns the number of messages purged during queue deletion.
         /// </remarks>
-        uint QueueDelete(string queue, bool ifUnused, bool ifEmpty);
+        ValueTask<uint> QueueDeleteAsync(string queue, bool ifUnused, bool ifEmpty);
 
         /// <summary>
         ///Same as QueueDelete but sets nowait parameter to true
         ///and returns void (as there will be no response from the server)
         /// </summary>
+        /// <param name="queue">The name of the queue.</param>
+        /// <param name="ifUnused">Only delete the queue if it is unused.</param>
+        /// <param name="ifEmpty">Only delete the queue if it is empty.</param>
+        /// <returns>Returns the number of messages purged during deletion.</returns>
         void QueueDeleteNoWait(string queue, bool ifUnused, bool ifEmpty);
 
         /// <summary>

projects/RabbitMQ.Client/client/framing/Channel.cs

@@ -234,6 +234,12 @@ public override void BasicAck(ulong deliveryTag, bool multiple)
             ChannelSend(new BasicAck(deliveryTag, multiple));
         }
 
+        public override ValueTask BasicAckAsync(ulong deliveryTag, bool multiple)
+        {
+            var method = new BasicAck(deliveryTag, multiple);
+            return ModelSendAsync(method);
+        }
+
         public override void BasicNack(ulong deliveryTag, bool multiple, bool requeue)
         {
             ChannelSend(new BasicNack(deliveryTag, multiple, requeue));
@@ -254,6 +260,12 @@ public override void BasicReject(ulong deliveryTag, bool requeue)
             ChannelSend(new BasicReject(deliveryTag, requeue));
         }
 
+        public override ValueTask BasicRejectAsync(ulong deliveryTag, bool requeue)
+        {
+            var method = new BasicReject(deliveryTag, requeue);
+            return ModelSendAsync(method);
+        }
+
         public override void QueueUnbind(string queue, string exchange, string routingKey, IDictionary<string, object> arguments)
         {
             ChannelRpc(new QueueUnbind(queue, exchange, routingKey, arguments), ProtocolCommandId.QueueUnbindOk);
@@ -295,13 +307,11 @@ protected override bool DispatchAsynchronous(in IncomingCommand cmd)
                     }
                 case ProtocolCommandId.BasicCancelOk:
                     {
-                        HandleBasicCancelOk(in cmd);
-                        return true;
+                        return HandleBasicCancelOk(in cmd);
                     }
                 case ProtocolCommandId.BasicConsumeOk:
                     {
-                        HandleBasicConsumeOk(in cmd);
-                        return true;
+                        return HandleBasicConsumeOk(in cmd);
                     }
                 case ProtocolCommandId.BasicGetEmpty:
                     {