MessageBatch -> BasicPublishBatch

BasicPublishBatch can be used for more efficient batch publishing that will better utilise TCP and provides more throughput. [#152041636]
rabbitmq · Nov 22, 2017 · 74c9c24 · 74c9c24
1 parent c020263
commit 74c9c24
Show file tree

Hide file tree

Showing 10 changed files with 38 additions and 37 deletions.
diff --git a/projects/client/ApigenBootstrap/ApigenBootstrap.csproj b/projects/client/ApigenBootstrap/ApigenBootstrap.csproj
@@ -15,7 +15,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <Compile Include="..\RabbitMQ.Client\src\client\api\AmqpTimestamp.cs;..\RabbitMQ.Client\src\client\api\IBasicConsumer.cs;..\RabbitMQ.Client\src\client\api\IBasicProperties.cs;..\RabbitMQ.Client\src\client\api\IContentHeader.cs;..\RabbitMQ.Client\src\client\api\IModel.cs;..\RabbitMQ.Client\src\client\api\PublicationAddress.cs;..\RabbitMQ.Client\src\client\api\IMessageBatch.cs;..\RabbitMQ.Client\src\client\api\BasicGetResult.cs;..\RabbitMQ.Client\src\client\api\QueueDeclareOk.cs;..\RabbitMQ.Client\src\client\api\ShutdownEventArgs.cs;..\RabbitMQ.Client\src\client\api\ShutdownInitiator.cs;..\RabbitMQ.Client\src\client\events\BasicReturnEventArgs.cs;..\RabbitMQ.Client\src\client\events\BasicAckEventArgs.cs;..\RabbitMQ.Client\src\client\events\BasicNackEventArgs.cs;..\RabbitMQ.Client\src\client\events\CallbackExceptionEventArgs.cs;..\RabbitMQ.Client\src\client\events\ConsumerEventArgs.cs;..\RabbitMQ.Client\src\client\events\FlowControlEventArgs.cs;..\RabbitMQ.Client\src\client\impl\IFullModel.cs" />
+    <Compile Include="..\RabbitMQ.Client\src\client\api\AmqpTimestamp.cs;..\RabbitMQ.Client\src\client\api\IBasicConsumer.cs;..\RabbitMQ.Client\src\client\api\IBasicProperties.cs;..\RabbitMQ.Client\src\client\api\IContentHeader.cs;..\RabbitMQ.Client\src\client\api\IModel.cs;..\RabbitMQ.Client\src\client\api\PublicationAddress.cs;..\RabbitMQ.Client\src\client\api\IBasicPublishBatch.cs;..\RabbitMQ.Client\src\client\api\BasicGetResult.cs;..\RabbitMQ.Client\src\client\api\QueueDeclareOk.cs;..\RabbitMQ.Client\src\client\api\ShutdownEventArgs.cs;..\RabbitMQ.Client\src\client\api\ShutdownInitiator.cs;..\RabbitMQ.Client\src\client\events\BasicReturnEventArgs.cs;..\RabbitMQ.Client\src\client\events\BasicAckEventArgs.cs;..\RabbitMQ.Client\src\client\events\BasicNackEventArgs.cs;..\RabbitMQ.Client\src\client\events\CallbackExceptionEventArgs.cs;..\RabbitMQ.Client\src\client\events\ConsumerEventArgs.cs;..\RabbitMQ.Client\src\client\events\FlowControlEventArgs.cs;..\RabbitMQ.Client\src\client\impl\IFullModel.cs" />
   </ItemGroup>
 
 </Project>
diff --git a/...MQ.Client/src/client/api/IMessageBatch.cs → ...ient/src/client/api/IBasicPublishBatch.cs b/...MQ.Client/src/client/api/IMessageBatch.cs → ...ient/src/client/api/IBasicPublishBatch.cs
@@ -39,7 +39,7 @@
 //---------------------------------------------------------------------------
 namespace RabbitMQ.Client
 {
-    public interface IMessageBatch
+    public interface IBasicPublishBatch
     {
         void Add(string exchange, string routingKey, bool mandatory, IBasicProperties properties, byte[] body);
         void Publish();

diff --git a/projects/client/RabbitMQ.Client/src/client/api/IModel.cs b/projects/client/RabbitMQ.Client/src/client/api/IModel.cs
@@ -280,10 +280,10 @@ void BasicPublish(string exchange, string routingKey, bool mandatory,
         void ConfirmSelect();
 
         /// <summary>
-        ///  Creates a MessagBatch instance
+        ///  Creates a BasicPublishBatch instance
         /// </summary>
         [AmqpMethodDoNotImplement(null)]
-        IMessageBatch CreateMessageBatch();
+        IBasicPublishBatch CreateBasicPublishBatch();
 
         /// <summary>
         /// Construct a completely empty content header for use with the Basic content class.

diff --git a/projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringModel.cs b/projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringModel.cs
@@ -1213,9 +1213,9 @@ protected void RunRecoveryEventHandlers()
             }
         }
 
-        public IMessageBatch CreateMessageBatch()
+        public IBasicPublishBatch CreateBasicPublishBatch()
         {
-            return ((IFullModel)m_delegate).CreateMessageBatch();
+            return ((IFullModel)m_delegate).CreateBasicPublishBatch();
         }
     }
 }
