Update PerfTest for new qos regime, and add a convenience method for per-channel prefetch.

Author: Simon MacMullen

src/com/rabbitmq/client/Channel.java

@@ -222,11 +222,24 @@ public interface Channel extends ShutdownNotifier {
      * @param prefetchCount maximum number of messages that the server
      * will deliver, 0 if unlimited
      * @param global true if the settings should be applied to the
-     * entire connection rather than just the current channel
+     * entire channel rather than each consumer
      * @throws java.io.IOException if an error is encountered
      */
     void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;
 
+    /**
+     * Request a specific prefetchCount "quality of service" settings
+     * for this channel.
+     *
+     * @see #basicQos(int, int, boolean)
+     * @param prefetchCount maximum number of messages that the server
+     * will deliver, 0 if unlimited
+     * @param global true if the settings should be applied to the
+     * entire channel rather than each consumer
+     * @throws java.io.IOException if an error is encountered
+     */
+    void basicQos(int prefetchCount, boolean global) throws IOException;
+
     /**
      * Request a specific prefetchCount "quality of service" settings
      * for this channel.

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

@@ -595,6 +595,13 @@ public void basicQos(int prefetchSize, int prefetchCount, boolean global)
 	exnWrappingRpc(new Basic.Qos(prefetchSize, prefetchCount, global));
     }
 
+    /** Public API - {@inheritDoc} */
+    public void basicQos(int prefetchCount, boolean global)
+            throws IOException
+    {
+        basicQos(0, prefetchCount, global);
+    }
+
     /** Public API - {@inheritDoc} */
     public void basicQos(int prefetchCount)
 	throws IOException

test/src/com/rabbitmq/examples/PerfTest.java

@@ -58,8 +58,8 @@ public static void main(String[] args) {
             long confirm         = intArg(cmd, 'c', -1);
             boolean autoAck      = cmd.hasOption('a');
             int multiAckEvery    = intArg(cmd, 'A', 0);
-            int channelPrefetch  = intArg(cmd, 'q', 0);
-            int consumerPrefetch = intArg(cmd, 'P', 0);
+            int channelPrefetch  = intArg(cmd, 'G', 0);
+            int consumerPrefetch = intArg(cmd, 'q', 0);
             int minMsgSize       = intArg(cmd, 's', 0);
             int timeLimit        = intArg(cmd, 'z', 0);
             int producerMsgCount = intArg(cmd, 'C', 0);
@@ -148,8 +148,8 @@ private static Options getOptions() {
         options.addOption(new Option("c", "confirm",       true, "max unconfirmed publishes"));
         options.addOption(new Option("a", "autoack",       false,"auto ack"));
         options.addOption(new Option("A", "multiAckEvery", true, "multi ack every"));
-        options.addOption(new Option("q", "qos",           true, "qos (channel) prefetch count"));
-        options.addOption(new Option("P", "prefetch",      true, "consumer prefetch count"));
+        options.addOption(new Option("q", "qos",           true, "consumer prefetch count"));
+        options.addOption(new Option("G", "globalQos",     true, "channel prefetch count"));
         options.addOption(new Option("s", "size",          true, "message size"));
         options.addOption(new Option("z", "time",          true, "time limit"));
         options.addOption(new Option("C", "pmessages", true, "producer message count"));

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

@@ -25,16 +25,13 @@
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
 
 public class Consumer implements Runnable {
 
     private QueueingConsumer q;
     private Channel channel;
     private String           id;
     private String           queueName;
-    private int              prefetch;
     private int              txSize;
     private boolean          autoAck;
     private int              multiAckEvery;
@@ -43,12 +40,11 @@ public class Consumer implements Runnable {
     private long             timeLimit;
 
     public Consumer(Channel channel, String id,
-                    String queueName, int prefetch, int txSize, boolean autoAck,
+                    String queueName, int txSize, boolean autoAck,
                     int multiAckEvery, Stats stats, int msgLimit, int timeLimit) {
 
         this.channel       = channel;
         this.id            = id;
-        this.prefetch      = prefetch;
         this.queueName     = queueName;
         this.txSize        = txSize;
         this.autoAck       = autoAck;
@@ -66,7 +62,7 @@ public void run() {
 
         try {
             q = new QueueingConsumer(channel);
-            channel.basicConsume(queueName, autoAck, args(prefetch), q);
+            channel.basicConsume(queueName, autoAck, q);
 
             while ((timeLimit == 0 || now < startTime + timeLimit) &&
                    (msgLimit == 0 || totalMsgCount < msgLimit)) {
@@ -81,7 +77,7 @@ public void run() {
                 } catch (ConsumerCancelledException e) {
                     System.out.println("Consumer cancelled by broker. Re-consuming.");
                     q = new QueueingConsumer(channel);
-                    channel.basicConsume(queueName, autoAck, args(prefetch), q);
+                    channel.basicConsume(queueName, autoAck, q);
                     continue;
                 }
                 totalMsgCount++;
@@ -118,12 +114,4 @@ public void run() {
             throw new RuntimeException(e);
         }
     }
-
-    private Map<String, Object> args(int prefetch) {
-        Map<String, Object> a = new HashMap<String, Object>();
-        if (prefetch != 0) {
-            a.put("x-prefetch", prefetch);
-        }
-        return a;
-    }
 }

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

@@ -185,8 +185,9 @@ public Consumer createConsumer(Connection connection, Stats stats, String id) th
         Channel channel = connection.createChannel();
         if (consumerTxSize > 0) channel.txSelect();
         String qName = configureQueue(connection, id);
-        if (channelPrefetch > 0) channel.basicQos(channelPrefetch);
-        return new Consumer(channel, id, qName, consumerPrefetch,
+        if (consumerPrefetch > 0) channel.basicQos(consumerPrefetch);
+        if (channelPrefetch > 0) channel.basicQos(channelPrefetch, true);
+        return new Consumer(channel, id, qName,
                                          consumerTxSize, autoAck, multiAckEvery,
                                          stats, consumerMsgCount, timeLimit);
     }