Remove the extra convenience method for basic.consume which is no longer especially convenient, and fix the interaction of recovering channels with new basic.qos.

Author: Simon MacMullen

src/com/rabbitmq/client/Channel.java

@@ -628,23 +628,6 @@ 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

@@ -946,14 +946,6 @@ 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)

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

@@ -34,8 +34,8 @@ public class AutorecoveringChannel implements Channel, Recoverable {
     private List<ReturnListener> returnListeners = new ArrayList<ReturnListener>();
     private List<ConfirmListener> confirmListeners = new ArrayList<ConfirmListener>();
     private List<FlowListener> flowListeners = new ArrayList<FlowListener>();
-    private int prefetchCount;
-    private boolean globalQos;
+    private int prefetchCountConsumer;
+    private int prefetchCountGlobal;
     private boolean usesPublisherConfirms;
     private boolean usesTransactions;
 
@@ -138,16 +138,23 @@ public void setDefaultConsumer(Consumer consumer) {
     }
 
     public void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException {
-        this.prefetchCount = prefetchCount;
-        this.globalQos = global;
+        if (global) {
+            this.prefetchCountGlobal = prefetchCount;
+        } else {
+            this.prefetchCountConsumer = prefetchCount;
+        }
+
         delegate.basicQos(prefetchSize, prefetchCount, global);
     }
 
     public void basicQos(int prefetchCount) throws IOException {
-
         basicQos(0, prefetchCount, false);
     }
 
+    public void basicQos(int prefetchCount, boolean global) throws IOException {
+        basicQos(0, prefetchCount, global);
+    }
+
     public void basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body) throws IOException {
         delegate.basicPublish(exchange, routingKey, props, body);
     }
@@ -432,7 +439,12 @@ private void recoverFlowListeners() {
     }
 
     private void recoverState() throws IOException {
-        basicQos(0, this.prefetchCount, this.globalQos);
+        if (this.prefetchCountConsumer != 0) {
+            basicQos(this.prefetchCountConsumer, false);
+        }
+        if (this.prefetchCountGlobal != 0) {
+            basicQos(this.prefetchCountGlobal, true);
+        }
         if(this.usesPublisherConfirms) {
             this.confirmSelect();
         }

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, args, new QueueingConsumer(ch));
+            ch.basicConsume(queue, true, "", false, false, 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);
-        String high = channel.basicConsume(queue, true, args(1), highConsumer);
-        String med = channel.basicConsume(queue, true, medConsumer);
-        channel.basicConsume(queue, true, args(-1), lowConsumer);
+        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);
 
         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");