Add ExchangeBindAsync

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

@@ -276,8 +276,6 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         /// </remarks>
         void ExchangeBind(string destination, string source, string routingKey, IDictionary<string, object> arguments);
 
-        /*
-         * TODO LRB rabbitmq/rabbitmq-dotnet-client#1347
         /// <summary>
         /// Asynchronously binds an exchange to an exchange.
         /// </summary>
@@ -287,7 +285,6 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
         ///   </para>
         /// </remarks>
         ValueTask ExchangeBindAsync(string destination, string source, string routingKey, IDictionary<string, object> arguments);
-        */
 
         /// <summary>
         /// Like ExchangeBind but sets nowait to true.
@@ -308,10 +305,10 @@ ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString rou
 
         /// <summary>Asynchronously declare an exchange.</summary>
         /// <remarks>
-        /// The exchange is declared non-passive and non-internal.
-        /// The "nowait" option is not exercised.
+        /// The exchange is declared non-internal.
+        /// The "nowait" option is not used.
         /// </remarks>
-        ValueTask ExchangeDeclareAsync(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments);
+        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
@@ -403,11 +400,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.

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

@@ -54,7 +54,7 @@ public AsyncRpcContinuation(TimeSpan continuationTimeout)
             {
                 if (_tcs.TrySetCanceled())
                 {
-                    // TODO LRB #1347
+                    // TODO LRB rabbitmq/rabbitmq-dotnet-client#1347
                     // Cancellation was successful, does this mean we should set a TimeoutException
                     // in the same manner as BlockingCell?
                 }
@@ -107,7 +107,7 @@ public override void HandleCommand(in IncomingCommand cmd)
                 else if (cmd.CommandId == ProtocolCommandId.ConnectionTune)
                 {
                     var tune = new ConnectionTune(cmd.MethodBytes.Span);
-                    // TODO LRB #1347
+                    // TODO LRB rabbitmq/rabbitmq-dotnet-client#1347
                     // What to do if setting a result fails?
                     _tcs.TrySetResult(new ConnectionSecureOrTune
                     {
@@ -155,6 +155,13 @@ public override void HandleCommand(in IncomingCommand cmd)
         }
     }
 
+    internal class ExchangeBindAsyncRpcContinuation : SimpleAsyncRpcContinuation
+    {
+        public ExchangeBindAsyncRpcContinuation(TimeSpan continuationTimeout) : base(ProtocolCommandId.ExchangeBindOk, continuationTimeout)
+        {
+        }
+    }
+
     internal class ExchangeDeclareAsyncRpcContinuation : SimpleAsyncRpcContinuation
     {
         public ExchangeDeclareAsyncRpcContinuation(TimeSpan continuationTimeout) : base(ProtocolCommandId.ExchangeDeclareOk, continuationTimeout)
@@ -197,7 +204,7 @@ public override void HandleCommand(in IncomingCommand cmd)
         }
     }
 
-    internal class QueueDeleteAsyncRpcContinuation : AsyncRpcContinuation<QueueDeleteOk>
+    internal class QueueDeleteAsyncRpcContinuation : AsyncRpcContinuation<uint>
     {
         public QueueDeleteAsyncRpcContinuation(TimeSpan continuationTimeout) : base(continuationTimeout)
         {
@@ -210,8 +217,7 @@ public override void HandleCommand(in IncomingCommand cmd)
                 var method = new Client.Framing.Impl.QueueDeleteOk(cmd.MethodBytes.Span);
                 if (cmd.CommandId == ProtocolCommandId.QueueDeleteOk)
                 {
-                    var result = new QueueDeleteOk(method._messageCount);
-                    _tcs.TrySetResult(result);
+                    _tcs.TrySetResult(method._messageCount);
                 }
                 else
                 {

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

@@ -362,6 +362,13 @@ public void ExchangeBind(string destination, string source, string routingKey, I
             _innerChannel.ExchangeBind(destination, source, routingKey, arguments);
         }
 
+        public async ValueTask ExchangeBindAsync(string destination, string source, string routingKey, IDictionary<string, object> arguments)
+        {
+            ThrowIfDisposed();
+            await _innerChannel.ExchangeBindAsync(destination, source, routingKey, arguments);
+            _connection.RecordBinding(new RecordedBinding(false, destination, source, routingKey, arguments));
+        }
+
         public void ExchangeBindNoWait(string destination, string source, string routingKey, IDictionary<string, object> arguments)
             => InnerChannel.ExchangeBindNoWait(destination, source, routingKey, arguments);
 
@@ -372,11 +379,14 @@ public void ExchangeDeclare(string exchange, string type, bool durable, bool aut
             _connection.RecordExchange(new RecordedExchange(exchange, type, durable, autoDelete, arguments));
         }
 
-        public async ValueTask ExchangeDeclareAsync(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments)
+        public async ValueTask ExchangeDeclareAsync(string exchange, string type, bool passive, bool durable, bool autoDelete, IDictionary<string, object> arguments)
         {
             ThrowIfDisposed();
-            await _innerChannel.ExchangeDeclareAsync(exchange, type, durable, autoDelete, arguments);
-            _connection.RecordExchange(new RecordedExchange(exchange, type, durable, autoDelete, arguments));
+            await _innerChannel.ExchangeDeclareAsync(exchange, type, passive, durable, autoDelete, arguments);
+            if (false == passive)
+            {
+                _connection.RecordExchange(new RecordedExchange(exchange, type, durable, autoDelete, arguments));
+            }    
         }
 
         public void ExchangeDeclareNoWait(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments)
@@ -443,11 +453,14 @@ public void QueueDeclareNoWait(string queue, bool durable, bool exclusive, bool
             _connection.RecordQueue(new RecordedQueue(queue, queue.Length == 0, durable, exclusive, autoDelete, arguments));
         }
 
-        public async ValueTask<QueueDeclareOk> QueueDeclareAsync(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments)
+        public async ValueTask<QueueDeclareOk> QueueDeclareAsync(string queue, bool passive, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments)
         {
             ThrowIfDisposed();
-            QueueDeclareOk result = await _innerChannel.QueueDeclareAsync(queue, durable, exclusive, autoDelete, arguments);
-            _connection.RecordQueue(new RecordedQueue(result.QueueName, queue.Length == 0, durable, exclusive, autoDelete, arguments));
+            QueueDeclareOk result = await _innerChannel.QueueDeclareAsync(queue, passive, durable, exclusive, autoDelete, arguments);
+            if (false == passive)
+            {
+                _connection.RecordQueue(new RecordedQueue(result.QueueName, queue.Length == 0, durable, exclusive, autoDelete, arguments));
+            }
             return result;
         }