Handle TimeoutException in RecoveryAwareAMQConnectionFactory

Fixes #262

Author: acogoluegnes

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

@@ -51,28 +51,42 @@ public RecoveryAwareAMQConnectionFactory(ConnectionParams params, FrameHandlerFa
      * @return an interface to the connection
      * @throws java.io.IOException if it encounters a problem
      */
-    RecoveryAwareAMQConnection newConnection() throws IOException, TimeoutException {
-        IOException lastException = null;
+    // package protected API, made public for testing only
+    public RecoveryAwareAMQConnection newConnection() throws IOException, TimeoutException {
+        Exception lastException = null;
         List<Address> shuffled = shuffle(addressResolver.getAddresses());
 
         for (Address addr : shuffled) {
             try {
                 FrameHandler frameHandler = factory.create(addr);
-                RecoveryAwareAMQConnection conn = new RecoveryAwareAMQConnection(params, frameHandler, metricsCollector);
+                RecoveryAwareAMQConnection conn = createConnection(params, frameHandler, metricsCollector);
                 conn.start();
                 metricsCollector.newConnection(conn);
                 return conn;
             } catch (IOException e) {
                 lastException = e;
+            } catch (TimeoutException te) {
+                lastException = te;
             }
         }
 
-        throw (lastException != null) ? lastException : new IOException("failed to connect");
+        if (lastException != null) {
+            if (lastException instanceof IOException) {
+                throw (IOException) lastException;
+            } else if (lastException instanceof TimeoutException) {
+                throw (TimeoutException) lastException;
+            }
+        }
+        throw new IOException("failed to connect");
     }
 
     private static List<Address> shuffle(List<Address> addrs) {
         List<Address> list = new ArrayList<Address>(addrs);
         Collections.shuffle(list);
         return list;
     }
+
+    protected RecoveryAwareAMQConnection createConnection(ConnectionParams params, FrameHandler handler, MetricsCollector metricsCollector) {
+        return new RecoveryAwareAMQConnection(params, handler, metricsCollector);
+    }
 }

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

@@ -43,7 +43,9 @@
     DnsRecordIpAddressResolverTests.class,
     StandardMetricsCollectorTest.class,
     DnsSrvRecordAddressResolverTest.class,
-    JavaNioTest.class
+    JavaNioTest.class,
+    ConnectionFactoryTest.class,
+    RecoveryAwareAMQConnectionFactoryTest.class
 })
 public class ClientTests {
 

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

@@ -23,19 +23,15 @@
 import com.rabbitmq.client.impl.ConnectionParams;
 import com.rabbitmq.client.impl.FrameHandler;
 import com.rabbitmq.client.impl.FrameHandlerFactory;
-import org.junit.Assert;
 import org.junit.Test;
-import org.mockito.Mockito;
 
 import java.io.IOException;
 import java.util.Queue;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.TimeoutException;
 
 import static org.junit.Assert.assertSame;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.*;
 
 public class ConnectionFactoryTest {
 

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

@@ -0,0 +1,68 @@
+// Copyright (c) 2007-Present Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+
+package com.rabbitmq.client.test;
+
+import com.rabbitmq.client.Address;
+import com.rabbitmq.client.AddressResolver;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.MetricsCollector;
+import com.rabbitmq.client.impl.ConnectionParams;
+import com.rabbitmq.client.impl.FrameHandler;
+import com.rabbitmq.client.impl.FrameHandlerFactory;
+import com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnection;
+import com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnectionFactory;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.Assert.assertSame;
+import static org.mockito.Mockito.*;
+
+public class RecoveryAwareAMQConnectionFactoryTest {
+
+    // see https://github.com/rabbitmq/rabbitmq-java-client/issues/262
+    @Test public void tryNextAddressIfTimeoutException() throws IOException, TimeoutException {
+        final RecoveryAwareAMQConnection connectionThatThrowsTimeout = mock(RecoveryAwareAMQConnection.class);
+        final RecoveryAwareAMQConnection connectionThatSucceeds = mock(RecoveryAwareAMQConnection.class);
+        final Queue<RecoveryAwareAMQConnection> connections = new ArrayBlockingQueue<RecoveryAwareAMQConnection>(10);
+        connections.add(connectionThatThrowsTimeout);
+        connections.add(connectionThatSucceeds);
+        AddressResolver addressResolver = new AddressResolver() {
+            @Override
+            public List<Address> getAddresses() throws IOException {
+                return Arrays.asList(new Address("host1"), new Address("host2"));
+            }
+        };
+        RecoveryAwareAMQConnectionFactory connectionFactory = new RecoveryAwareAMQConnectionFactory(
+            new ConnectionParams(), mock(FrameHandlerFactory.class), addressResolver
+        ) {
+            @Override
+            protected RecoveryAwareAMQConnection createConnection(ConnectionParams params, FrameHandler handler, MetricsCollector metricsCollector) {
+                return connections.poll();
+            }
+        };
+        doThrow(TimeoutException.class).when(connectionThatThrowsTimeout).start();
+        doNothing().when(connectionThatSucceeds).start();
+        Connection returnedConnection = connectionFactory.newConnection();
+        assertSame(connectionThatSucceeds, returnedConnection);
+    }
+
+}