moved return listener setup to Producer's constructor

Author: Alexandru Scvortov

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

@@ -90,7 +90,7 @@ public static void main(String[] args) {
 
             //setup
             String id = UUID.randomUUID().toString();
-            Stats stats = new Stats(1000L * samplingInterval);            
+            Stats stats = new Stats(1000L * samplingInterval);
             ConnectionFactory factory = new ConnectionFactory();
             factory.setHost(hostName);
             factory.setPort(portNumber);
@@ -127,22 +127,10 @@ public static void main(String[] args) {
                 Channel channel = conn.createChannel();
                 if (producerTxSize > 0) channel.txSelect();
                 channel.exchangeDeclare(exchangeName, exchangeType);
-                final Producer p = new Producer(channel, exchangeName, id,
-                                          flags, producerTxSize,
-                                          1000L * samplingInterval,
-                                          rateLimit, minMsgSize, timeLimit);
-                channel.setReturnListener(new ReturnListener() {
-                        public void handleBasicReturn(int replyCode,
-                                                      String replyText,
-                                                      String exchange,
-                                                      String routingKey,
-                                                      AMQP.BasicProperties properties,
-                                                      byte[] body)
-                            throws IOException {
-                            p.logBasicReturn();
-                        }
-                    });
-                Thread t = new Thread(p);
+                Thread t = new Thread(new Producer(channel, exchangeName, id,
+                                                   flags, producerTxSize,
+                                                   1000L * samplingInterval,
+                                                   rateLimit, minMsgSize, timeLimit));
                 producerThreads[i] = t;
                 t.start();
             }
@@ -245,6 +233,18 @@ 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 synchronized void logBasicReturn() {