When sending methods with concent, flush socket stream once

References rabbitmq/rabbitmq-server#156.

Author: michaelklishin

projects/client/RabbitMQ.Client/src/client/impl/Command.cs

@@ -153,6 +153,18 @@ public byte[] ConsolidateBody()
         }
 
         public void Transmit(int channelNumber, Connection connection)
+        {
+            if(Method.HasContent)
+            {
+                TransmitAsFrameSet(channelNumber, connection);
+            }
+            else
+            {
+                TransmitAsSingleFrame(channelNumber, connection);
+            }
+        }
+
+        public void TransmitAsSingleFrame(int channelNumber, Connection connection)
         {
             var frame = new Frame(Constants.FrameMethod, channelNumber);
             NetworkBinaryWriter writer = frame.GetWriter();
@@ -162,6 +174,20 @@ public void Transmit(int channelNumber, Connection connection)
             Method.WriteArgumentsTo(argWriter);
             argWriter.Flush();
             connection.WriteFrame(frame);
+        }
+
+        public void TransmitAsFrameSet(int channelNumber, Connection connection)
+        {
+            var frame = new Frame(Constants.FrameMethod, channelNumber);
+            NetworkBinaryWriter writer = frame.GetWriter();
+            writer.Write((ushort)Method.ProtocolClassId);
+            writer.Write((ushort)Method.ProtocolMethodId);
+            var argWriter = new MethodArgumentWriter(writer);
+            Method.WriteArgumentsTo(argWriter);
+            argWriter.Flush();
+
+            var frames = new List<Frame>();
+            frames.Add(frame);
 
             if (Method.HasContent)
             {
@@ -171,7 +197,7 @@ public void Transmit(int channelNumber, Connection connection)
                 writer = frame.GetWriter();
                 writer.Write((ushort)Header.ProtocolClassId);
                 Header.WriteTo(writer, (ulong)body.Length);
-                connection.WriteFrame(frame);
+                frames.Add(frame);
 
                 var frameMax = (int)Math.Min(int.MaxValue, connection.FrameMax);
                 int bodyPayloadMax = (frameMax == 0)
@@ -185,9 +211,11 @@ public void Transmit(int channelNumber, Connection connection)
                     writer = frame.GetWriter();
                     writer.Write(body, offset,
                         (remaining < bodyPayloadMax) ? remaining : bodyPayloadMax);
-                    connection.WriteFrame(frame);
+                    frames.Add(frame);
                 }
             }
+
+            connection.WriteFrameSet(frames);
         }
     }
 }

projects/client/RabbitMQ.Client/src/client/impl/Connection.cs

@@ -1015,6 +1015,12 @@ public void WriteFrame(Frame f)
             m_heartbeatWrite.Set();
         }
 
+        public void WriteFrameSet(IList<Frame> f)
+        {
+            m_frameHandler.WriteFrameSet(f);
+            m_heartbeatWrite.Set();
+        }
+
         ///<summary>API-side invocation of connection abort.</summary>
         public void Abort()
         {

projects/client/RabbitMQ.Client/src/client/impl/IFrameHandler.cs

@@ -39,6 +39,7 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.Collections.Generic;
 using System.Net;
 
 namespace RabbitMQ.Client.Impl
@@ -69,6 +70,8 @@ public interface IFrameHandler
 
         void WriteFrame(Frame frame);
 
+        void WriteFrameSet(IList<Frame> frames);
+
         void Flush();
     }
 }

projects/client/RabbitMQ.Client/src/client/impl/SocketFrameHandler.cs

@@ -43,6 +43,7 @@
 using System;
 using System.IO;
 using System.Net;
+using System.Collections.Generic;
 using System.Net.Sockets;
 using System.Text;
 
@@ -222,6 +223,18 @@ public void WriteFrame(Frame frame)
             }
         }
 
+        public void WriteFrameSet(IList<Frame> frames)
+        {
+            lock (m_writer)
+            {
+                foreach(var f in frames)
+                {
+                    f.WriteTo(m_writer);
+                }
+                m_writer.Flush();
+            }
+        }
+
         public void Flush()
         {
             lock (m_writer)