Merge bug14587

Author: Simon MacMullen

src/com/rabbitmq/client/Address.java

@@ -43,7 +43,7 @@ public Address(String host, int port) {
      */
     public Address(String host) {
         _host = host;
-        _port = -1;
+        _port = ConnectionFactory.USE_DEFAULT_PORT;
     }
 
     /**

src/com/rabbitmq/client/Connection.java

@@ -233,5 +233,4 @@ public interface Connection extends ShutdownNotifier { // rename to AMQPConnecti
      * Remove all {@link BlockedListener}s.
      */
     void clearBlockedListeners();
-
 }

src/com/rabbitmq/client/ConnectionFactory.java

@@ -22,8 +22,6 @@
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 
-import java.net.InetSocketAddress;
-import java.net.Socket;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.URLDecoder;
@@ -34,8 +32,10 @@
 import javax.net.ssl.TrustManager;
 
 import com.rabbitmq.client.impl.AMQConnection;
+import com.rabbitmq.client.impl.ConnectionParams;
 import com.rabbitmq.client.impl.FrameHandler;
-import com.rabbitmq.client.impl.SocketFrameHandler;
+import com.rabbitmq.client.impl.FrameHandlerFactory;
+import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
 
 /**
  * Convenience "factory" class to facilitate opening a {@link Connection} to an AMQP broker.
@@ -89,6 +89,12 @@ public class ConnectionFactory implements Cloneable {
     private SocketFactory factory                 = SocketFactory.getDefault();
     private SaslConfig saslConfig                 = DefaultSaslConfig.PLAIN;
     private ExecutorService sharedExecutor;
+    private SocketConfigurator socketConf         = new DefaultSocketConfigurator();
+
+    private boolean automaticRecovery             = false;
+    private boolean topologyRecovery              = true;
+
+    private int networkRecoveryInterval           = 5000;
 
     /** @return number of consumer threads in default {@link ExecutorService} */
     @Deprecated
@@ -112,15 +118,15 @@ public void setHost(String host) {
         this.host = host;
     }
 
