Recycling command buffers to reduce allocations when receiving messages.

Author: stebet

projects/client/RabbitMQ.Client/src/client/api/AsyncDefaultBasicConsumer.cs

@@ -113,7 +113,7 @@ public virtual Task HandleBasicDeliver(string consumerTag,
             string exchange,
             string routingKey,
             IBasicProperties properties,
-            byte[] body)
+            ReadOnlyMemory<byte> body)
         {
             // Nothing to do here.
             return TaskExtensions.CompletedTask;
@@ -167,7 +167,7 @@ void IBasicConsumer.HandleBasicConsumeOk(string consumerTag)
             throw new InvalidOperationException("Should never be called.");
         }
 
-        void IBasicConsumer.HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, byte[] body)
+        void IBasicConsumer.HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, ReadOnlyMemory<byte> body)
         {
             throw new InvalidOperationException("Should never be called.");
         }

projects/client/RabbitMQ.Client/src/client/api/BasicGetResult.cs

@@ -38,6 +38,8 @@
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
 
+using System;
+
 namespace RabbitMQ.Client
 {
     /// <summary>Represents Basic.GetOk responses from the server.</summary>
@@ -57,7 +59,7 @@ public class BasicGetResult
         /// <param name="basicProperties">The Basic-class content header properties for the message.</param>
         /// <param name="body"></param>
         public BasicGetResult(ulong deliveryTag, bool redelivered, string exchange,
-            string routingKey, uint messageCount, IBasicProperties basicProperties, byte[] body)
+            string routingKey, uint messageCount, IBasicProperties basicProperties, ReadOnlyMemory<byte> body)
         {
             DeliveryTag = deliveryTag;
             Redelivered = redelivered;
@@ -76,7 +78,7 @@ public BasicGetResult(ulong deliveryTag, bool redelivered, string exchange,
         /// <summary>
         /// Retrieves the body of this message.
         /// </summary>
-        public byte[] Body { get; private set; }
+        public ReadOnlyMemory<byte> Body { get; private set; }
 
         /// <summary>
         /// Retrieve the delivery tag for this message. See also <see cref="IModel.BasicAck"/>.

projects/client/RabbitMQ.Client/src/client/api/DefaultBasicConsumer.cs

@@ -161,7 +161,7 @@ public virtual void HandleBasicDeliver(string consumerTag,
             string exchange,
             string routingKey,
             IBasicProperties properties,
-            byte[] body)
+            ReadOnlyMemory<byte> body)
         {
             // Nothing to do here.
         }

projects/client/RabbitMQ.Client/src/client/api/IAsyncBasicConsumer.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Threading.Tasks;
 
 using RabbitMQ.Client.Events;
@@ -51,7 +52,7 @@ Task HandleBasicDeliver(string consumerTag,
             string exchange,
             string routingKey,
             IBasicProperties properties,
-            byte[] body);
+            ReadOnlyMemory<byte> body);
 
         /// <summary>
         ///  Called when the model shuts down.
@@ -60,4 +61,4 @@ Task HandleBasicDeliver(string consumerTag,
         /// <param name="reason"> Information about the reason why a particular model, session, or connection was destroyed.</param>
         Task HandleModelShutdown(object model, ShutdownEventArgs reason);
     }
-}
+}

projects/client/RabbitMQ.Client/src/client/api/IBasicConsumer.cs

@@ -104,7 +104,7 @@ void HandleBasicDeliver(string consumerTag,
             string exchange,
             string routingKey,
             IBasicProperties properties,
-            byte[] body);
+            ReadOnlyMemory<byte> body);
 
         /// <summary>
         ///  Called when the model shuts down.

projects/client/RabbitMQ.Client/src/client/events/AsyncEventingBasicConsumer.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Threading.Tasks;
 
 namespace RabbitMQ.Client.Events
@@ -37,7 +38,7 @@ public override async Task HandleBasicConsumeOk(string consumerTag)
         }
 
         ///<summary>Fires the Received event.</summary>
-        public override async Task HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, byte[] body)
+        public override async Task HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, ReadOnlyMemory<byte> body)
         {
             await base.HandleBasicDeliver(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body).ConfigureAwait(false);
             await (Received?.Invoke(

projects/client/RabbitMQ.Client/src/client/events/BasicDeliverEventArgs.cs

@@ -59,7 +59,7 @@ public BasicDeliverEventArgs(string consumerTag,
             string exchange,
             string routingKey,
             IBasicProperties properties,
-            byte[] body)
+            ReadOnlyMemory<byte> body)
         {
             ConsumerTag = consumerTag;
             DeliveryTag = deliveryTag;
@@ -74,7 +74,7 @@ public BasicDeliverEventArgs(string consumerTag,
         public IBasicProperties BasicProperties { get; set; }
 
         ///<summary>The message body.</summary>
-        public byte[] Body { get; set; }
+        public ReadOnlyMemory<byte> Body { get; set; }
 
         ///<summary>The consumer tag of the consumer that the message
         ///was delivered to.</summary>

projects/client/RabbitMQ.Client/src/client/events/BasicReturnEventArgs.cs

@@ -50,7 +50,7 @@ public class BasicReturnEventArgs : EventArgs
         public IBasicProperties BasicProperties { get; set; }
 
         ///<summary>The message body.</summary>
-        public byte[] Body { get; set; }
+        public ReadOnlyMemory<byte> Body { get; set; }
 
         ///<summary>The exchange the returned message was originally
         ///published to.</summary>

projects/client/RabbitMQ.Client/src/client/events/EventingBasicConsumer.cs

@@ -79,7 +79,7 @@ public override void HandleBasicConsumeOk(string consumerTag)
         }
 
         ///<summary>Fires the Received event.</summary>
-        public override void HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, byte[] body)
+        public override void HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, ReadOnlyMemory<byte> body)
         {
             base.HandleBasicDeliver(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body);
             Received?.Invoke(

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

@@ -1,4 +1,6 @@
-namespace RabbitMQ.Client.Impl
+using System;
+
+namespace RabbitMQ.Client.Impl
 {
     internal sealed class AsyncConsumerDispatcher : IConsumerDispatcher
     {
@@ -46,7 +48,7 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
             string exchange,
             string routingKey,
             IBasicProperties basicProperties,
-            byte[] body)
+            ReadOnlyMemory<byte> body)
         {
             ScheduleUnlessShuttingDown(new BasicDeliver(consumer, consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, body));
         }