Merge branch 'hotfix/2.5.1' into develop
This commit is contained in:
commit
4b2ced7956
@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -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)
|
||||
|
@ -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));
|
||||
|
||||
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user