Merge pull request #1028 from bollhals/feature/delete.batch

Remove batch publishing as its optimizations are now performed for "regular" publishes/outgoing frames

Author: michaelklishin

projects/RabbitMQ.Client/client/api/IBasicPublishBatch.cs

@@ -229,16 +229,6 @@ string BasicConsume(
         /// </summary>
         void ConfirmSelect();
 
-        /// <summary>
-        ///  Creates a BasicPublishBatch instance
-        /// </summary>
-        IBasicPublishBatch CreateBasicPublishBatch();
-
-        /// <summary>
-        ///  Creates a BasicPublishBatch instance
-        /// </summary>
-        IBasicPublishBatch CreateBasicPublishBatch(int sizeHint);
-
         /// <summary>
         /// Construct a completely empty content header for use with the Basic content class.
         /// </summary>

projects/RabbitMQ.Client/client/api/IModel.cs

@@ -420,12 +420,6 @@ private void RecoverState()
             }
         }
 
-        public IBasicPublishBatch CreateBasicPublishBatch()
-            => InnerChannel.CreateBasicPublishBatch();
-
-        public IBasicPublishBatch CreateBasicPublishBatch(int sizeHint)
-            => InnerChannel.CreateBasicPublishBatch(sizeHint);
-
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void ThrowIfDisposed()
         {

projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs

@@ -30,8 +30,6 @@
 //---------------------------------------------------------------------------
 
 using System;
-using System.Collections.Generic;
-
 using RabbitMQ.Client.Framing.Impl;
 
 namespace RabbitMQ.Client.Impl
@@ -75,6 +73,5 @@ internal interface ISession
         bool HandleFrame(in InboundFrame frame);
         void Notify();
         void Transmit<T>(in T cmd) where T : struct, IOutgoingCommand;
-        void Transmit<T>(List<T> cmds) where T : struct, IOutgoingCommand;
     }
 }

projects/RabbitMQ.Client/client/impl/BasicPublishBatch.cs

@@ -851,17 +851,6 @@ public BasicGetResult BasicGet(string queue, bool autoAck)
 
         public abstract void BasicNack(ulong deliveryTag, bool multiple, bool requeue);
 
-        private void AllocatePublishSeqNos(int count)
-        {
-            lock (_confirmLock)
-            {
-                for (int i = 0; i < count; i++)
-                {
-                    _pendingDeliveryTags.AddLast(NextPublishSeqNo++);
-                }
-            }
-        }
-
         public void BasicPublish(string exchange, string routingKey, bool mandatory, IBasicProperties basicProperties, ReadOnlyMemory<byte> body)
         {
             if (routingKey is null)
@@ -941,19 +930,7 @@ public void ConfirmSelect()
             _Private_ConfirmSelect(false);
         }
 
-        ///////////////////////////////////////////////////////////////////////////
-
         public abstract IBasicProperties CreateBasicProperties();
-        public IBasicPublishBatch CreateBasicPublishBatch()
-        {
-            return new BasicPublishBatch(this);
-        }
-
-        public IBasicPublishBatch CreateBasicPublishBatch(int sizeHint)
-        {
-            return new BasicPublishBatch(this, sizeHint);
-        }
-
 
         public void ExchangeBind(string destination, string source, string routingKey, IDictionary<string, object> arguments)
         {
@@ -1143,16 +1120,6 @@ await CloseAsync(new ShutdownEventArgs(ShutdownInitiator.Application,
             }
         }
 
-        internal void SendCommands(List<OutgoingContentCommand> commands)
-        {
-            _flowControlBlock.Wait();
-            if (NextPublishSeqNo > 0)
-            {
-                AllocatePublishSeqNos(commands.Count);
-            }
-            Session.Transmit(commands);
-        }
-
         private QueueDeclareOk QueueDeclare(string queue, bool passive, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments)
         {
             var k = new QueueDeclareRpcContinuation();

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

@@ -30,7 +30,6 @@
 //---------------------------------------------------------------------------
 
 using System;
-using System.Collections.Generic;
 using System.Threading;
 
 using RabbitMQ.Client.Exceptions;
@@ -136,14 +135,5 @@ public virtual void Transmit<T>(in T cmd) where T : struct, IOutgoingCommand
             // of frames within a channel.  But that is fixed in socket frame handler instead, so no need to lock.
             Connection.Write(cmd.SerializeToFrames(ChannelNumber, Connection.FrameMax));
         }
-
-        public virtual void Transmit<T>(List<T> cmds) where T : struct, IOutgoingCommand
-        {
-            uint frameMax = Connection.FrameMax;
-            for (int i = 0; i < cmds.Count; i++)
-            {
-                Connection.Write(cmds[i].SerializeToFrames(ChannelNumber, frameMax));
-            }
-        }
     }
 }

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

@@ -42,15 +42,13 @@ public static void Main()
             {
                 while (messagesSent < BatchesToSend * ItemsPerBatch)
                 {
-                    var batch = publisher.CreateBasicPublishBatch();
                     for (int i = 0; i < ItemsPerBatch; i++)
                     {
                         var properties = publisher.CreateBasicProperties();
                         properties.AppId = "testapp";
                         properties.CorrelationId = Guid.NewGuid().ToString();
-                        batch.Add("test", "myawesome.routing.key", false, properties, payload);
+                        publisher.BasicPublish("test", "myawesome.routing.key", false, properties, payload);
                     }
-                    batch.Publish();
                     messagesSent += ItemsPerBatch;
                     await publisher.WaitForConfirmsOrDieAsync().ConfigureAwait(false);
                 }

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

@@ -297,12 +297,6 @@ namespace RabbitMQ.Client
         bool IsTypePresent();
         bool IsUserIdPresent();
     }
-    public interface IBasicPublishBatch
-    {
-        void Add(RabbitMQ.Client.CachedString exchange, RabbitMQ.Client.CachedString routingKey, bool mandatory, RabbitMQ.Client.IBasicProperties basicProperties, System.ReadOnlyMemory<byte> body);
-        void Add(string exchange, string routingKey, bool mandatory, RabbitMQ.Client.IBasicProperties properties, System.ReadOnlyMemory<byte> body);
-        void Publish();
-    }
     public interface IConnection : RabbitMQ.Client.INetworkConnection, System.IDisposable
     {
         ushort ChannelMax { get; }
@@ -403,8 +397,6 @@ namespace RabbitMQ.Client
         void ConfirmSelect();
         uint ConsumerCount(string queue);
         RabbitMQ.Client.IBasicProperties CreateBasicProperties();
-        RabbitMQ.Client.IBasicPublishBatch CreateBasicPublishBatch();
-        RabbitMQ.Client.IBasicPublishBatch CreateBasicPublishBatch(int sizeHint);
         void ExchangeBind(string destination, string source, string routingKey, System.Collections.Generic.IDictionary<string, object> arguments);
         void ExchangeBindNoWait(string destination, string source, string routingKey, System.Collections.Generic.IDictionary<string, object> arguments);
         void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, System.Collections.Generic.IDictionary<string, object> arguments);