Merge branch 'hotfix/2.5.1'

Author: MvRens

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,26 +24,23 @@ public TapetiBasicConsumer(IConsumer consumer, Func<ulong, ConsumeResult, Task>
 
 
         /// <inheritdoc />
-        public override void HandleBasicDeliver(string consumerTag,
+        public override async Task HandleBasicDeliver(string consumerTag,
             ulong deliveryTag,
             bool redelivered,
             string exchange,
             string routingKey,
             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);
+            }
         }
     }
 }

Tapeti/Connection/TapetiClient.cs

@@ -656,7 +656,8 @@ private IModel GetChannel()
                 Password = connectionParams.Password,
                 AutomaticRecoveryEnabled = false,
                 TopologyRecoveryEnabled = false,
-                RequestedHeartbeat = TimeSpan.FromSeconds(30)
+                RequestedHeartbeat = TimeSpan.FromSeconds(30),
+                DispatchConsumersAsync = true
             };
 
             if (connectionParams.ClientProperties != null)

Tapeti/Connection/TapetiConsumer.cs

@@ -39,12 +39,12 @@ public TapetiConsumer(CancellationToken cancellationToken, ITapetiConfig config,
 
 
         /// <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));
 

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 @@ public interface IConsumer
         /// <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);
     }
 }