GitHub-Mirror/Tapeti
1
0
Fork 0
mirror of synced 2024-11-05 02:59:16 +00:00
Code Issues Projects Releases Wiki Activity

Fixed #29: Deserialization errors due to not buffering RabbitMQ Client's body

Browse Source 
- TIL there is an AsyncDefaultBasicConsumer which eliminates the need for the Task.Run
This commit is contained in:
Mark van Renswoude 2021-06-25 09:28:25 +02:00
parent 21a47ea6f5
commit e33201dc17
3 changed files with 15 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
Tapeti/Connection/TapetiBasicConsumer.cs
View File

@ -9,7 +9,7 @@ namespace Tapeti.Connection
/// <summary> /// <summary>
/// Implements the bridge between the RabbitMQ Client consumer and a Tapeti Consumer /// Implements the bridge between the RabbitMQ Client consumer and a Tapeti Consumer
/// </summary> /// </summary>
internal class TapetiBasicConsumer : DefaultBasicConsumer internal class TapetiBasicConsumer : AsyncDefaultBasicConsumer
{ {
private readonly IConsumer consumer; private readonly IConsumer consumer;
private readonly Func<ulong, ConsumeResult, Task> onRespond; private readonly Func<ulong, ConsumeResult, Task> onRespond;
@ -24,26 +24,23 @@ namespace Tapeti.Connection
/// <inheritdoc /> /// <inheritdoc />
public override void HandleBasicDeliver(string consumerTag, public override async Task HandleBasicDeliver(string consumerTag,
ulong deliveryTag, ulong deliveryTag,
bool redelivered, bool redelivered,
string exchange, string exchange,
string routingKey, string routingKey,
IBasicProperties properties, IBasicProperties properties,
ReadOnlyMemory<byte> body) ReadOnlyMemory<byte> body)
{
Task.Run(async () =>
{ {
try try
{ {
var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), body.ToArray()); var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), body);
await onRespond(deliveryTag, response); await onRespond(deliveryTag, response);
} }
catch catch
{ {
await onRespond(deliveryTag, ConsumeResult.Error); await onRespond(deliveryTag, ConsumeResult.Error);
} }
});
} }
} }
} }

4
Tapeti/Connection/TapetiConsumer.cs
View File

@ -39,12 +39,12 @@ namespace Tapeti.Connection
/// <inheritdoc /> /// <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; object message = null;
try try
{ {
message = messageSerializer.Deserialize(body, properties); message = messageSerializer.Deserialize(body.ToArray(), properties);
if (message == null) if (message == null)
throw new ArgumentException("Message body could not be deserialized into a message object", nameof(body)); throw new ArgumentException("Message body could not be deserialized into a message object", nameof(body));

5
Tapeti/IConsumer.cs
View File

@ -1,4 +1,5 @@
using System.Threading.Tasks; using System;
using System.Threading.Tasks;
using Tapeti.Config; using Tapeti.Config;
namespace Tapeti namespace Tapeti
@ -16,6 +17,6 @@ namespace Tapeti
/// <param name="properties">Metadata included in the message</param> /// <param name="properties">Metadata included in the message</param>
/// <param name="body">The raw body of the message</param> /// <param name="body">The raw body of the message</param>
/// <returns></returns> /// <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);
} }
} }