flush after every command rather than every frame

Author: rade

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

@@ -183,6 +183,9 @@ public void transmit(AMQChannel channel) throws IOException {
                 connection.writeFrame(frame);
             }
         }
+
+        connection.flush();
+
     }
 
     @Override public String toString() {

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

@@ -409,6 +409,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 int negotiatedMaxValue(int clientValue, int serverValue) {
         return (clientValue == 0 || serverValue == 0) ?
             Math.max(clientValue, serverValue) :

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

@@ -72,6 +72,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).
      */

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

@@ -130,6 +130,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

@@ -152,12 +152,16 @@ 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 {
+            flush();
             _socket.close();
         } catch (IOException ioe) {
             // Ignore.