diff --git a/...MQ.Client/src/client/impl/MessageBatch.cs → ...ient/src/client/impl/BasicPublishBatch.cs b/...MQ.Client/src/client/impl/MessageBatch.cs → ...ient/src/client/impl/BasicPublishBatch.cs
@@ -45,11 +45,11 @@ namespace RabbitMQ.Client.Impl
     using RabbitMQ.Client.Framing.Impl;
     using RabbitMQ.Client.Impl;
 
-    public class MessageBatch : IMessageBatch
+    public class BasicPublishBatch : IBasicPublishBatch
     {
         private List<Command> commands = new List<Command>();
         private ModelBase model;
-        internal MessageBatch(ModelBase model)
+        internal BasicPublishBatch (ModelBase model)
         {
             this.model = model;
         }

diff --git a/projects/client/RabbitMQ.Client/src/client/impl/Command.cs b/projects/client/RabbitMQ.Client/src/client/impl/Command.cs
@@ -170,28 +170,29 @@ public void TransmitAsFrameSet(int channelNumber, Connection connection)
         }
 
 
-        public static IList<OutboundFrame> CalculateFrames(int channelNumber, Connection connection, IEnumerable<Command> commands)
+        public static List<OutboundFrame> CalculateFrames(int channelNumber, Connection connection, IList<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));
-                    }
-                }
+            var frames = new List<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;
         }
     }

diff --git a/projects/client/RabbitMQ.Client/src/client/impl/ISession.cs b/projects/client/RabbitMQ.Client/src/client/impl/ISession.cs
@@ -39,8 +39,8 @@
 //---------------------------------------------------------------------------
 
 using System;
-using System.Collections.Generic;
-
+using System.Collections.Generic;
+
 namespace RabbitMQ.Client.Impl
 {
     public interface ISession
@@ -80,6 +80,6 @@ public interface ISession
         void HandleFrame(InboundFrame frame);
         void Notify();
         void Transmit(Command cmd);
-        void Transmit(IEnumerable<Command> cmd);
+        void Transmit(IList<Command> cmd);
     }
 }
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs b/projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs
@@ -1293,9 +1293,9 @@ public void ConfirmSelect()
         ///////////////////////////////////////////////////////////////////////////
 
         public abstract IBasicProperties CreateBasicProperties();
-        public IMessageBatch CreateMessageBatch()
+        public IBasicPublishBatch CreateBasicPublishBatch()
         {
-            return new MessageBatch(this);
+            return new BasicPublishBatch(this);
         }
 
 

diff --git a/projects/client/RabbitMQ.Client/src/client/impl/SessionBase.cs b/projects/client/RabbitMQ.Client/src/client/impl/SessionBase.cs
@@ -200,7 +200,7 @@ 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)
+        public virtual void Transmit(IList<Command> commands)
         {
             Connection.WriteFrameSet(Command.CalculateFrames(ChannelNumber, Connection, commands));
         }

diff --git a/.../client/Unit/src/unit/TestMessageBatch.cs → ...nt/Unit/src/unit/TestBasicPublishBatch.cs b/.../client/Unit/src/unit/TestMessageBatch.cs → ...nt/Unit/src/unit/TestBasicPublishBatch.cs
@@ -45,14 +45,14 @@
 
 namespace RabbitMQ.Client.Unit
 {
-    internal class TestMessageBatch : IntegrationFixture
+    internal class TestBasicPublishBatch : IntegrationFixture
     {
         [Test]
-        public void TestMessageBatchSend()
+        public void TestBasicPublishBatchSend()
         {
             Model.QueueDeclare(queue: "test-message-batch-a", durable: false);
             Model.QueueDeclare(queue: "test-message-batch-b", durable: false);
-            var batch = Model.CreateMessageBatch();
+            var batch = Model.CreateBasicPublishBatch();
             batch.Add("", "test-message-batch-a", false, null, new byte [] {});
             batch.Add("", "test-message-batch-b", false, null, new byte [] {});
             batch.Publish();