stable to default

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/Channel.java

@@ -26,7 +26,6 @@
 import com.rabbitmq.client.AMQP.Tx;
 import com.rabbitmq.client.AMQP.Basic;
 import com.rabbitmq.client.AMQP.Confirm;
-import com.rabbitmq.client.AMQP.Channel.FlowOk;
 
 /**
  * Public API: Interface to an AMQ channel. See the <a href="http://www.amqp.org/">spec</a> for details.
@@ -90,17 +89,11 @@ public interface Channel extends ShutdownNotifier {
     void close(int closeCode, String closeMessage) throws IOException;
 
     /**
-     * Set flow on the channel
-     *
-     * @param active if true, the server is asked to start sending. If false, the server is asked to stop sending.
-     * @throws IOException
-     */
-    FlowOk flow(boolean active) throws IOException;
-
-    /**
-     * Return the current Channel.Flow settings.
+     * Indicates whether the server has asked this client to stop
+     * sending content-bearing commands (such as basic.publish) by
+     * issueing a channel.flow{active=false}.
      */
-    FlowOk getFlow();
+    boolean flowBlocked();
 
     /**
      * Abort this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code

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.
@@ -88,6 +88,13 @@ public class ConnectionFactory implements Cloneable {
     private Map<String, Object> _clientProperties = AMQConnection.defaultClientProperties();
     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
@@ -111,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());
     }
 
     /**
@@ -178,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
@@ -384,6 +392,40 @@ 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.
+     *
+     * It's developer's responsibility to shut down the executor
+     * when it is no longer needed.
+     *
+     * @param executor
+     */
+    public void setSharedExecutor(ExecutorService executor) {
+        this.sharedExecutor = executor;
+    }
+
     public boolean isSSL(){
         return getSocketFactory() instanceof SSLSocketFactory;
     }
@@ -433,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());
     }
 
     /**
@@ -485,43 +519,47 @@ protected void configureSocket(Socket socket) throws IOException{
      * @throws IOException if it encounters a problem
      */
     public Connection newConnection(Address[] addrs) throws IOException {
-        return newConnection(null, addrs);
+        return newConnection(this.sharedExecutor, addrs);
     }
 
     /**
      * Create a new broker connection
      * @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);
     }
 
     /**
@@ -530,7 +568,7 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
      * @throws IOException if it encounters a problem
      */
     public Connection newConnection() throws IOException {
-        return newConnection(null,
+        return newConnection(this.sharedExecutor,
                              new Address[] {new Address(getHost(), getPort())}
                             );
     }
@@ -554,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/Envelope.java

@@ -48,7 +48,11 @@ public long getDeliveryTag() {
     }
 
     /**
-     * Get the redelivery flag included in this parameter envelope
+     * Get the redelivery flag included in this parameter envelope. This is a
+     * hint as to whether this message may have been delivered before (but not
+     * acknowledged). If the flag is not set, the message definitely has not
+     * been delivered before. If it is set, it may have been delivered before.
+     *
      * @return the redelivery flag
      */
     public boolean isRedeliver() {
@@ -70,4 +74,14 @@ public String getExchange() {
     public String getRoutingKey() {
         return _routingKey;
     }
+
+    @Override public String toString() {
+        StringBuilder sb = new StringBuilder();
+        sb.append("Envelope(deliveryTag=").append(_deliveryTag);
+        sb.append(", redeliver=").append(_redeliver);
+        sb.append(", exchange=").append(_exchange);
+        sb.append(", routingKey=").append(_routingKey);
+        sb.append(")");
+        return sb.toString();
+    }
 }

src/com/rabbitmq/client/GetResponse.java

@@ -86,4 +86,15 @@ public byte[] getBody() {
     public int getMessageCount() {
         return messageCount;
     }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder();
+        sb.append("GetResponse(envelope=").append(envelope);
+        sb.append(", props=").append(props);
+        sb.append(", messageCount=").append(messageCount);
+        sb.append(", body=(elided, ").append(body.length).append(" bytes long)");
+        sb.append(")");
+        return sb.toString();
+    }
 }