Fixing stuff.

Author: stebet

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

@@ -1,4 +1,5 @@
 using System;
+using System.Buffers;
 using System.Threading.Tasks;
 
 namespace RabbitMQ.Client.Impl
@@ -46,7 +47,9 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
             IBasicProperties basicProperties,
             ReadOnlyMemory<byte> body)
         {
-            ScheduleUnlessShuttingDown(new BasicDeliver(consumer, consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, body));
+            IMemoryOwner<byte> bodyCopy = MemoryPool<byte>.Shared.Rent(body.Length);
+            body.CopyTo(bodyCopy.Memory);
+            ScheduleUnlessShuttingDown(new BasicDeliver(consumer, consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, bodyCopy, body.Length));
         }
 
         public void HandleBasicCancelOk(IBasicConsumer consumer, string consumerTag)

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

@@ -1,4 +1,5 @@
 using System;
+using System.Buffers;
 using System.Collections.Generic;
 using System.Threading.Tasks;
 
@@ -14,7 +15,8 @@ sealed class BasicDeliver : Work
         readonly string _exchange;
         readonly string _routingKey;
         readonly IBasicProperties _basicProperties;
-        readonly ReadOnlyMemory<byte> _body;
+        readonly IMemoryOwner<byte> _body;
+        readonly int _bodyLength;
 
         public BasicDeliver(IBasicConsumer consumer, 
             string consumerTag, 
@@ -23,7 +25,8 @@ public BasicDeliver(IBasicConsumer consumer,
             string exchange, 
             string routingKey, 
             IBasicProperties basicProperties,
-            ReadOnlyMemory<byte> body) : base(consumer)
+            IMemoryOwner<byte> body,
+            int bodyLength) : base(consumer)
         {
             _consumerTag = consumerTag;
             _deliveryTag = deliveryTag;
@@ -32,6 +35,7 @@ public BasicDeliver(IBasicConsumer consumer,
             _routingKey = routingKey;
             _basicProperties = basicProperties;
             _body = body;
+            _bodyLength = bodyLength;
         }
 
         protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
@@ -44,7 +48,7 @@ await consumer.HandleBasicDeliver(_consumerTag,
                     _exchange,
                     _routingKey,
                     _basicProperties,
-                    _body).ConfigureAwait(false);
+                    _body.Memory.Slice(0, _bodyLength)).ConfigureAwait(false);
             }
             catch (Exception e)
             {
@@ -55,6 +59,10 @@ await consumer.HandleBasicDeliver(_consumerTag,
                 };
                 model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
+            finally
+            {
+                _body.Dispose();
+            }
         }
     }
 }

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

@@ -1,4 +1,5 @@
 using System;
+using System.Buffers;
 using System.Collections.Generic;
 using System.Threading.Tasks;
 using RabbitMQ.Client.Events;
@@ -63,6 +64,8 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
                                        IBasicProperties basicProperties,
                                        ReadOnlyMemory<byte> body)
         {
+            IMemoryOwner<byte> memoryCopy = MemoryPool<byte>.Shared.Rent(body.Length);
+            body.CopyTo(memoryCopy.Memory);
             UnlessShuttingDown(() =>
             {
                 try
@@ -73,7 +76,7 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
                                                 exchange,
                                                 routingKey,
                                                 basicProperties,
-                                                body);
+                                                memoryCopy.Memory.Slice(0, body.Length));
                 }
                 catch (Exception e)
                 {
@@ -84,6 +87,10 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
                     };
                     _model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
                 }
+                finally
+                {
+                    memoryCopy.Dispose();
+                }
             });
         }