Backport eb09ae778764 (Merge of bug24030; Remove flushes after every frame to boost performance with SSL)

Author: Simon MacMullen

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

@@ -121,6 +121,8 @@ public void transmit(AMQChannel channel) throws IOException {
                 }
             }
         }
+
+        connection.flush();
     }
 
     @Override public String toString() {

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

@@ -474,6 +474,13 @@ public void writeFrame(Frame f) throws IOException {
         _heartbeatSender.signalActivity();
     }
 
+    /**
+     * Public API - flush the output buffers
+     */
+    public void flush() throws IOException {
+        _frameHandler.flush();
+    }
+
     private static final int negotiatedMaxValue(int clientValue, int serverValue) {
         return (clientValue == 0 || serverValue == 0) ?
             Math.max(clientValue, serverValue) :

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

@@ -71,6 +71,13 @@ public interface FrameHandler {
      */
     void writeFrame(Frame frame) throws IOException;
 
+    /**
+     * Flush the underlying data connection.
+     *
+     * @throws IOException if there is a problem accessing the connection
+     */
+    void flush() throws IOException;
+
     /** Close the underlying data connection (complaint not permitted). */
     void close();
 }

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

@@ -131,6 +131,7 @@ public void run() {
 
                 if (now > (lastActivityTime + this.heartbeatNanos)) {
                     frameHandler.writeFrame(new Frame(AMQP.FRAME_HEARTBEAT, 0));
+                    frameHandler.flush();
                 }
             } catch (IOException e) {
                 // ignore

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

@@ -135,12 +135,15 @@ public Frame readFrame() throws IOException {
     public void writeFrame(Frame frame) throws IOException {
         synchronized (_outputStream) {
             frame.writeTo(_outputStream);
-            _outputStream.flush();
         }
     }
 
+    public void flush() throws IOException {
+        _outputStream.flush();
+    }
+
     public void close() {
         try { _socket.setSoLinger(true, SOCKET_CLOSING_TIMEOUT); } catch (Exception _) {}
-        try { _socket.close();                                   } catch (Exception _) {}
+        try { flush(); _socket.close();                          } catch (Exception _) {}
     }
 }

test/src/com/rabbitmq/client/test/AMQConnectionTest.java

@@ -216,6 +216,10 @@ public InetAddress getAddress() {
         public int getPort() {
             return -1;
         }
+
+        public void flush() throws IOException {
+            // no need to implement this: don't bother writing the frame
+        }
     }
 
     /** Exception handler to facilitate testing. */

test/src/com/rabbitmq/client/test/BrokenFramesTest.java

@@ -173,6 +173,11 @@ public InetAddress getAddress() {
         public int getPort() {
             return -1;
         }
+
+        @Override
+        public void flush() throws IOException {
+            // no need to implement this: don't bother writing the frame
+        }
     }
 
 }