merge bug22159 into default

Author: Matthias Radestock

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;
 
@@ -62,37 +56,25 @@ public class SocketFrameHandler implements FrameHandler {
     /** Socket's outputstream - data to the broker */
     public final DataOutputStream _outputStream;
 
-    // Note, we use each of these to synchronize on to make sure we don't try to use them
-    // twice simultaneously.
+    // Note, we use each of these to synchronize on to make sure we
+    // don't try to use them 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/BufferPerformanceMetrics.java

@@ -0,0 +1,112 @@
+package com.rabbitmq.examples;
+
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.MessageProperties;
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.AMQP.Queue;
+
+import java.net.Socket;
+import java.io.IOException;
+import java.util.Random;
+
+
+/**
+ * Class to explore how performance of sending and receiving messages
+ * varies with the buffer size and enabling/disabling Nagle's
+ * algorithm.
+ */
+public class BufferPerformanceMetrics {
+
+    public static final int MESSAGE_COUNT  = 100000;
+    public static final byte[] MESSAGE     = "".getBytes();
+    public static final int REPEATS        = 1000000;
+    public static final int PEAK_SIZE      = 20 * 1024;
+
+    public static double NANOSECONDS_PER_SECOND = 1000 * 1000 * 1000;
+
+    public static void main(String[] args) throws Exception {
+        String hostName = args.length > 0 ? args[0] : "localhost";
+
+        Random rnd = new Random();
+
+        System.out.println("buffer size, " +
+                           "publish rate with nagle, " +
+                           "consume rate with nagle, " +
+                           "publish rate without nagle, " +
+                           "consume rate without nagle");
+
+        for(int repeat = 0; repeat < REPEATS; repeat++) {
+            final int bufferSize = 1 + rnd.nextInt(PEAK_SIZE);
+
+            double
+                publishRateNagle   = 0,
+                publishRateNoNagle = 0,
+                consumeRateNagle   = 0,
+                consumeRateNoNagle = 0;
+
+            for(final boolean useNagle : new boolean[] { false, true }) {
+                ConnectionFactory factory = new ConnectionFactory() {
+                        public void configureSocket(Socket socket)
+                            throws IOException {
+                            socket.setTcpNoDelay(!useNagle);
+                            socket.setReceiveBufferSize(bufferSize);
+                            socket.setSendBufferSize(bufferSize);
+                        }
+                    };
+
+                Connection connection = factory.newConnection(hostName);
+                Channel channel = connection.createChannel();
+                Queue.DeclareOk res = channel.queueDeclare();
+                String queueName = res.getQueue();
+
+                long start;
+
+                start = System.nanoTime();
+
+                for(int i = 0; i < MESSAGE_COUNT; i++) {
+                    channel.basicPublish("", queueName,
+                                         MessageProperties.BASIC, MESSAGE);
+                }
+
+                QueueingConsumer consumer = new QueueingConsumer(channel);
+                channel.basicConsume(queueName, true, consumer);
+
+                long publishTime = System.nanoTime() - start;
+
+                start = System.nanoTime();
+
+                for(int i = 0; i < MESSAGE_COUNT; i++){
+                    consumer.nextDelivery();
+                }
+
+                long consumeTime = System.nanoTime() - start;
+
+                double publishRate =
+                    MESSAGE_COUNT / (publishTime / NANOSECONDS_PER_SECOND);
+                double consumeRate =
+                    MESSAGE_COUNT / (consumeTime / NANOSECONDS_PER_SECOND);
+
+                if(useNagle){
+                    publishRateNagle = publishRate;
+                    consumeRateNagle = consumeRate;
+                } else {
+                    publishRateNoNagle = publishRate;
+                    consumeRateNoNagle = consumeRate;
+                }
+
+                connection.close();
+                // Small sleep to remove noise from hammering the server.
+                Thread.sleep(100);
+            }
+
+            System.out.println(bufferSize + ", " +
+                               publishRateNagle + ", " +
+                               consumeRateNagle + ", " +
+                               publishRateNoNagle + ", " +
+                               consumeRateNoNagle);
+        }
+    }
+}

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);
                     }