Fix RabbitMQ consuming body reusing (#1727)

* Fix rabbitMQ consuming body reusing

* ArrayPool for memory management

* Rollback to simply array copy

---------

Co-authored-by: Pavel Dudnikov <[email protected]>

Author: otemnov

src/DotNetCore.CAP.RabbitMQ/RabbitMQBasicConsumer.cs

@@ -45,6 +45,8 @@ public override async Task HandleBasicDeliverAsync(string consumerTag, ulong del
         string routingKey, IReadOnlyBasicProperties properties, ReadOnlyMemory<byte> body,
         CancellationToken cancellationToken = default)
     {
+        var safeBody = _usingTaskRun ? body.ToArray() : body;
+
         if (_usingTaskRun)
         {
             await _semaphore.WaitAsync(cancellationToken);
@@ -73,15 +75,16 @@ Task Consume()
 
             if (_customHeadersBuilder != null)
             {
-                var e = new BasicDeliverEventArgs(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body);
+                var e = new BasicDeliverEventArgs(consumerTag, deliveryTag, redelivered, exchange, routingKey,
+                    properties, safeBody);
                 var customHeaders = _customHeadersBuilder(e, _serviceProvider);
                 foreach (var customHeader in customHeaders)
                 {
                     headers[customHeader.Key] = customHeader.Value;
                 }
             }
 
-            var message = new TransportMessage(headers, body);
+            var message = new TransportMessage(headers, safeBody);
 
             return _msgCallback(message, deliveryTag);
         }