Enable automatic recovery by default (WIP)

Fixes #210

Author: acogoluegnes

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/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/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/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());
                 }});
     }