Introduce Channel#consumerCount

Author: michaelklishin

src/com/rabbitmq/client/Channel.java

@@ -909,4 +909,14 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      * @throws IOException Problem transmitting method.
      */
     long messageCount(String queue) throws IOException;
+
+    /**
+     * Returns the number of consumers on a queue.
+     * This method assumes the queue exists. If it doesn't,
+     * an exception will be closed with an exception.
+     * @param queue the name of the queue
+     * @return the number of consumers
+     * @throws IOException Problem transmitting method.
+     */
+    long consumerCount(String queue) throws IOException;
 }

src/com/rabbitmq/client/impl/ChannelN.java

@@ -889,6 +889,12 @@ public long messageCount(String queue) throws IOException {
         return ok.getMessageCount();
     }
 
+    /** Public API - {@inheritDoc} */
+    public long consumerCount(String queue) throws IOException {
+        Queue.DeclareOk ok = queueDeclarePassive(queue);
+        return ok.getConsumerCount();
+    }
+
     /** Public API - {@inheritDoc} */
     public Queue.DeleteOk queueDelete(String queue, boolean ifUnused, boolean ifEmpty)
         throws IOException

src/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java

@@ -291,6 +291,10 @@ public long messageCount(String queue) throws IOException {
         return delegate.messageCount(queue);
     }
 
+    public long consumerCount(String queue) throws IOException {
+        return delegate.consumerCount(queue);
+    }
+
     public AMQP.Queue.DeleteOk queueDelete(String queue) throws IOException {
         return queueDelete(queue, false, false);
     }

test/src/com/rabbitmq/client/test/functional/ConsumerCount.java

@@ -0,0 +1,20 @@
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.test.BrokerTestCase;
+
+import java.io.IOException;
+
+public class ConsumerCount extends BrokerTestCase {
+    public void testConsumerCount() throws IOException {
+        String q = generateQueueName();
+        channel.queueDeclare(q, false, true, false, null);
+        assertEquals(0, channel.consumerCount(q));
+
+        String tag = channel.basicConsume(q, new DefaultConsumer(channel));
+        assertEquals(1, channel.consumerCount(q));
+
+        channel.basicCancel(tag);
+        assertEquals(0, channel.consumerCount(q));
+    }
+}

test/src/com/rabbitmq/client/test/functional/MessageCount.java

@@ -7,7 +7,7 @@
 public class MessageCount extends BrokerTestCase {
     public void testMessageCount() throws IOException {
         String q = generateQueueName();
-        channel.queueDeclareNoWait(q, false, true, true, null);
+        channel.queueDeclare(q, false, true, false, null);
         assertEquals(0, channel.messageCount(q));
 
         basicPublishVolatile(q);