Make MulticastParams create producers and consumers, and thus prevent knowledge of its entire state leaking into MulticastSet.

Author: Simon MacMullen

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

@@ -16,33 +16,36 @@
 
 package com.rabbitmq.examples.perf;
 
+import com.rabbitmq.client.Channel;
+
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
 public class MulticastParams {
-    protected long confirm = -1;
-    protected int consumerCount = 1;
-    protected int producerCount = 1;
-    protected int consumerTxSize = 0;
-    protected int producerTxSize = 0;
-    protected int prefetchCount = 0;
-    protected int minMsgSize = 0;
-
-    protected int timeLimit = 0;
-    protected int rateLimit = 0;
-    protected int producerMsgCount = 0;
-    protected int consumerMsgCount = 0;
-
-    protected String exchangeName = "direct";
-    protected String exchangeType = "direct";
-    protected String queueName = "";
-
-    protected List<?> flags = new ArrayList<Object>();
-
-    protected int multiAckEvery = 0;
-    protected boolean autoAck = true;
-    protected boolean exclusive = true;
-    protected boolean autoDelete = false;
+    private long confirm = -1;
+    private int consumerCount = 1;
+    private int producerCount = 1;
+    private int consumerTxSize = 0;
+    private int producerTxSize = 0;
+    private int prefetchCount = 0;
+    private int minMsgSize = 0;
+
+    private int timeLimit = 0;
+    private int rateLimit = 0;
+    private int producerMsgCount = 0;
+    private int consumerMsgCount = 0;
+
+    private String exchangeName = "direct";
+    private String exchangeType = "direct";
+    private String queueName = "";
+
+    private List<?> flags = new ArrayList<Object>();
+
+    private int multiAckEvery = 0;
+    private boolean autoAck = true;
+    private boolean exclusive = true;
+    private boolean autoDelete = false;
 
     public void setExchangeType(String exchangeType) {
         this.exchangeType = exchangeType;
@@ -124,4 +127,58 @@ public void setExclusive(boolean exclusive) {
     public void setAutoDelete(boolean autoDelete) {
         this.autoDelete = autoDelete;
     }
+
+    public int getConsumerCount() {
+        return consumerCount;
+    }
+
+    public int getProducerCount() {
+        return producerCount;
+    }
+
+    public int getMinMsgSize() {
+        return minMsgSize;
+    }
+
+    public Producer createProducer(Channel channel, Stats stats, String id) throws IOException {
+        if (producerTxSize > 0) channel.txSelect();
+        if (confirm >= 0) channel.confirmSelect();
+        channel.exchangeDeclare(exchangeName, exchangeType);
+        final Producer producer = new Producer(channel, exchangeName, id,
+                                               flags, producerTxSize,
+                                               rateLimit, producerMsgCount,
+                                               minMsgSize, timeLimit,
+                                               confirm, stats);
+        channel.addReturnListener(producer);
+        channel.addConfirmListener(producer);
+        return producer;
+    }
+
+    public Consumer createConsumer(Channel channel, Stats stats, String id) throws IOException {
+        if (consumerTxSize > 0) channel.txSelect();
+        channel.exchangeDeclare(exchangeName, exchangeType);
+        String qName =
+                channel.queueDeclare(queueName,
+                                     flags.contains("persistent"),
+                                     exclusive, autoDelete,
+                                     null).getQueue();
+        if (prefetchCount > 0) channel.basicQos(prefetchCount);
+        channel.queueBind(qName, exchangeName, id);
+        return new Consumer(channel, id, qName,
+                                         consumerTxSize, autoAck, multiAckEvery,
+                                         stats, consumerMsgCount, timeLimit);
+    }
+
+    public boolean shouldConfigureQueue() {
+        return consumerCount == 0 && !queueName.equals("");
+    }
+
+    public void configureQueue(Channel channel, String id) throws IOException {
+        channel.exchangeDeclare(exchangeName, exchangeType);
+        channel.queueDeclare(queueName,
+                             flags.contains("persistent"),
+                             exclusive, autoDelete,
+                             null).getQueue();
+        channel.queueBind(queueName, exchangeName, id);
+    }
 }

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

@@ -27,99 +27,72 @@ public class MulticastSet {
     private final String id;
     private final Stats stats;
     private final ConnectionFactory factory;
-    private final MulticastParams p;
+    private final MulticastParams params;
 
     public MulticastSet(Stats stats, ConnectionFactory factory,
                         MulticastParams params) {
         this.id = UUID.randomUUID().toString();
         this.stats = stats;
         this.factory = factory;
-        this.p = params;
+        this.params = params;
     }
 
     public void run() throws IOException, InterruptedException {
         run(false);
     }
 
     public void run(boolean announceStartup) throws IOException, InterruptedException {
-        Thread[] consumerThreads = new Thread[p.consumerCount];
-        Connection[] consumerConnections = new Connection[p.consumerCount];
-        for (int i = 0; i < p.consumerCount; i++) {
+        Thread[] consumerThreads = new Thread[params.getConsumerCount()];
+        Connection[] consumerConnections = new Connection[consumerThreads.length];
+        for (int i = 0; i < consumerConnections.length; i++) {
             if (announceStartup) {
                 System.out.println("starting consumer #" + i);
             }
             Connection conn = factory.newConnection();
             consumerConnections[i] = conn;
             Channel channel = conn.createChannel();
-            if (p.consumerTxSize > 0) channel.txSelect();
-            channel.exchangeDeclare(p.exchangeName, p.exchangeType);
-            String qName =
-                    channel.queueDeclare(p.queueName,
-                                         p.flags.contains("persistent"),
-                                         p.exclusive, p.autoDelete,
-                                         null).getQueue();
-            if (p.prefetchCount > 0) channel.basicQos(p.prefetchCount);
-            channel.queueBind(qName, p.exchangeName, id);
-            Thread t =
-                new Thread(new Consumer(channel, id, qName,
-                                        p.consumerTxSize, p.autoAck, p.multiAckEvery,
-                                        stats, p.consumerMsgCount, p.timeLimit));
+            Thread t = new Thread(params.createConsumer(channel, stats, id));
             consumerThreads[i] = t;
         }
 
-        if (p.consumerCount == 0 && !p.queueName.equals("")) {
+        if (params.shouldConfigureQueue()) {
             Connection conn = factory.newConnection();
             Channel channel = conn.createChannel();
-            channel.exchangeDeclare(p.exchangeName, p.exchangeType);
-            channel.queueDeclare(p.queueName,
-                                 p.flags.contains("persistent"),
-                                 p.exclusive, p.autoDelete,
-                                 null).getQueue();
-            channel.queueBind(p.queueName, p.exchangeName, id);
+            params.configureQueue(channel, id);
             conn.close();
         }
 
-        Thread[] producerThreads = new Thread[p.producerCount];
-        Connection[] producerConnections = new Connection[p.producerCount];
-        Channel[] producerChannels = new Channel[p.producerCount];
-        for (int i = 0; i < p.producerCount; i++) {
+        Thread[] producerThreads = new Thread[params.getProducerCount()];
+        Connection[] producerConnections = new Connection[producerThreads.length];
+        Channel[] producerChannels = new Channel[producerConnections.length];
+        for (int i = 0; i < producerChannels.length; i++) {
             if (announceStartup) {
                 System.out.println("starting producer #" + i);
             }
             Connection conn = factory.newConnection();
             producerConnections[i] = conn;
             Channel channel = conn.createChannel();
             producerChannels[i] = channel;
-            if (p.producerTxSize > 0) channel.txSelect();
-            if (p.confirm >= 0) channel.confirmSelect();
-            channel.exchangeDeclare(p.exchangeName, p.exchangeType);
-            final Producer producer = new Producer(channel, p.exchangeName, id,
-                                                   p.flags, p.producerTxSize,
-                                                   p.rateLimit, p.producerMsgCount,
-                                                   p.minMsgSize, p.timeLimit,
-                                                   p.confirm, stats);
-            channel.addReturnListener(producer);
-            channel.addConfirmListener(producer);
-            Thread t = new Thread(producer);
+            Thread t = new Thread(params.createProducer(channel, stats, id));
             producerThreads[i] = t;
         }
 
-        for (int i = 0; i < p.consumerCount; i++) {
-            consumerThreads[i].start();
+        for (Thread consumerThread : consumerThreads) {
+            consumerThread.start();
         }
 
-        for (int i = 0; i < p.producerCount; i++) {
-            producerThreads[i].start();
+        for (Thread producerThread : producerThreads) {
+            producerThread.start();
         }
 
-        for (int i = 0; i < p.producerCount; i++) {
+        for (int i = 0; i < producerThreads.length; i++) {
             producerThreads[i].join();
             producerChannels[i].clearReturnListeners();
             producerChannels[i].clearConfirmListeners();
             producerConnections[i].close();
         }
 
-        for (int i = 0; i < p.consumerCount; i++) {
+        for (int i = 0; i < consumerThreads.length; i++) {
             consumerThreads[i].join();
             consumerConnections[i].close();
         }

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

@@ -19,7 +19,6 @@
 import com.rabbitmq.client.ConnectionFactory;
 
 import java.io.IOException;
-import java.util.Map;
 
 public class SimpleScenario implements Scenario {
     private String name;
@@ -44,7 +43,7 @@ public void run() throws IOException, InterruptedException {
         this.stats = new SimpleScenarioStats(interval);
         for (MulticastParams p : params) {
             MulticastSet set = new MulticastSet(stats, factory, p);
-            stats.setMinMsgSize(p.minMsgSize);
+            stats.setup(p);
             set.run();
         }
     }

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

@@ -70,8 +70,8 @@ public Map<String, Object> results() {
         return map;
     }
 
-    public void setMinMsgSize(long minMsgSize) {
-        this.minMsgSize = minMsgSize;
+    public void setup(MulticastParams params) {
+        this.minMsgSize = params.getMinMsgSize();
     }
 
     public double getSendRate() {

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

@@ -64,7 +64,7 @@ private void run(Variable[] variables, List<VariableValue> values) throws Except
                     value.setup(p);
                 }
                 MulticastSet set = new MulticastSet(stats0, factory, p);
-                stats0.setMinMsgSize(p.minMsgSize);
+                stats0.setup(p);
                 set.run();
                 for (VariableValue value : values) {
                     value.teardown(p);