From e33201dc17d1d540c21aaae93a13aec582994094 Mon Sep 17 00:00:00 2001
From: Mark van Renswoude <mark@x2software.net>
Date: Fri, 25 Jun 2021 09:28:25 +0200
Subject: [PATCH 1/2] Fixed #29: Deserialization errors due to not buffering
 RabbitMQ Client's body

- TIL there is an AsyncDefaultBasicConsumer which eliminates the need for the Task.Run
---
 Tapeti/Connection/TapetiBasicConsumer.cs | 23 ++++++++++-------------
 Tapeti/Connection/TapetiConsumer.cs      |  4 ++--
 Tapeti/IConsumer.cs                      |  5 +++--
 3 files changed, 15 insertions(+), 17 deletions(-)

diff --git a/Tapeti/Connection/TapetiBasicConsumer.cs b/Tapeti/Connection/TapetiBasicConsumer.cs
index 7dd969a..1cc8e92 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 : DefaultBasicConsumer
+    internal class TapetiBasicConsumer : AsyncDefaultBasicConsumer
     {
         private readonly IConsumer consumer;
         private readonly Func<ulong, ConsumeResult, Task> onRespond;
@@ -24,7 +24,7 @@ namespace Tapeti.Connection
 
 
         /// <inheritdoc />
-        public override void HandleBasicDeliver(string consumerTag,
+        public override async Task HandleBasicDeliver(string consumerTag,
             ulong deliveryTag,
             bool redelivered,
             string exchange,
@@ -32,18 +32,15 @@ namespace Tapeti.Connection
             IBasicProperties properties,
             ReadOnlyMemory<byte> body)
         {
-            Task.Run(async () =>
+            try
             {
-                try
-                {
-                    var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), body.ToArray());
-                    await onRespond(deliveryTag, response);
-                }
-                catch
-                {
-                    await onRespond(deliveryTag, ConsumeResult.Error);
-                }
-            });
+                var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), body);
+                await onRespond(deliveryTag, response);
+            }
+            catch
+            {
+                await onRespond(deliveryTag, ConsumeResult.Error);
+            }
         }
     }
 }
diff --git a/Tapeti/Connection/TapetiConsumer.cs b/Tapeti/Connection/TapetiConsumer.cs
index 19fb870..ddbb4d7 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, byte[] body)
+        public async Task<ConsumeResult> Consume(string exchange, string routingKey, IMessageProperties properties, ReadOnlyMemory<byte> body)
         {
             object message = null;
             try
             {
-                message = messageSerializer.Deserialize(body, properties);
+                message = messageSerializer.Deserialize(body.ToArray(), 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 7204bc5..cd20f30 100644
--- a/Tapeti/IConsumer.cs
+++ b/Tapeti/IConsumer.cs
@@ -1,4 +1,5 @@
-using System.Threading.Tasks;
+using System;
+using System.Threading.Tasks;
 using Tapeti.Config;
 
 namespace Tapeti
@@ -16,6 +17,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, byte[] body);
+        Task<ConsumeResult> Consume(string exchange, string routingKey, IMessageProperties properties, ReadOnlyMemory<byte> body);
     }
 }

From bbfb580a97fa2e4ab43c3934643db08574e9871d Mon Sep 17 00:00:00 2001
From: Mark van Renswoude <mark@x2software.net>
Date: Fri, 25 Jun 2021 09:35:26 +0200
Subject: [PATCH 2/2] Actually fixed #29

---
 Tapeti/Connection/TapetiClient.cs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/Tapeti/Connection/TapetiClient.cs b/Tapeti/Connection/TapetiClient.cs
index c41e7b4..d30f512 100644
--- a/Tapeti/Connection/TapetiClient.cs
+++ b/Tapeti/Connection/TapetiClient.cs
@@ -656,7 +656,8 @@ namespace Tapeti.Connection
                 Password = connectionParams.Password,
                 AutomaticRecoveryEnabled = false,
                 TopologyRecoveryEnabled = false,
-                RequestedHeartbeat = TimeSpan.FromSeconds(30)
+                RequestedHeartbeat = TimeSpan.FromSeconds(30),
+                DispatchConsumersAsync = true
             };
 
             if (connectionParams.ClientProperties != null)