move transactional basic.return tests from TestMain to junit

and expand them somewhat

Author: rade

test/src/com/rabbitmq/client/test/functional/Routing.java

@@ -28,6 +28,7 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeoutException;
 
 public class Routing extends BrokerTestCase
 {
@@ -232,29 +233,19 @@ public void testHeadersRouting() throws Exception {
         checkGet(Q2, false);
     }
 
-    public void testBasicReturn() throws Exception {
-        channel.addReturnListener(new ReturnListener() {
-                public void handleReturn(int replyCode,
-                                         String replyText,
-                                         String exchange,
-                                         String routingKey,
-                                         AMQP.BasicProperties properties,
-                                         byte[] body)
-                    throws IOException {
-                    Routing.this.returnCell.set(replyCode);
-                }
-            });
+    public void testBasicReturn() throws IOException {
+        channel.addReturnListener(makeReturnListener());
         returnCell = new BlockingCell<Integer>();
+
+        //returned 'mandatory' publish
         channel.basicPublish("", "unknown", true, false, null, "mandatory1".getBytes());
-        int replyCode = returnCell.uninterruptibleGet();
-        assertEquals(replyCode, AMQP.NO_ROUTE);
+        checkReturn(AMQP.NO_ROUTE);
 
-        returnCell = new BlockingCell<Integer>();
+        //routed 'mandatory' publish
         channel.basicPublish("", Q1, true, false, null, "mandatory2".getBytes());
-        GetResponse r = channel.basicGet(Q1, true);
-        assertNotNull(r);
-        assertEquals(new String(r.getBody()), "mandatory2");
+        assertNotNull(channel.basicGet(Q1, true));
 
+        //'immediate' publish
         channel.basicPublish("", Q1, false, true, null, "immediate".getBytes());
         try {
             channel.basicQos(0); //flush
@@ -264,4 +255,51 @@ public void handleReturn(int replyCode,
         }
     }
 
+    public void testBasicReturnTransactional() throws IOException {
+        channel.txSelect();
+        channel.addReturnListener(makeReturnListener());
+        returnCell = new BlockingCell<Integer>();
+
+        //returned 'mandatory' publish
+        channel.basicPublish("", "unknown", true, false, null, "mandatory1".getBytes());
+        try {
+            returnCell.uninterruptibleGet(200);
+            fail("basic.return issued prior to tx.commit");
+        } catch (TimeoutException toe) {}
+        channel.txCommit();
+        checkReturn(AMQP.NO_ROUTE);
+
+        //routed 'mandatory' publish
+        channel.basicPublish("", Q1, true, false, null, "mandatory2".getBytes());
+        channel.txCommit();
+        assertNotNull(channel.basicGet(Q1, true));
+
+        //returned 'mandatory' publish when message is routable on
+        //publish but not on commit
+        channel.basicPublish("", Q1, true, false, null, "mandatory2".getBytes());
+        channel.queueDelete(Q1);
+        channel.txCommit();
+        checkReturn(AMQP.NO_ROUTE);
+        channel.queueDeclare(Q1, false, false, false, null);
+    }
+
+    protected ReturnListener makeReturnListener() {
+        return new ReturnListener() {
+            public void handleReturn(int replyCode,
+                                     String replyText,
+                                     String exchange,
+                                     String routingKey,
+                                     AMQP.BasicProperties properties,
+                                     byte[] body)
+                throws IOException {
+                Routing.this.returnCell.set(replyCode);
+            }
+        };
+    }
+
+    protected void checkReturn(int replyCode) {
+        assertEquals((int)returnCell.uninterruptibleGet(), AMQP.NO_ROUTE);
+        returnCell = new BlockingCell<Integer>();
+    }
+
 }

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

@@ -253,8 +253,6 @@ public void run() throws IOException {
         sendLotsOfTrivialMessages(batchSize, queueName);
         expect(batchSize, drain(batchSize, queueName, true));
 
-        tryTransaction(queueName);
-
         _ch1.close();
 
         log("Closing.");
@@ -463,30 +461,6 @@ 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());
-        _ch1.basicPublish("", queueName, true, false, null, "mandatory".getBytes());
-        _ch1.basicPublish("", "bogus", true, false, null, "mandatory".getBytes());
-        _ch1.txCommit();
-        doBasicReturn(returnCell, AMQP.NO_ROUTE);
-        returnCell = new BlockingCell<Object>();
-        expect(2, drain(10, queueName, false));
-
-        unsetChannelReturnListener();
-        log("Finished tryTransaction");
-    }
-
-
-
     // utility: tell what Java compiler version a class was compiled with
     public static String getCompilerVersion(Class<?> clazz) throws IOException {
         String resourceName = "/" + clazz.getName().replace('.', '/') + ".class";