Added log() calls to TestMain.tryTransaction

Author: Steve Powell

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

@@ -201,15 +201,15 @@ public static void sleep(int ms) {
         }
     }
 
-    public Connection _connection;
+    private Connection _connection;
 
-    public Channel _ch1;
+    private Channel _ch1;
 
-    public int _messageId = 0;
+    private int _messageId = 0;
 
-    private boolean _silent;
+    private final boolean _silent;
 
-    private BlockingCell<Object> returnCell;
+    private volatile BlockingCell<Object> returnCell;
 
     public TestMain(Connection connection, boolean silent) {
         _connection = connection;
@@ -230,15 +230,6 @@ public void run() throws IOException {
 
         _ch1 = createChannel();
 
-        _ch1.addReturnListener(new ReturnListener() {
-            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body)
-                    throws IOException {
-                Method method = new AMQImpl.Basic.Return(replyCode, replyText, exchange, routingKey);
-                log("Handling return with body " + new String(body));
-                returnCell.set(new Object[] { method, properties, body });
-            }
-        });
-
         String queueName =_ch1.queueDeclare().getQueue();
 
         sendLotsOfTrivialMessages(batchSize, queueName);
@@ -276,6 +267,18 @@ public void handleReturn(int replyCode, String replyText, String exchange, Strin
         log("Leaving TestMain.run().");
     }
 
+    private void setChannelReturnListener() {
+        log("Setting return listener..");
+        _ch1.addReturnListener(new ReturnListener() {
+            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body)
+                    throws IOException {
+                Method method = new AMQImpl.Basic.Return(replyCode, replyText, exchange, routingKey);
+                log("Handling return with body " + new String(body));
+                TestMain.this.returnCell.set(new Object[] { method, properties, body });
+            }
+        });
+    }
+
     public class UnexpectedSuccessException extends IOException {
         /**
          * Default version UID for serializable class
@@ -447,6 +450,8 @@ public void tryBasicReturn() throws IOException {
         String mx = "mandatoryTestExchange";
         _ch1.exchangeDeclare(mx, "fanout", false, true, null);
 
+        setChannelReturnListener();
+
         returnCell = new BlockingCell<Object>();
         _ch1.basicPublish(mx, "", true, false, null, "one".getBytes());
         doBasicReturn(returnCell, AMQP.NO_ROUTE);
@@ -472,7 +477,15 @@ public void tryBasicReturn() throws IOException {
         drain(1, mq, true);
         _ch1.queueDelete(mq, true, true);
 
+        unsetChannelReturnListener();
+
         log("Completed basic.return testing.");
+        
+    }
+
+    private void unsetChannelReturnListener() {
+        _ch1.clearReturnListeners();
+        log("ReturnListeners unset");
     }
 
     public void waitForKey(String prompt) throws IOException {
@@ -487,8 +500,12 @@ public void waitForKey(String prompt) throws IOException {
 
     public void tryTransaction(String queueName) throws IOException {
 
+        log("About to tryTranscation");
+
         _ch1.txSelect();
 
+        setChannelReturnListener();
+
         //test basicReturn handling in tx context
         returnCell = new BlockingCell<Object>();
         _ch1.basicPublish("", queueName, false, false, null, "normal".getBytes());
@@ -502,6 +519,8 @@ public void tryTransaction(String queueName) throws IOException {
         _ch1.txCommit();
         expect(2, drain(10, queueName, false));
 
+        unsetChannelReturnListener();
+        log("Finished tryTransaction");
     }