-    private int portOrDefault(int port){
-        if(port != USE_DEFAULT_PORT) return port;
-        else if(isSSL()) return DEFAULT_AMQP_OVER_SSL_PORT;
+    public static int portOrDefault(int port, boolean ssl) {
+        if (port != USE_DEFAULT_PORT) return port;
+        else if (ssl) return DEFAULT_AMQP_OVER_SSL_PORT;
         else return DEFAULT_AMQP_PORT;
     }
 
     /** @return the default port to use for connections */
     public int getPort() {
-        return portOrDefault(port);
+        return portOrDefault(port, isSSL());
     }
 
     /**
@@ -179,6 +185,7 @@ public void setVirtualHost(String virtualHost) {
         this.virtualHost = virtualHost;
     }
 
+
     /**
      * Convenience method for setting the fields in an AMQP URI: host,
      * port, username, password and virtual host.  If any part of the
@@ -385,6 +392,27 @@ public void setSocketFactory(SocketFactory factory) {
         this.factory = factory;
     }
 
+    /**
+     * Get the socket configurator.
+     *
+     * @see #setSocketConfigurator(SocketConfigurator)
+     */
+    @SuppressWarnings("unused")
+    public SocketConfigurator getSocketConfigurator() {
+        return socketConf;
+    }
+
+    /**
+     * Set the socket configurator. This gets a chance to "configure" a socket
+     * after it has been opened. The default socket configurator disables
+     * Nagle's algorithm.
+     *
+     * @param socketConfigurator the configurator to use
+     */
+    public void setSocketConfigurator(SocketConfigurator socketConfigurator) {
+        this.socketConf = socketConfigurator;
+    }
+
     /**
      * Set the executor to use by default for newly created connections.
      * All connections that use this executor share it.
@@ -447,49 +475,41 @@ public void useSslProtocol(SSLContext context)
         setSocketFactory(context.getSocketFactory());
     }
 
-    protected FrameHandler createFrameHandler(Address addr)
-        throws IOException {
-
-        String hostName = addr.getHost();
-        int portNumber = portOrDefault(addr.getPort());
-        Socket socket = null;
-        try {
-            socket = factory.createSocket();
-            configureSocket(socket);
-            socket.connect(new InetSocketAddress(hostName, portNumber),
-                    connectionTimeout);
-            return createFrameHandler(socket);
-        } catch (IOException ioe) {
-            quietTrySocketClose(socket);
-            throw ioe;
-        }
+    /**
+     * Returns true if automatic connection recovery is enabled, false otherwise
+     * @return true if automatic connection recovery is enabled, false otherwise
+     */
+    public boolean isAutomaticRecoveryEnabled() {
+        return automaticRecovery;
     }
 
-    private static void quietTrySocketClose(Socket socket) {
-        if (socket != null)
-            try { socket.close(); } catch (Exception _) {/*ignore exceptions*/}
+    /**
+     * Enables or disables automatic connection recovery
+     * @param automaticRecovery if true, enables connection recovery
+     */
+    public void setAutomaticRecoveryEnabled(boolean automaticRecovery) {
+        this.automaticRecovery = automaticRecovery;
     }
 
-    protected FrameHandler createFrameHandler(Socket sock)
-        throws IOException
-    {
-        return new SocketFrameHandler(sock);
+    /**
+     * Returns true if topology recovery is enabled, false otherwise
+     * @return true if topology recovery is enabled, false otherwise
+     */
+    @SuppressWarnings("unused")
+    public boolean isTopologyRecoveryEnabled() {
+        return topologyRecovery;
     }
 
     /**
-     *  Provides a hook to insert custom configuration of the sockets
-     *  used to connect to an AMQP server before they connect.
-     *
-     *  The default behaviour of this method is to disable Nagle's
-     *  algorithm to get more consistently low latency.  However it
-     *  may be overridden freely and there is no requirement to retain
-     *  this behaviour.
-     *
-     *  @param socket The socket that is to be used for the Connection
+     * Enables or disables topology recovery
+     * @param topologyRecovery if true, enables topology recovery
      */
-    protected void configureSocket(Socket socket) throws IOException{
-        // disable Nagle's algorithm, for more consistently low latency
-        socket.setTcpNoDelay(true);
+    public void setTopologyRecoveryEnabled(boolean topologyRecovery) {
+        this.topologyRecovery = topologyRecovery;
+    }
+
+    protected FrameHandlerFactory createFrameHandlerFactory() throws IOException {
+        return new FrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL());
     }
 
     /**
@@ -507,35 +527,39 @@ public Connection newConnection(Address[] addrs) throws IOException {
      * @param executor thread execution service for consumers on the connection
      * @param addrs an array of known broker addresses (hostname/port pairs) to try in order
      * @return an interface to the connection
-     * @throws IOException if it encounters a problem
+     * @throws java.io.IOException if it encounters a problem
      */
     public Connection newConnection(ExecutorService executor, Address[] addrs)
         throws IOException
     {
-        IOException lastException = null;
-        for (Address addr : addrs) {
-            try {
-                FrameHandler frameHandler = createFrameHandler(addr);
-                AMQConnection conn =
-                    new AMQConnection(username,
-                                      password,
-                                      frameHandler,
-                                      executor,
-                                      virtualHost,
-                                      getClientProperties(),
-                                      requestedFrameMax,
-                                      requestedChannelMax,
-                                      requestedHeartbeat,
-                                      saslConfig);
-                conn.start();
-                return conn;
-            } catch (IOException e) {
-                lastException = e;
+        FrameHandlerFactory fhFactory = createFrameHandlerFactory();
+        ConnectionParams params = params(executor);
+
+        if (isAutomaticRecoveryEnabled()) {
+            // see com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnectionFactory#newConnection
+            AutorecoveringConnection conn = new AutorecoveringConnection(params, fhFactory, addrs);
+            conn.init();
+            return conn;
+        } else {
+            IOException lastException = null;
+            for (Address addr : addrs) {
+                try {
+                    FrameHandler handler = fhFactory.create(addr);
+                    AMQConnection conn = new AMQConnection(params, handler);
+                    conn.start();
+                    return conn;
+                } catch (IOException e) {
+                    lastException = e;
+                }
             }
+            throw (lastException != null) ? lastException : new IOException("failed to connect");
         }
+    }
 
-        throw (lastException != null) ? lastException
-                                      : new IOException("failed to connect");
+    public ConnectionParams params(ExecutorService executor) {
+        return new ConnectionParams(username, password, executor, virtualHost, getClientProperties(),
+                                    requestedFrameMax, requestedChannelMax, requestedHeartbeat, saslConfig,
+                                    networkRecoveryInterval, topologyRecovery);
     }
 
     /**
@@ -568,4 +592,20 @@ public Connection newConnection(ExecutorService executor) throws IOException {
             throw new Error(e);
         }
     }
+
+    /**
+     * Returns automatic connection recovery interval in milliseconds.
+     * @return how long will automatic recovery wait before attempting to reconnect, in ms; default is 5000
+     */
+    public int getNetworkRecoveryInterval() {
+        return networkRecoveryInterval;
+    }
+
+    /**
+     * Sets connection recovery interval. Default is 5000.
+     * @param networkRecoveryInterval how long will automatic recovery wait before attempting to reconnect, in ms
+     */
+    public void setNetworkRecoveryInterval(int networkRecoveryInterval) {
+        this.networkRecoveryInterval = networkRecoveryInterval;
+    }
 }

src/com/rabbitmq/client/DefaultSocketConfigurator.java

@@ -0,0 +1,22 @@
+package com.rabbitmq.client;
+
+import java.io.IOException;
+import java.net.Socket;
+
+public class DefaultSocketConfigurator implements SocketConfigurator {
+    /**
+     *  Provides a hook to insert custom configuration of the sockets
+     *  used to connect to an AMQP server before they connect.
+     *
+     *  The default behaviour of this method is to disable Nagle's
+     *  algorithm to get more consistently low latency.  However it
+     *  may be overridden freely and there is no requirement to retain
+     *  this behaviour.
+     *
+     *  @param socket The socket that is to be used for the Connection
+     */
+    public void configure(Socket socket) throws IOException {
+        // disable Nagle's algorithm, for more consistently low latency
+        socket.setTcpNoDelay(true);
+    }
+}

src/com/rabbitmq/client/Recoverable.java

@@ -0,0 +1,16 @@
+package com.rabbitmq.client;
+
+/**
+ * Provides a way to register (network, AMQP 0-9-1) connection recovery
+ * callbacks.
+ */
+public interface Recoverable {
+    /**
+     * Registers a connection recovery callback.
+     *
+     * @param f Callback function
+     */
+    public void addRecoveryListener(RecoveryListener listener);
+
+    public void removeRecoveryListener(RecoveryListener listener);
+}

src/com/rabbitmq/client/RecoveryListener.java

@@ -0,0 +1,11 @@
+package com.rabbitmq.client;
+
+/**
+ * A RecoveryListener receives notifications about completed automatic connection
+ * recovery.
+ *
+ * @since 3.3.0
+ */
+public interface RecoveryListener {
+    public void handleRecovery(Recoverable recoverable);
+}

src/com/rabbitmq/client/SocketConfigurator.java

@@ -0,0 +1,12 @@
+package com.rabbitmq.client;
+
+import java.io.IOException;
+import java.net.Socket;
+
+public interface SocketConfigurator {
+    /**
+     * Provides a hook to insert custom configuration of the sockets
+     * used to connect to an AMQP server before they connect.
+     */
+    void configure(Socket socket) throws IOException;
+}

src/com/rabbitmq/client/TopologyRecoveryException.java

@@ -0,0 +1,13 @@
+package com.rabbitmq.client;
+
+/**
+ * Indicates an exception thrown during topology recovery.
+ *
+ * @see com.rabbitmq.client.ConnectionFactory#setTopologyRecovery(boolean)
+ * @since 3.3.0
+ */
+public class TopologyRecoveryException extends Exception {
+    public TopologyRecoveryException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}