Record exchanges declared with Channel#exchangeDeclareNowait

Author: michaelklishin

src/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java

@@ -192,6 +192,12 @@ public AMQP.Exchange.DeclareOk exchangeDeclare(String exchange, String type, boo
 
     @Override
     public void exchangeDeclareNowait(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) throws IOException {
+        RecordedExchange x = new RecordedExchange(this, exchange).
+          type(type).
+          durable(durable).
+          autoDelete(autoDelete).
+          arguments(arguments);
+        recordExchange(exchange, x);
         delegate.exchangeDeclareNowait(exchange, type, durable, autoDelete, internal, arguments);
     }
 

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

@@ -146,6 +146,26 @@ public void handleNack(long deliveryTag, boolean multiple) throws IOException {
         wait(latch);
     }
 
+    public void testExchangeRecovery() throws IOException, InterruptedException, TimeoutException {
+        Channel ch = connection.createChannel();
+        String x = "java-client.test.recovery.x1";
+        declareExchange(ch, x);
+        closeAndWaitForRecovery();
+        expectChannelRecovery(ch);
+        expectExchangeRecovery(ch, x);
+        ch.exchangeDelete(x);
+    }
+
+    public void testExchangeRecoveryWithNoWait() throws IOException, InterruptedException, TimeoutException {
+        Channel ch = connection.createChannel();
+        String x = "java-client.test.recovery.x1-nowait";
+        declareExchangeNoWait(ch, x);
+        closeAndWaitForRecovery();
+        expectChannelRecovery(ch);
+        expectExchangeRecovery(ch, x);
+        ch.exchangeDelete(x);
+    }
+
     public void testClientNamedQueueRecovery() throws IOException, InterruptedException, TimeoutException {
         Channel ch = connection.createChannel();
         String q = "java-client.test.recovery.q1";
@@ -357,6 +377,14 @@ private AMQP.Queue.DeclareOk declareClientNamedQueue(Channel ch, String q) throw
         return ch.queueDeclare(q, true, false, false, null);
     }
 
+    private AMQP.Exchange.DeclareOk declareExchange(Channel ch, String x) throws IOException {
+        return ch.exchangeDeclare(x, "fanout", false);
+    }
+
+    private void declareExchangeNoWait(Channel ch, String x) throws IOException {
+        ch.exchangeDeclareNowait(x, "fanout", false, false, false, null);
+    }
+
     private void expectQueueRecovery(Channel ch, String q) throws IOException, InterruptedException, TimeoutException {
         ch.confirmSelect();
         ch.queuePurge(q);
@@ -368,6 +396,16 @@ private void expectQueueRecovery(Channel ch, String q) throws IOException, Inter
         assertEquals(1, ok2.getMessageCount());
     }
 
+    private void expectExchangeRecovery(Channel ch, String x) throws IOException, InterruptedException, TimeoutException {
+        ch.confirmSelect();
+        String q = ch.queueDeclare().getQueue();
+        final String rk = "routing-key";
+        ch.queueBind(q, x, rk);
+        ch.basicPublish(x, rk, null, "msg".getBytes());
+        waitForConfirms(ch);
+        ch.exchangeDeclarePassive(x);
+    }
+
     private CountDownLatch prepareForRecovery(Connection conn) {
         final CountDownLatch latch = new CountDownLatch(1);
         ((AutorecoveringConnection)conn).addRecoveryListener(new RecoveryListener() {