Add a shorter form of Channel.basicConsume() which takes args, and use it in a few places where it makes sense.

Author: Simon MacMullen

src/com/rabbitmq/client/Channel.java

@@ -622,6 +622,23 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      */
     String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;
 
+    /**
+     * Start a non-nolocal, non-exclusive consumer, with
+     * a server-generated consumerTag and specified arguments.
+     * @param queue the name of the queue
+     * @param autoAck true if the server should consider messages
+     * acknowledged once delivered; false if the server should expect
+     * explicit acknowledgements
+     * @param arguments a set of arguments for the consume
+     * @param callback an interface to the consumer object
+     * @return the consumerTag generated by the server
+     * @throws java.io.IOException if an error is encountered
+     * @see com.rabbitmq.client.AMQP.Basic.Consume
+     * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
+     * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
+     */
+    String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments, Consumer callback) throws IOException;
+
     /**
      * Start a non-nolocal, non-exclusive consumer.
      * @param queue the name of the queue

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

@@ -935,6 +935,14 @@ public String basicConsume(String queue, boolean autoAck, Consumer callback)
         return basicConsume(queue, autoAck, "", callback);
     }
 
+    /** Public API - {@inheritDoc} */
+    public String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments,
+                               Consumer callback)
+        throws IOException
+    {
+        return basicConsume(queue, autoAck, "", false, false, arguments, callback);
+    }
+
     /** Public API - {@inheritDoc} */
     public String basicConsume(String queue, boolean autoAck, String consumerTag,
                                Consumer callback)

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

@@ -23,7 +23,7 @@ private void assertFailValidation(Map<String, Object> args) throws IOException {
         Channel ch = connection.createChannel();
         String queue = ch.queueDeclare().getQueue();
         try {
-            ch.basicConsume(queue, true, "", false, false, args, new QueueingConsumer(ch));
+            ch.basicConsume(queue, true, args, new QueueingConsumer(ch));
             fail("Validation should fail for " + args);
         } catch (IOException ioe) {
             checkShutdownSignal(AMQP.PRECONDITION_FAILED, ioe);
@@ -37,14 +37,14 @@ public void testConsumerPriorities() throws Exception {
         QueueingConsumer highConsumer = new QueueingConsumer(channel);
         QueueingConsumer medConsumer = new QueueingConsumer(channel);
         QueueingConsumer lowConsumer = new QueueingConsumer(channel);
-        channel.basicConsume(queue, true, "high", false, false, args(1), highConsumer);
-        channel.basicConsume(queue, true, "med", false, false, null, medConsumer);
-        channel.basicConsume(queue, true, "low", false, false, args(-1), lowConsumer);
+        String high = channel.basicConsume(queue, true, args(1), highConsumer);
+        String med = channel.basicConsume(queue, true, medConsumer);
+        channel.basicConsume(queue, true, args(-1), lowConsumer);
 
         publish(queue, COUNT, "high");
-        channel.basicCancel("high");
+        channel.basicCancel(high);
         publish(queue, COUNT, "med");
-        channel.basicCancel("med");
+        channel.basicCancel(med);
         publish(queue, COUNT, "low");
 
         assertContents(highConsumer, COUNT, "high");

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

@@ -109,7 +109,7 @@ private void publish(String q, int n) throws IOException {
     }
 
     private void consume(QueueingConsumer c, int prefetch, boolean autoAck) throws IOException {
-        channel.basicConsume(q, autoAck, "", false, false, args(prefetch), c);
+        channel.basicConsume(q, autoAck, args(prefetch), c);
     }
 
     private void ack(Delivery del, boolean multi) throws IOException {

test/src/com/rabbitmq/examples/perf/Consumer.java

@@ -66,7 +66,7 @@ public void run() {
 
         try {
             q = new QueueingConsumer(channel);
-            channel.basicConsume(queueName, autoAck, "", false, false, args(prefetch), q);
+            channel.basicConsume(queueName, autoAck, args(prefetch), q);
 
             while ((timeLimit == 0 || now < startTime + timeLimit) &&
                    (msgLimit == 0 || totalMsgCount < msgLimit)) {
@@ -81,7 +81,7 @@ public void run() {
                 } catch (ConsumerCancelledException e) {
                     System.out.println("Consumer cancelled by broker. Re-consuming.");
                     q = new QueueingConsumer(channel);
-                    channel.basicConsume(queueName, autoAck, "", false, false, args(prefetch), q);
+                    channel.basicConsume(queueName, autoAck, args(prefetch), q);
                     continue;
                 }
                 totalMsgCount++;