From 09efa161b4021e39716e270175f75e5e1e813a99 Mon Sep 17 00:00:00 2001
From: Mark van Renswoude <mark@x2software.net>
Date: Fri, 25 Jun 2021 09:35:35 +0200
Subject: [PATCH] Revert "Merge branch 'hotfix/2.5.1' into develop"

This reverts commit 4b2ced7956b29e0bfd4f794a92356fa75999f7d9.
---
 Tapeti/Connection/TapetiBasicConsumer.cs | 23 +++++++++++++----------
 Tapeti/Connection/TapetiClient.cs        |  3 +--
 Tapeti/Connection/TapetiConsumer.cs      |  4 ++--
 Tapeti/IConsumer.cs                      |  5 ++---
 4 files changed, 18 insertions(+), 17 deletions(-)

diff --git a/Tapeti/Connection/TapetiBasicConsumer.cs b/Tapeti/Connection/TapetiBasicConsumer.cs
index 1cc8e92..7dd969a 100644
--- a/Tapeti/Connection/TapetiBasicConsumer.cs
+++ b/Tapeti/Connection/TapetiBasicConsumer.cs
@@ -9,7 +9,7 @@ namespace Tapeti.Connection
     /// <summary>
     /// Implements the bridge between the RabbitMQ Client consumer and a Tapeti Consumer
     /// </summary>
-    internal class TapetiBasicConsumer : AsyncDefaultBasicConsumer
+    internal class TapetiBasicConsumer : DefaultBasicConsumer
     {
         private readonly IConsumer consumer;
         private readonly Func<ulong, ConsumeResult, Task> onRespond;
@@ -24,7 +24,7 @@ namespace Tapeti.Connection
 
 
         /// <inheritdoc />
-        public override async Task HandleBasicDeliver(string consumerTag,
+        public override void HandleBasicDeliver(string consumerTag,
             ulong deliveryTag,
             bool redelivered,
             string exchange,
@@ -32,15 +32,18 @@ namespace Tapeti.Connection
             IBasicProperties properties,
             ReadOnlyMemory<byte> body)
         {
-            try
+            Task.Run(async () =>
             {
-                var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), body);
-                await onRespond(deliveryTag, response);
-            }
-            catch
-            {
-                await onRespond(deliveryTag, ConsumeResult.Error);
-            }
+                try
+                {
+                    var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), body.ToArray());
+                    await onRespond(deliveryTag, response);
+                }
+                catch
+                {
+                    await onRespond(deliveryTag, ConsumeResult.Error);
+                }
+            });
         }
     }
 }
diff --git a/Tapeti/Connection/TapetiClient.cs b/Tapeti/Connection/TapetiClient.cs
index d30f512..c41e7b4 100644
--- a/Tapeti/Connection/TapetiClient.cs
+++ b/Tapeti/Connection/TapetiClient.cs
@@ -656,8 +656,7 @@ namespace Tapeti.Connection
                 Password = connectionParams.Password,
                 AutomaticRecoveryEnabled = false,
                 TopologyRecoveryEnabled = false,
-                RequestedHeartbeat = TimeSpan.FromSeconds(30),
-                DispatchConsumersAsync = true
+                RequestedHeartbeat = TimeSpan.FromSeconds(30)
             };
 
             if (connectionParams.ClientProperties != null)
diff --git a/Tapeti/Connection/TapetiConsumer.cs b/Tapeti/Connection/TapetiConsumer.cs
index ddbb4d7..19fb870 100644
--- a/Tapeti/Connection/TapetiConsumer.cs
+++ b/Tapeti/Connection/TapetiConsumer.cs
@@ -39,12 +39,12 @@ namespace Tapeti.Connection
 
 
         /// <inheritdoc />
-        public async Task<ConsumeResult> Consume(string exchange, string routingKey, IMessageProperties properties, ReadOnlyMemory<byte> body)
+        public async Task<ConsumeResult> Consume(string exchange, string routingKey, IMessageProperties properties, byte[] body)
         {
             object message = null;
             try
             {
-                message = messageSerializer.Deserialize(body.ToArray(), properties);
+                message = messageSerializer.Deserialize(body, properties);
                 if (message == null)
                     throw new ArgumentException("Message body could not be deserialized into a message object", nameof(body));
 
diff --git a/Tapeti/IConsumer.cs b/Tapeti/IConsumer.cs
index cd20f30..7204bc5 100644
--- a/Tapeti/IConsumer.cs
+++ b/Tapeti/IConsumer.cs
@@ -1,5 +1,4 @@
-using System;
-using System.Threading.Tasks;
+using System.Threading.Tasks;
 using Tapeti.Config;
 
 namespace Tapeti
@@ -17,6 +16,6 @@ namespace Tapeti
         /// <param name="properties">Metadata included in the message</param>
         /// <param name="body">The raw body of the message</param>
         /// <returns></returns>
-        Task<ConsumeResult> Consume(string exchange, string routingKey, IMessageProperties properties, ReadOnlyMemory<byte> body);
+        Task<ConsumeResult> Consume(string exchange, string routingKey, IMessageProperties properties, byte[] body);
     }
 }