Refactored MulticastMain

Author: Emile Joubert

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

@@ -127,10 +127,12 @@ public static void main(String[] args) {
                 Channel channel = conn.createChannel();
                 if (producerTxSize > 0) channel.txSelect();
                 channel.exchangeDeclare(exchangeName, exchangeType);
-                Thread t = new Thread(new Producer(channel, exchangeName, id,
+                final Producer p = new Producer(channel, exchangeName, id,
                                                    flags, producerTxSize,
                                                    1000L * samplingInterval,
-                                                   rateLimit, minMsgSize, timeLimit));
+                                                   rateLimit, minMsgSize, timeLimit);
+                channel.setReturnListener(p);                
+                Thread t = new Thread(p);
                 producerThreads[i] = t;
                 t.start();
             }
@@ -197,7 +199,7 @@ private static List lstArg(CommandLine cmd, char opt) {
         return Arrays.asList(vals);
     }
 
-    public static class Producer implements Runnable {
+    public static class Producer implements Runnable, ReturnListener {
 
         private Channel channel;
         private String  exchangeName;
@@ -233,18 +235,15 @@ public Producer(Channel channel, String exchangeName, String id,
             this.rateLimit    = rateLimit;
             this.timeLimit    = 1000L * timeLimit;
             this.message      = new byte[minMsgSize];
+        }
 
-            channel.setReturnListener(new ReturnListener() {
-                    public void handleBasicReturn(int replyCode,
-                                                  String replyText,
-                                                  String exchange,
-                                                  String routingKey,
-                                                  AMQP.BasicProperties properties,
-                                                  byte[] body)
-                        throws IOException {
-                        logBasicReturn();
-                    }
-                });
+        public void handleBasicReturn(int replyCode,
+                                      String replyText,
+                                      String exchange,
+                                      String routingKey,
+                                      AMQP.BasicProperties properties,
+                                      byte[] body) throws IOException {
+            logBasicReturn();
         }
 
         public synchronized void logBasicReturn() {