add test for basic.return{immediate=true} failure

and move 'mandatory' tests over to junit

Author: rade

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

@@ -20,6 +20,8 @@
 import com.rabbitmq.client.test.BrokerTestCase;
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.GetResponse;
+import com.rabbitmq.client.ReturnListener;
+import com.rabbitmq.utility.BlockingCell;
 
 import java.io.IOException;
 import java.util.List;
@@ -34,6 +36,8 @@ public class Routing extends BrokerTestCase
     protected final String Q1 = "foo";
     protected final String Q2 = "bar";
 
+    private volatile BlockingCell<Integer> returnCell;
+
     protected void createResources() throws IOException {
         channel.exchangeDeclare(E, "direct");
         channel.queueDeclare(Q1, false, false, false, null);
@@ -228,5 +232,36 @@ 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);
+                }
+            });
+        returnCell = new BlockingCell<Integer>();
+        channel.basicPublish("", "unknown", true, false, null, "mandatory1".getBytes());
+        int replyCode = returnCell.uninterruptibleGet();
+        assertEquals(replyCode, AMQP.NO_ROUTE);
+
+        returnCell = new BlockingCell<Integer>();
+        channel.basicPublish("", Q1, true, false, null, "mandatory2".getBytes());
+        GetResponse r = channel.basicGet(Q1, true);
+        assertNotNull(r);
+        assertEquals(new String(r.getBody()), "mandatory2");
+
+        channel.basicPublish("", Q1, false, true, null, "immediate".getBytes());
+        try {
+            channel.basicQos(0); //flush
+            fail("basic.publish{immediate=true} should not be supported");
+        } catch (IOException ioe) {
+            checkShutdownSignal(AMQP.NOT_IMPLEMENTED, ioe);
+        }
+    }
 
+}

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

@@ -248,7 +248,6 @@ public void run() throws IOException {
         _ch1.basicCancel(cTag2);
 
         tryTopics();
-        tryBasicReturn();
 
         queueName =_ch1.queueDeclare().getQueue();
         sendLotsOfTrivialMessages(batchSize, queueName);
@@ -449,33 +448,6 @@ public void doBasicReturn(BlockingCell<Object> cell, int expectedCode) {
         }
     }
 
-    public void tryBasicReturn() throws IOException {
-        log("About to try mandatory publications");
-
-        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);
-
-        String mq = "mandatoryTestQueue";
-        _ch1.queueDeclare(mq, false, false, true, null);
-        _ch1.queueBind(mq, mx, "");
-
-        returnCell = new BlockingCell<Object>();
-        _ch1.basicPublish(mx, "", true, false, null, "two".getBytes());
-        drain(1, mq, true);
-        _ch1.queueDelete(mq, true, true);
-
-        unsetChannelReturnListener();
-
-        log("Completed basic.return testing.");
-
-    }
-
     private void unsetChannelReturnListener() {
         _ch1.clearReturnListeners();
         log("ReturnListeners unset");