stable to default

Author: Simon MacMullen

test/src/com/rabbitmq/client/test/server/LoopbackUsers.java

@@ -71,20 +71,19 @@ private ConnectionFactory getFactory(String name, String addr) {
 
     // Find the first IP address of a network interface that is up, not loopback, not point to point (e.g. VPN thing)
     private static InetAddress findRealIPAddress() throws SocketException {
-        throw new RuntimeException("this test will be enabled once we stop supporting Java 1.5");
-//        Enumeration<NetworkInterface> ifs = NetworkInterface.getNetworkInterfaces();
-//        while (ifs.hasMoreElements()) {
-//            NetworkInterface nif = ifs.nextElement();
-//            if (nif.isUp() && !nif.isPointToPoint() && !nif.isLoopback() && !nif.isVirtual()) {
-//                Enumeration<InetAddress> addrs = nif.getInetAddresses();
-//                while (addrs.hasMoreElements()) {
-//                    InetAddress addr = addrs.nextElement();
-//                    if (addr instanceof Inet4Address) {
-//                        return addr;
-//                    }
-//                }
-//            }
-//        }
-//        throw new RuntimeException("Could not determine real network address");
+        Enumeration<NetworkInterface> ifs = NetworkInterface.getNetworkInterfaces();
+        while (ifs.hasMoreElements()) {
+            NetworkInterface nif = ifs.nextElement();
+            if (nif.isUp() && !nif.isPointToPoint() && !nif.isLoopback() && !nif.isVirtual()) {
+                Enumeration<InetAddress> addrs = nif.getInetAddresses();
+                while (addrs.hasMoreElements()) {
+                    InetAddress addr = addrs.nextElement();
+                    if (addr instanceof Inet4Address) {
+                        return addr;
+                    }
+                }
+            }
+        }
+        throw new RuntimeException("Could not determine real network address");
     }
 }

test/src/com/rabbitmq/client/test/server/ServerTests.java

@@ -42,6 +42,6 @@ public static void add(TestSuite suite) {
         suite.addTestSuite(Shutdown.class);
         suite.addTestSuite(BlockedConnection.class);
         suite.addTestSuite(ChannelLimitNegotiation.class);
-        //suite.addTestSuite(LoopbackUsers.class);
+        suite.addTestSuite(LoopbackUsers.class);
     }
 }

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

@@ -45,31 +45,32 @@ public static void main(String[] args) {
                 System.exit(0);
             }
 
