Merge pull request #214 from rabbitmq/rabbitmq-java-client-210

Enable automatic recovery by default

Author: michaelklishin

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -100,7 +100,7 @@ public class ConnectionFactory implements Cloneable {
     private SocketConfigurator socketConf         = new DefaultSocketConfigurator();
     private ExceptionHandler exceptionHandler     = new DefaultExceptionHandler();
 
-    private boolean automaticRecovery             = false;
+    private boolean automaticRecovery             = true;
     private boolean topologyRecovery              = true;
 
     // long is used to make sure the users can use both ints

src/main/java/com/rabbitmq/client/impl/recovery/RecoveryAwareChannelN.java

@@ -82,16 +82,20 @@ private AMQImpl.Basic.Deliver offsetDeliveryTag(AMQImpl.Basic.Deliver method) {
 
     @Override
     public void basicAck(long deliveryTag, boolean multiple) throws IOException {
+        // FIXME no check if deliveryTag = 0 (ack all)
         long realTag = deliveryTag - activeDeliveryTagOffset;
-        if (realTag > 0) {
+        // 0 tag means ack all
+        if (realTag >= 0) {
             super.basicAck(realTag, multiple);
         }
     }
 
     @Override
     public void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException {
+        // FIXME no check if deliveryTag = 0 (nack all)
         long realTag = deliveryTag - activeDeliveryTagOffset;
-        if (realTag > 0) {
+        // 0 tag means nack all
+        if (realTag >= 0) {
             super.basicNack(realTag, multiple, requeue);
         }
     }

src/test/java/com/rabbitmq/client/test/BrokerTestCase.java

@@ -60,9 +60,14 @@ protected void finished(Description description) {
 
     protected ConnectionFactory newConnectionFactory() {
         ConnectionFactory connectionFactory = TestUtils.connectionFactory();
+        connectionFactory.setAutomaticRecoveryEnabled(isAutomaticRecoveryEnabled());
         return connectionFactory;
     }
 
+    protected boolean isAutomaticRecoveryEnabled() {
+        return true;
+    }
+
     protected Connection connection;
     protected Channel channel;
 

src/test/java/com/rabbitmq/client/test/SharedThreadPoolTest.java

@@ -31,6 +31,7 @@
 public class SharedThreadPoolTest extends BrokerTestCase {
     @Test public void willShutDownExecutor() throws IOException, TimeoutException {
         ConnectionFactory cf = TestUtils.connectionFactory();
+        cf.setAutomaticRecoveryEnabled(false);
         ExecutorService executor = Executors.newFixedThreadPool(8);
         cf.setSharedExecutor(executor);
 

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

@@ -15,35 +15,19 @@
 
 package com.rabbitmq.client.test.functional;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import com.rabbitmq.client.*;
+import com.rabbitmq.client.AMQP.BasicProperties;
+import com.rabbitmq.client.QueueingConsumer.Delivery;
+import com.rabbitmq.client.test.BrokerTestCase;
+import org.junit.Test;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Test;
-
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.AMQP.BasicProperties;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.DefaultConsumer;
-import com.rabbitmq.client.Envelope;
-import com.rabbitmq.client.GetResponse;
-import com.rabbitmq.client.MessageProperties;
-import com.rabbitmq.client.QueueingConsumer;
-import com.rabbitmq.client.QueueingConsumer.Delivery;
-import com.rabbitmq.client.test.BrokerTestCase;
+import static org.junit.Assert.*;
 
 public class DeadLetterExchange extends BrokerTestCase {
     public static final String DLX = "dead.letter.exchange";
@@ -149,23 +133,23 @@ protected void releaseResources() throws IOException {
             Map<String, Object> args = new HashMap<String, Object>();
             args.put(DLX_RK_ARG, "foo");
 
-            channel.queueDeclare("bar", false, true, false, args);
+            channel.queueDeclare(randomQueueName(), false, true, false, args);
             fail("dlx must be defined if dl-rk is set");
         } catch (IOException ex) {
             checkShutdownSignal(AMQP.PRECONDITION_FAILED, ex);
         }
     }
 
     @Test public void redeclareQueueWithRoutingKeyButNoDeadLetterExchange()
-        throws IOException
-    {
+        throws IOException, InterruptedException {
         try {
+            String queueName = randomQueueName();
             Map<String, Object> args = new HashMap<String, Object>();
-            channel.queueDeclare("bar", false, true, false, args);
+            channel.queueDeclare(queueName, false, true, false, args);
 
             args.put(DLX_RK_ARG, "foo");
 
-            channel.queueDeclare("bar", false, true, false, args);
+            channel.queueDeclare(queueName, false, true, false, args);
             fail("x-dead-letter-exchange must be specified if " +
                     "x-dead-letter-routing-key is set");
         } catch (IOException ex) {
@@ -711,4 +695,8 @@ public void process(GetResponse getResponse) {
 
         public void process(GetResponse response);
     }
+
+    private static String randomQueueName() {
+        return DeadLetterExchange.class.getSimpleName() + "-" + UUID.randomUUID().toString();
+    }
 }

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

@@ -16,16 +16,13 @@
 
 package com.rabbitmq.client.test.functional;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
+import com.rabbitmq.client.test.BrokerTestCase;
+import org.junit.Test;
 
 import java.io.IOException;
 
-import org.junit.Test;
-
-import com.rabbitmq.client.impl.AMQConnection;
-import com.rabbitmq.client.test.BrokerTestCase;
+import static org.junit.Assert.*;
 
 public class Heartbeat extends BrokerTestCase {
 
@@ -41,7 +38,7 @@ public Heartbeat()
         assertEquals(1, connection.getHeartbeat());
         Thread.sleep(3100);
         assertTrue(connection.isOpen());
-        ((AMQConnection)connection).setHeartbeat(0);
+        ((AutorecoveringConnection)connection).getDelegate().setHeartbeat(0);
         assertEquals(0, connection.getHeartbeat());
         Thread.sleep(3100);
         assertFalse(connection.isOpen());

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

@@ -15,19 +15,18 @@
 
 package com.rabbitmq.client.test.functional;
 
-import java.io.IOException;
-import java.net.Socket;
-
-import javax.net.SocketFactory;
-
-import com.rabbitmq.client.impl.*;
-import com.rabbitmq.client.test.TestUtils;
-import org.junit.Test;
-
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.DefaultSocketConfigurator;
+import com.rabbitmq.client.impl.*;
+import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
 import com.rabbitmq.client.test.BrokerTestCase;
+import com.rabbitmq.client.test.TestUtils;
+import org.junit.Test;
+
+import javax.net.SocketFactory;
+import java.io.IOException;
+import java.net.Socket;
 
 /**
  * Test that the server correctly handles us when we send it bad frames
@@ -177,7 +176,7 @@ public Frame confuse(Frame frame) {
     }
 
     private void expectError(int error, Confuser confuser) throws IOException {
-        ((ConfusedFrameHandler)((AMQConnection)connection).getFrameHandler()).
+        ((ConfusedFrameHandler)((AutorecoveringConnection)connection).getDelegate().getFrameHandler()).
             confuser = confuser;
 
         //NB: the frame confuser relies on the encoding of the

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

@@ -25,6 +25,7 @@
 
 import javax.net.SocketFactory;
 
+import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
 import com.rabbitmq.client.test.TestUtils;
 import org.junit.Test;
 
@@ -103,7 +104,7 @@ protected int negotiateChannelMax(int requestedChannelMax, int serverMax) {
             assertNull(conn.createChannel(n + 1));
 
             // Construct a channel directly
-            final ChannelN ch = new ChannelN((AMQConnection) conn, n + 1,
+            final ChannelN ch = new ChannelN(((AutorecoveringConnection) conn).getDelegate(), n + 1,
                                              new ConsumerWorkService(Executors.newSingleThreadExecutor(),
                                                      Executors.defaultThreadFactory(), ConnectionFactory.DEFAULT_SHUTDOWN_TIMEOUT));
             conn.addShutdownListener(new ShutdownListener() {

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

@@ -33,6 +33,14 @@
  */
 public class ExclusiveQueueDurability extends BrokerTestCase {
 
+    @Override
+    protected boolean isAutomaticRecoveryEnabled() {
+        // With automatic recovery enabled, queue can be re-created when launching the test suite
+        // (because FunctionalTests are launched independently and as part of the HATests)
+        // This then makes this test fail.
+        return false;
+    }
+
     void verifyQueueMissing(Channel channel, String queueName)
             throws IOException {
         try {

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

@@ -16,28 +16,20 @@
 
 package com.rabbitmq.client.test.server;
 
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import com.rabbitmq.client.*;
+import com.rabbitmq.client.impl.AMQChannel;
+import com.rabbitmq.client.impl.recovery.AutorecoveringChannel;
+import com.rabbitmq.client.test.BrokerTestCase;
+import com.rabbitmq.client.test.TestUtils;
+import com.rabbitmq.tools.Host;
+import org.junit.Test;
 
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeoutException;
 
-import com.rabbitmq.client.test.TestUtils;
-import org.junit.Test;
-
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.AlreadyClosedException;
-import com.rabbitmq.client.AuthenticationFailureException;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.QueueingConsumer;
-import com.rabbitmq.client.impl.AMQChannel;
-import com.rabbitmq.client.test.BrokerTestCase;
-import com.rabbitmq.tools.Host;
+import static org.junit.Assert.*;
 
 public class Permissions extends BrokerTestCase
 {
@@ -222,9 +214,9 @@ public void with(String name) throws IOException {
     {
         runTest(false, false, true, false, new WithName() {
                 public void with(String name) throws IOException {
-                    ((AMQChannel)channel)
-                    .exnWrappingRpc(new AMQP.Queue.Purge.Builder()
-                                                        .queue(name)
+                    AMQChannel channelDelegate = (AMQChannel) ((AutorecoveringChannel)channel).getDelegate();
+                    channelDelegate.exnWrappingRpc(new AMQP.Queue.Purge.Builder()
+                                    .queue(name)
                                     .build());
                 }});
     }