Merge pull request #270 from rabbitmq/rabbitmq-java-client-262

michaelklishin · web-flow · commit 8bd7c8c3cf3e · 2017-05-16T09:24:18.000+01:00
Handle TimeoutException on connection creation
diff --git a/src/main/java/com/rabbitmq/client/ConnectionFactory.java b/src/main/java/com/rabbitmq/client/ConnectionFactory.java
@@ -901,19 +901,28 @@ public Connection newConnection(ExecutorService executor, AddressResolver addres
             return conn;
         } else {
             List<Address> addrs = addressResolver.getAddresses();
-            IOException lastException = null;
+            Exception lastException = null;
             for (Address addr : addrs) {
                 try {
                     FrameHandler handler = fhFactory.create(addr);
-                    AMQConnection conn = new AMQConnection(params, handler, metricsCollector);
+                    AMQConnection conn = createConnection(params, handler, metricsCollector);
                     conn.start();
                     this.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");
         }
     }
 
@@ -940,6 +949,10 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         return result;
     }
 
+    protected AMQConnection createConnection(ConnectionParams params, FrameHandler frameHandler, MetricsCollector metricsCollector) {
+        return new AMQConnection(params, frameHandler, metricsCollector);
+    }
+
     /**
      * Create a new broker connection.
      *
diff --git a/src/main/java/com/rabbitmq/client/impl/recovery/RecoveryAwareAMQConnectionFactory.java b/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);
+    }
 }
diff --git a/src/test/java/com/rabbitmq/client/test/ClientTests.java b/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 {
 
diff --git a/src/test/java/com/rabbitmq/client/test/ConnectionFactoryTest.java b/src/test/java/com/rabbitmq/client/test/ConnectionFactoryTest.java
@@ -0,0 +1,66 @@
+// 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.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.MetricsCollector;
+import com.rabbitmq.client.impl.AMQConnection;
+import com.rabbitmq.client.impl.ConnectionParams;
+import com.rabbitmq.client.impl.FrameHandler;
+import com.rabbitmq.client.impl.FrameHandlerFactory;
+import org.junit.Test;
+
+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.*;
+
+public class ConnectionFactoryTest {
+
+    // see https://github.com/rabbitmq/rabbitmq-java-client/issues/262
+    @Test public void tryNextAddressIfTimeoutExceptionNoAutoRecovery() throws IOException, TimeoutException {
+        final AMQConnection connectionThatThrowsTimeout = mock(AMQConnection.class);
+        final AMQConnection connectionThatSucceeds = mock(AMQConnection.class);
+        final Queue<AMQConnection> connections = new ArrayBlockingQueue<AMQConnection>(10);
+        connections.add(connectionThatThrowsTimeout);
+        connections.add(connectionThatSucceeds);
+        ConnectionFactory connectionFactory = new ConnectionFactory() {
+
+            @Override
+            protected AMQConnection createConnection(ConnectionParams params, FrameHandler frameHandler, MetricsCollector metricsCollector) {
+                return connections.poll();
+            }
+
+            @Override
+            protected synchronized FrameHandlerFactory createFrameHandlerFactory() throws IOException {
+                return mock(FrameHandlerFactory.class);
+            }
+        };
+        connectionFactory.setAutomaticRecoveryEnabled(false);
+        doThrow(TimeoutException.class).when(connectionThatThrowsTimeout).start();
+        doNothing().when(connectionThatSucceeds).start();
+        Connection returnedConnection = connectionFactory.newConnection(
+            new Address[] { new Address("host1"), new Address("host2") }
+        );
+        assertSame(connectionThatSucceeds, returnedConnection);
+    }
+
+}
diff --git a/src/test/java/com/rabbitmq/client/test/RecoveryAwareAMQConnectionFactoryTest.java b/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);
+    }
+
+}