-            String exchangeType  = strArg(cmd, 't', "direct");
-            String exchangeName  = strArg(cmd, 'e', exchangeType);
-            String queueName     = strArg(cmd, 'u', "");
-            String routingKey    = strArg(cmd, 'k', null);
-            int samplingInterval = intArg(cmd, 'i', 1);
-            float rateLimit      = floatArg(cmd, 'r', 0.0f);
-            int producerCount    = intArg(cmd, 'x', 1);
-            int consumerCount    = intArg(cmd, 'y', 1);
-            int producerTxSize   = intArg(cmd, 'm', 0);
-            int consumerTxSize   = intArg(cmd, 'n', 0);
-            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, 'q', 0);
-            int minMsgSize       = intArg(cmd, 's', 0);
-            int timeLimit        = intArg(cmd, 'z', 0);
-            int producerMsgCount = intArg(cmd, 'C', 0);
-            int consumerMsgCount = intArg(cmd, 'D', 0);
-            List<?> flags        = lstArg(cmd, 'f');
-            int frameMax         = intArg(cmd, 'M', 0);
-            int heartbeat        = intArg(cmd, 'b', 0);
-            boolean predeclared  = cmd.hasOption('p');
-
-            String uri           = strArg(cmd, 'h', "amqp://localhost");
+            String exchangeType     = strArg(cmd, 't', "direct");
+            String exchangeName     = strArg(cmd, 'e', exchangeType);
+            String queueName        = strArg(cmd, 'u', "");
+            String routingKey       = strArg(cmd, 'k', null);
+            int samplingInterval    = intArg(cmd, 'i', 1);
+            float producerRateLimit = floatArg(cmd, 'r', 0.0f);
+            float consumerRateLimit = floatArg(cmd, 'R', 0.0f);
+            int producerCount       = intArg(cmd, 'x', 1);
+            int consumerCount       = intArg(cmd, 'y', 1);
+            int producerTxSize      = intArg(cmd, 'm', 0);
+            int consumerTxSize      = intArg(cmd, 'n', 0);
+            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, 'q', 0);
+            int minMsgSize          = intArg(cmd, 's', 0);
+            int timeLimit           = intArg(cmd, 'z', 0);
+            int producerMsgCount    = intArg(cmd, 'C', 0);
+            int consumerMsgCount    = intArg(cmd, 'D', 0);
+            List<?> flags           = lstArg(cmd, 'f');
+            int frameMax            = intArg(cmd, 'M', 0);
+            int heartbeat           = intArg(cmd, 'b', 0);
+            boolean predeclared     = cmd.hasOption('p');
+
+            String uri              = strArg(cmd, 'h', "amqp://localhost");
 
             //setup
             PrintlnStats stats = new PrintlnStats(1000L * samplingInterval,
@@ -91,6 +92,7 @@ public static void main(String[] args) {
             p.setConfirm(          confirm);
             p.setConsumerCount(    consumerCount);
             p.setConsumerMsgCount( consumerMsgCount);
+            p.setConsumerRateLimit(consumerRateLimit);
             p.setConsumerTxSize(   consumerTxSize);
             p.setExchangeName(     exchangeName);
             p.setExchangeType(     exchangeType);
@@ -105,7 +107,7 @@ public static void main(String[] args) {
             p.setProducerTxSize(   producerTxSize);
             p.setQueueName(        queueName);
             p.setRoutingKey(       routingKey);
-            p.setRateLimit(        rateLimit);
+            p.setProducerRateLimit(producerRateLimit);
             p.setTimeLimit(        timeLimit);
 
             MulticastSet set = new MulticastSet(stats, factory, p);
@@ -137,7 +139,8 @@ private static Options getOptions() {
         options.addOption(new Option("u", "queue",         true, "queue name"));
         options.addOption(new Option("k", "routingKey",    true, "routing key"));
         options.addOption(new Option("i", "interval",      true, "sampling interval"));
-        options.addOption(new Option("r", "rate",          true, "rate limit"));
+        options.addOption(new Option("r", "rate",          true, "producer rate limit"));
+        options.addOption(new Option("R", "consumerRate",  true, "consumer rate limit"));
         options.addOption(new Option("x", "producers",     true, "producer count"));
         options.addOption(new Option("y", "consumers",     true, "consumer count"));
         options.addOption(new Option("m", "ptxsize",       true, "producer tx size"));

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

@@ -16,20 +16,21 @@
 
 package com.rabbitmq.examples.perf;
 
+import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.ConsumerCancelledException;
+import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
-import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.ShutdownSignalException;
 
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
 import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
 
-public class Consumer implements Runnable {
+public class Consumer extends ProducerConsumerBase implements Runnable {
 
-    private QueueingConsumer q;
-    private Channel channel;
+    private ConsumerImpl     q;
+    private Channel          channel;
     private String           id;
     private String           queueName;
     private int              txSize;
@@ -38,14 +39,16 @@ public class Consumer implements Runnable {
     private Stats stats;
     private int              msgLimit;
     private long             timeLimit;
+    private CountDownLatch   latch = new CountDownLatch(1);
 
     public Consumer(Channel channel, String id,
                     String queueName, int txSize, boolean autoAck,
-                    int multiAckEvery, Stats stats, int msgLimit, int timeLimit) {
+                    int multiAckEvery, Stats stats, float rateLimit, int msgLimit, int timeLimit) {
 
         this.channel       = channel;
         this.id            = id;
         this.queueName     = queueName;
+        this.rateLimit     = rateLimit;
         this.txSize        = txSize;
         this.autoAck       = autoAck;
         this.multiAckEvery = multiAckEvery;
@@ -55,40 +58,44 @@ public Consumer(Channel channel, String id,
     }
 
     public void run() {
+        try {
+            q = new ConsumerImpl(channel);
+            channel.basicConsume(queueName, autoAck, q);
+            latch.await();
+
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        } catch (ShutdownSignalException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private class ConsumerImpl extends DefaultConsumer {
         long now;
         long startTime;
-        startTime = now = System.currentTimeMillis();
         int totalMsgCount = 0;
 
-        try {
-            q = new QueueingConsumer(channel);
-            channel.basicConsume(queueName, autoAck, q);
+        private ConsumerImpl(Channel channel) {
+            super(channel);
+            startTime = now = System.currentTimeMillis();
+            lastStatsTime = startTime;
+            msgCount = 0;
+        }
 
-            while ((timeLimit == 0 || now < startTime + timeLimit) &&
-                   (msgLimit == 0 || totalMsgCount < msgLimit)) {
-                QueueingConsumer.Delivery delivery;
-                try {
-                    if (timeLimit == 0) {
-                        delivery = q.nextDelivery();
-                    } else {
-                        delivery = q.nextDelivery(startTime + timeLimit - now);
-                        if (delivery == null) break;
-                    }
-                } catch (ConsumerCancelledException e) {
-                    System.out.println("Consumer cancelled by broker. Re-consuming.");
-                    q = new QueueingConsumer(channel);
-                    channel.basicConsume(queueName, autoAck, q);
-                    continue;
-                }
+        @Override
+        public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
+            if ((timeLimit == 0 || now < startTime + timeLimit) &&
+                (msgLimit == 0 || msgCount < msgLimit)) {
                 totalMsgCount++;
+                msgCount++;
 
-                DataInputStream d = new DataInputStream(new ByteArrayInputStream(delivery.getBody()));
+                DataInputStream d = new DataInputStream(new ByteArrayInputStream(body));
                 d.readInt();
                 long msgNano = d.readLong();
                 long nano = System.nanoTime();
 
-                Envelope envelope = delivery.getEnvelope();
-
                 if (!autoAck) {
                     if (multiAckEvery == 0) {
                         channel.basicAck(envelope.getDeliveryTag(), false);
@@ -104,14 +111,17 @@ public void run() {
                 now = System.currentTimeMillis();
 
                 stats.handleRecv(id.equals(envelope.getRoutingKey()) ? (nano - msgNano) : 0L);
+                delay(now);
+            }
+            else {
+                latch.countDown();
             }
+        }
 
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        } catch (InterruptedException e) {
-            throw new RuntimeException (e);
-        } catch (ShutdownSignalException e) {
-            throw new RuntimeException(e);
+        @Override
+        public void handleCancel(String consumerTag) throws IOException {
+            System.out.println("Consumer cancelled by broker. Re-consuming.");
+            channel.basicConsume(queueName, autoAck, q);
         }
     }
 }

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

@@ -18,7 +18,6 @@
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ShutdownSignalException;
 
@@ -37,7 +36,8 @@ public class MulticastParams {
     private int minMsgSize = 0;
 
     private int timeLimit = 0;
-    private float rateLimit = 0;
+    private float producerRateLimit = 0;
+    private float consumerRateLimit = 0;
     private int producerMsgCount = 0;
     private int consumerMsgCount = 0;
 
@@ -70,14 +70,18 @@ public void setRoutingKey(String routingKey) {
         this.routingKey = routingKey;
     }
 
-    public void setRateLimit(float rateLimit) {
-        this.rateLimit = rateLimit;
+    public void setProducerRateLimit(float producerRateLimit) {
+        this.producerRateLimit = producerRateLimit;
     }
 
     public void setProducerCount(int producerCount) {
         this.producerCount = producerCount;
     }
 
+    public void setConsumerRateLimit(float consumerRateLimit) {
+        this.consumerRateLimit = consumerRateLimit;
+    }
+
     public void setConsumerCount(int consumerCount) {
         this.consumerCount = consumerCount;
     }
@@ -168,7 +172,7 @@ public Producer createProducer(Connection connection, Stats stats, String id) th
         }
         final Producer producer = new Producer(channel, exchangeName, id,
                                                flags, producerTxSize,
-                                               rateLimit, producerMsgCount,
+                                               producerRateLimit, producerMsgCount,
                                                minMsgSize, timeLimit,
                                                confirm, stats);
         channel.addReturnListener(producer);
@@ -184,7 +188,7 @@ public Consumer createConsumer(Connection connection, Stats stats, String id) th
         if (channelPrefetch > 0) channel.basicQos(channelPrefetch, true);
         return new Consumer(channel, id, qName,
                                          consumerTxSize, autoAck, multiAckEvery,
-                                         stats, consumerMsgCount, timeLimit);
+                                         stats, consumerRateLimit, consumerMsgCount, timeLimit);
     }
 
     public boolean shouldConfigureQueue() {

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

@@ -31,7 +31,7 @@
 import java.util.TreeSet;
 import java.util.concurrent.Semaphore;
 
-public class Producer implements Runnable, ReturnListener,
+public class Producer extends ProducerConsumerBase implements Runnable, ReturnListener,
         ConfirmListener
 {
     private Channel channel;
@@ -41,18 +41,13 @@ public class Producer implements Runnable, ReturnListener,
     private boolean immediate;
     private boolean persistent;
     private int     txSize;
-    private float   rateLimit;
     private int     msgLimit;
     private long    timeLimit;
 
     private Stats   stats;
 
     private byte[]  message;
 
-    private long    startTime;
-    private long    lastStatsTime;
-    private int     msgCount;
-
     private Semaphore confirmPool;
     private volatile SortedSet<Long> unconfirmedSet =
         Collections.synchronizedSortedSet(new TreeSet<Long>());
@@ -124,7 +119,10 @@ private void handleAckNack(long seqNo, boolean multiple,
     }
 
     public void run() {
-        long now = startTime = lastStatsTime = System.currentTimeMillis();
+        long now;
+        long startTime;
+        startTime = now = System.currentTimeMillis();
+        lastStatsTime = startTime;
         msgCount = 0;
         int totalMsgCount = 0;
 
@@ -164,21 +162,6 @@ private void publish(byte[] msg)
                              msg);
     }
 
-    private void delay(long now)
-        throws InterruptedException {
-
-        long elapsed = now - lastStatsTime;
-        //example: rateLimit is 5000 msg/s,
-        //10 ms have elapsed, we have sent 200 messages
-        //the 200 msgs we have actually sent should have taken us
-        //200 * 1000 / 5000 = 40 ms. So we pause for 40ms - 10ms
-        long pause = (long) (rateLimit == 0.0f ?
-            0.0f : (msgCount * 1000.0 / rateLimit - elapsed));
-        if (pause > 0) {
-            Thread.sleep(pause);
-        }
-    }
-
     private byte[] createMessage(int sequenceNumber)
         throws IOException {