Merge pull request #104 from rabbitmq/rabbitmq-server-156

When sending methods with concent, flush socket stream once

Author: michaelklishin

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

@@ -42,6 +42,7 @@
 using RabbitMQ.Util;
 using System;
 using System.IO;
+using System.Collections.Generic;
 using System.Text;
 using System.Threading;
 using Windows.Foundation;
@@ -195,6 +196,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)

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

@@ -1122,6 +1122,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
@@ -73,6 +74,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,20 +43,17 @@
 using System;
 using System.IO;
 using System.Net;
+using System.Collections.Generic;
 using System.Net.Sockets;
 using System.Text;
 
 namespace RabbitMQ.Client.Impl
 {
     public class SocketFrameHandler : IFrameHandler
     {
-        // ^^ System.Net.Sockets.SocketError doesn't exist in .NET 1.1
-
         // Timeout in seconds to wait for a clean socket close.
         public const int SOCKET_CLOSING_TIMEOUT = 1;
 
-        public const int WSAEWOULDBLOCK = 10035;
-
         public NetworkBinaryReader m_reader;
         public TcpClient m_socket;
         public NetworkBinaryWriter m_writer;
@@ -226,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)