Merge with default

Author: Rob Harrop

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

@@ -145,6 +145,12 @@ private void expectUnexpectedFrameError(Confuser confuser)
         //NB: the frame confuser relies on the encoding of the
         //method field to be at least 8 bytes long
         channel.basicPublish("", "routing key", null, "Hello".getBytes());
+        // TODO remove when bug 24086 is fixed.
+        try {
+            Thread.sleep(100);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
         expectError(AMQP.UNEXPECTED_FRAME);
     }
 

test/src/com/rabbitmq/client/test/server/Firehose.java

@@ -0,0 +1,69 @@
+package com.rabbitmq.client.test.server;
+
+import com.rabbitmq.client.GetResponse;
+import com.rabbitmq.client.test.BrokerTestCase;
+import com.rabbitmq.tools.Host;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+public class Firehose extends BrokerTestCase {
+    private String q;
+    private String firehose;
+
+    @Override
+    protected void createResources() throws IOException {
+        super.createResources();
+        channel.exchangeDeclare("trace", "fanout", false, true, null);
+        channel.exchangeDeclare("test", "fanout", false, true, null);
+        q = channel.queueDeclare().getQueue();
+        firehose = channel.queueDeclare().getQueue();
+        channel.queueBind(q, "test", "");
+        channel.queueBind(firehose, "trace", "");
+    }
+
+    public void testFirehose() throws IOException {
+        publishGet("not traced");
+        enable();
+        GetResponse msg = publishGet("traced");
+        disable();
+        publishGet("not traced");
+
+        GetResponse publish = basicGet(firehose);
+        GetResponse deliver = basicGet(firehose);
+
+        assertNotNull(publish);
+        assertNotNull(deliver);
+        assertDelivered(firehose, 0);
+
+        // We don't test everything, that would be a bit OTT
+        checkHeaders(publish.getProps().getHeaders());
+
+        Map<String,Object> delHeaders = deliver.getProps().getHeaders();
+        checkHeaders(delHeaders);
+        assertNotNull(delHeaders.get("redelivered"));
+
+        assertEquals(msg.getBody().length, publish.getBody().length);
+        assertEquals(msg.getBody().length, deliver.getBody().length);
+    }
+
+    private GetResponse publishGet(String key) throws IOException {
+        basicPublishVolatile("test", key);
+        return basicGet(q);
+    }
+
+    private void checkHeaders(Map<String, Object> pubHeaders) {
+        assertEquals("test", pubHeaders.get("exchange_name").toString());
+        List<Object> routing_keys = (List<Object>) pubHeaders.get("routing_keys");
+        assertEquals("traced", routing_keys.get(0).toString());
+    }
+
+    private void enable() throws IOException {
+        Host.rabbitmqctl("set_env trace_exchanges '[{<<\"/\">>, <<\"trace\">>}]'");
+    }
+
+    private void disable() throws IOException {
+        Host.rabbitmqctl("unset_env trace_exchanges");
+    }
+}

test/src/com/rabbitmq/client/test/server/Permissions.java

@@ -65,28 +65,21 @@ protected void tearDown()
     protected void addRestrictedAccount()
         throws IOException
     {
-        runCtl("add_user test test");
-        runCtl("add_user testadmin test");
-        runCtl("add_vhost /test");
-        runCtl("set_permissions -p /test test configure write read");
-        runCtl("set_permissions -p /test testadmin \".*\" \".*\" \".*\"");
+        Host.rabbitmqctl("add_user test test");
+        Host.rabbitmqctl("add_user testadmin test");
+        Host.rabbitmqctl("add_vhost /test");
+        Host.rabbitmqctl("set_permissions -p /test test configure write read");
+        Host.rabbitmqctl("set_permissions -p /test testadmin \".*\" \".*\" \".*\"");
     }
 
     protected void deleteRestrictedAccount()
         throws IOException
     {
-        runCtl("clear_permissions -p /test testadmin");
-        runCtl("clear_permissions -p /test test");
-        runCtl("delete_vhost /test");
-        runCtl("delete_user testadmin");
-        runCtl("delete_user test");
-    }
-
-    protected void runCtl(String command)
-        throws IOException
-    {
-        Host.executeCommand("../rabbitmq-server/scripts/rabbitmqctl " +
-                            command);
+        Host.rabbitmqctl("clear_permissions -p /test testadmin");
+        Host.rabbitmqctl("clear_permissions -p /test test");
+        Host.rabbitmqctl("delete_vhost /test");
+        Host.rabbitmqctl("delete_user testadmin");
+        Host.rabbitmqctl("delete_user test");
     }
 
     protected void createResources()
@@ -241,7 +234,7 @@ public void testAltExchConfiguration()
     public void testNoAccess()
         throws IOException, InterruptedException
     {
-        runCtl("set_permissions -p /test test \"\" \"\" \"\"");
+        Host.rabbitmqctl("set_permissions -p /test test \"\" \"\" \"\"");
         Thread.sleep(2000);
 
         expectExceptionRun(AMQP.ACCESS_REFUSED, new WithName() {

test/src/com/rabbitmq/client/test/server/ServerTests.java

@@ -30,6 +30,7 @@ public static TestSuite suite() {
         suite.addTestSuite(AlternateExchangeEquivalence.class);
         suite.addTestSuite(MemoryAlarms.class);
         suite.addTest(PersisterRestartTests.suite());
+        suite.addTestSuite(Firehose.class);
         return suite;
     }
 }

test/src/com/rabbitmq/tools/Host.java

@@ -68,4 +68,8 @@ public static void executeCommand(String command)
                                   "\nstderr:\n" + stderr + "\n");
         }
     }
+
+    public static void rabbitmqctl(String command) throws IOException {
+        executeCommand("../rabbitmq-server/scripts/rabbitmqctl " + command);
+    }
 }