Batch Publish

Batch publish allows sending multiple messages in one stream on the socket.
Sending in batches improves performance by reducing the number of TCP/IP messages sent and TCP/IP acknowledgments received.
This change compiles the commands for all publish messages into a memory buffer that is posted to the socket as a single stream.
Closing the socket/model/connection before the buffer has completed sending does not guarantee message delivery.

Author: YulerB

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

@@ -140,6 +140,8 @@ public void Transmit(int channelNumber, Connection connection)
             }
         }
 
+
+
         public void TransmitAsSingleFrame(int channelNumber, Connection connection)
         {
             connection.WriteFrame(new MethodOutboundFrame(channelNumber, Method));
@@ -166,5 +168,31 @@ public void TransmitAsFrameSet(int channelNumber, Connection connection)
 
             connection.WriteFrameSet(frames);
         }
+
+
+        public static IList<OutboundFrame> CalculateFrames(int channelNumber, Connection connection, IEnumerable<Command> commands)
+        {
+            List<OutboundFrame> frames = new List<Impl.OutboundFrame>();
+
+            foreach (var cmd in commands)
+            {
+                frames.Add(new MethodOutboundFrame(channelNumber, cmd.Method));
+                if (cmd.Method.HasContent)
+                {
+                    var body = cmd.Body;// var body = ConsolidateBody(); // Cache, since the property is compiled.
+
+                    frames.Add(new HeaderOutboundFrame(channelNumber, cmd.Header, body.Length));
+                    var frameMax = (int)Math.Min(int.MaxValue, connection.FrameMax);
+                    var bodyPayloadMax = (frameMax == 0) ? body.Length : frameMax - EmptyFrameSize;
+                    for (int offset = 0; offset < body.Length; offset += bodyPayloadMax)
+                    {
+                        var remaining = body.Length - offset;
+                        var count = (remaining < bodyPayloadMax) ? remaining : bodyPayloadMax;
+                        frames.Add(new BodySegmentOutboundFrame(channelNumber, body, offset, count));
+                    }
+                }
+            }
+            return frames;
+        }
     }
 }

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

@@ -39,7 +39,8 @@
 //---------------------------------------------------------------------------
 
 using System;
-
+using System.Collections.Generic;
+
 namespace RabbitMQ.Client.Impl
 {
     public interface ISession
@@ -79,5 +80,6 @@ public interface ISession
         void HandleFrame(InboundFrame frame);
         void Notify();
         void Transmit(Command cmd);
+        void Transmit(IEnumerable<Command> cmd);
     }
 }

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

@@ -472,6 +472,19 @@ public void ModelSend(MethodBase method, ContentHeaderBase header, byte[] body)
                 Session.Transmit(new Command(method, header, body));
             }
         }
+        public void ModelSend(MethodBase method, IEnumerable<BatchMessage> messages)
+        {
+            if (method.HasContent)
+            {
+                m_flowControlBlock.WaitOne();
+            }
+            List<Command> commands = new List<Impl.Command>();
+            foreach (var message in messages)
+            {
+                commands.Add(new Command(method, (ContentHeaderBase)message.basicProperties, message.Body));
+            }
+            Session.Transmit(commands);
+        }
 
         public virtual void OnBasicAck(BasicAckEventArgs args)
         {
@@ -1050,6 +1063,12 @@ public abstract void _Private_BasicPublish(string exchange,
             IBasicProperties basicProperties,
             byte[] body);
 
+        //public abstract void _Private_BasicBatchPublish(string exchange,
+        //    string routingKey,
+        //    bool mandatory,
+        //    IEnumerable<BatchMessage> messages);
+
+
         public abstract void _Private_BasicRecover(bool requeue);
 
         public abstract void _Private_ChannelClose(ushort replyCode,
@@ -1231,8 +1250,52 @@ public void BasicPublish(string exchange,
                 mandatory,
                 basicProperties,
                 body);
-        }
+        }
+
+        public void BasicBatchPublish(string exchange,
+    string routingKey,
+    bool mandatory,
+    IEnumerable<BatchMessage> messages)
+        {
+            foreach (var message in messages)
+            {
+                if (message.basicProperties == null)
+                {
+                    message.basicProperties = CreateBasicProperties();
+                }
+
+                if (NextPublishSeqNo > 0)
+                {
+                    lock (m_unconfirmedSet.SyncRoot)
+                    {
+                        if (!m_unconfirmedSet.Contains(NextPublishSeqNo))
+                        {
+                            m_unconfirmedSet.Add(NextPublishSeqNo);
+                        }
+                        NextPublishSeqNo++;
+                    }
+                }
+            }
 
+            _Private_BasicBatchPublish(exchange,
+                routingKey,
+                mandatory,
+                messages);
+        }
+        public void _Private_BasicBatchPublish(
+string @exchange,
+string @routingKey,
+bool @mandatory,
+//bool @immediate,
+IEnumerable<BatchMessage> messages)
+        {
+            BasicPublish __req = new BasicPublish();
+            __req.m_exchange = @exchange;
+            __req.m_routingKey = @routingKey;
+            __req.m_mandatory = @mandatory;
+            //__req.m_immediate = @immediate;
+            ModelSend(__req, messages);
+        }
         public abstract void BasicQos(uint prefetchSize,
             ushort prefetchCount,
             bool global);

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

@@ -41,7 +41,8 @@
 using System;
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Framing.Impl;
-
+using System.Collections.Generic;
+
 namespace RabbitMQ.Client.Impl
 {
     public abstract class SessionBase : ISession
@@ -199,5 +200,9 @@ public virtual void Transmit(Command cmd)
             // of frames within a channel.  But that is fixed in socket frame handler instead, so no need to lock.
             cmd.Transmit(ChannelNumber, Connection);
         }
+        public virtual void Transmit(IEnumerable<Command> commands)
+        {
+            Connection.WriteFrameSet(Command.CalculateFrames(ChannelNumber, Connection, commands));
+        }
     }
 }