refactor ConnectionFactory to allow configuration of the socket prior to connection

Author: David R. MacIver

src/com/rabbitmq/client/ConnectionFactory.java

@@ -36,6 +36,9 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import java.net.Socket;
+import java.net.InetSocketAddress;
+
 import javax.net.SocketFactory;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManager;
@@ -153,7 +156,26 @@ protected FrameHandler createFrameHandler(Address addr)
         String hostName = addr.getHost();
         int portNumber = addr.getPort();
         if (portNumber == -1) portNumber = AMQP.PROTOCOL.PORT;
-        return new SocketFrameHandler(_factory, hostName, portNumber);
+        Socket socket = _factory.createSocket();
+        configureSocket(socket);
+        socket.connect(new InetSocketAddress(hostName, portNumber));
+        return new SocketFrameHandler(socket);
+    }
+
+    /**
+     *  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
+     */
+    protected void configureSocket(Socket socket) throws IOException{
+        //disable Nagle's algorithm, for more consistently low latency
+        socket.setTcpNoDelay(true);
     }
 
     private Connection newConnection(Address[] addrs,

src/com/rabbitmq/client/impl/SocketFrameHandler.java

@@ -47,12 +47,6 @@
  */
 
 public class SocketFrameHandler implements FrameHandler {
-    /** Host we connect to */
-    public final String _host;
-
-    /** Port number we connect to */
-    public final int _port;
-
     /** The underlying socket */
     public final Socket _socket;
 
@@ -66,33 +60,21 @@ public class SocketFrameHandler implements FrameHandler {
     // twice simultaneously.
 
     /**
-     * Instantiate a SocketFrameHandler.
-     * @param factory the socket factory to use to build our Socket - may be SSLSocketFactory etc
-     * @param hostName the host name
-     * @param portNumber the port number
-     * @throws IOException if there is a problem accessing the connection
+     * @param socket the socket to use
      */
-    public SocketFrameHandler(SocketFactory factory,
-                              String hostName,
-                              int portNumber)
-        throws IOException
-    {
-        _host = hostName;
-        _port = portNumber;
-        _socket = factory.createSocket(_host, _port);
-        //disable Nagle's algorithm, for more consistently low latency
-        _socket.setTcpNoDelay(true);
+    public SocketFrameHandler(Socket socket) throws IOException{
+        _socket = socket;
 
         _inputStream = new DataInputStream(new BufferedInputStream(_socket.getInputStream()));
         _outputStream = new DataOutputStream(new BufferedOutputStream(_socket.getOutputStream()));
     }
 
     public String getHost() {
-        return _host;
+        return _socket.getInetAddress().getHostName(); 
     }
 
     public int getPort() {
-        return _port;
+        return _socket.getPort();
     }
 
     public void setTimeout(int timeoutMs)

test/src/com/rabbitmq/examples/TestMain.java

@@ -101,7 +101,7 @@ protected FrameHandler createFrameHandler(Address addr)
             String hostName = addr.getHost();
             int portNumber = addr.getPort();
             if (portNumber == -1) portNumber = AMQP.PROTOCOL.PORT;
-            return new SocketFrameHandler(getSocketFactory(), hostName, portNumber) {
+            return new SocketFrameHandler(getSocketFactory().createSocket(hostName, portNumber)) {
                     public void sendHeader() throws IOException {
                         sendHeader(protocolMajor, protocolMinor);
                     }