Merge pull request rabbitmq#48 from rabbitmq/rabbitmq-java-client-41

Introduce Channel#messageCount

Author: michaelklishin

src/com/rabbitmq/client/Channel.java

@@ -899,4 +899,14 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      * @throws IOException Problem transmitting method.
      */
     Command rpc(Method method) throws IOException;
+
+    /**
+     * Returns the number of messages in a queue ready to be delivered
+     * to consumers. 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 messages in ready state
+     * @throws IOException Problem transmitting method.
+     */
+    long messageCount(String queue) throws IOException;
 }

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

@@ -649,12 +649,12 @@ public void basicPublish(String exchange, String routingKey,
             useProps = MessageProperties.MINIMAL_BASIC;
         }
         transmit(new AMQCommand(new Basic.Publish.Builder()
-                                    .exchange(exchange)
-                                    .routingKey(routingKey)
-                                    .mandatory(mandatory)
-                                    .immediate(immediate)
-                                .build(),
-                                useProps, body));
+                                        .exchange(exchange)
+                                        .routingKey(routingKey)
+                                        .mandatory(mandatory)
+                                        .immediate(immediate)
+                                        .build(),
+                                       useProps, body));
     }
 
     /** Public API - {@inheritDoc} */
@@ -883,6 +883,12 @@ public Queue.DeclareOk queueDeclarePassive(String queue)
                .getMethod();
     }
 
+    /** Public API - {@inheritDoc} */
+    public long messageCount(String queue) throws IOException {
+        Queue.DeclareOk ok = queueDeclarePassive(queue);
+        return ok.getMessageCount();
+    }
+
     /** Public API - {@inheritDoc} */
     public Queue.DeleteOk queueDelete(String queue, boolean ifUnused, boolean ifEmpty)
         throws IOException

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

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

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

@@ -0,0 +1,21 @@
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.test.BrokerTestCase;
+
+import java.io.IOException;
+
+public class MessageCount extends BrokerTestCase {
+    public void testMessageCount() throws IOException {
+        String q = generateQueueName();
+        channel.queueDeclareNoWait(q, false, true, true, null);
+        assertEquals(0, channel.messageCount(q));
+
+        basicPublishVolatile(q);
+        assertEquals(1, channel.messageCount(q));
+        basicPublishVolatile(q);
+        assertEquals(2, channel.messageCount(q));
+
+        channel.queuePurge(q);
+        assertEquals(0, channel.messageCount(q));
+    }
+}