GitHub-Mirror
/
Tapeti
mirror of
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Reverted async changes as they caused message handlers to no longer run in parallel

This commit is contained in:
Mark van Renswoude 2023-04-06 07:25:01 +02:00
parent 1bd284166d
commit ab2cc1c1bb
2 changed files with 17 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
Tapeti/Connection/TapetiBasicConsumer.cs
View File

@ -17,7 +17,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 : AsyncDefaultBasicConsumer internal class TapetiBasicConsumer : DefaultBasicConsumer
{ {
private readonly IConsumer consumer; private readonly IConsumer consumer;
private readonly long connectionReference; private readonly long connectionReference;
@ -34,7 +34,7 @@ namespace Tapeti.Connection
/// <inheritdoc /> /// <inheritdoc />
public override async Task HandleBasicDeliver(string consumerTag, public override void HandleBasicDeliver(string consumerTag,
ulong deliveryTag, ulong deliveryTag,
bool redelivered, bool redelivered,
string exchange, string exchange,
@ -49,16 +49,21 @@ namespace Tapeti.Connection
// //
// See also: https://github.com/JamesNK/Newtonsoft.Json/issues/1761 // See also: https://github.com/JamesNK/Newtonsoft.Json/issues/1761
var bodyArray = body.ToArray(); var bodyArray = body.ToArray();
try // Changing to AsyncDefaultBasicConsumer does not mean HandleBasicDeliver runs in parallel, the Task.Run would
// still be necessary, which is why TapetiBasicConsumer is a DefaultBasicConsumer.
Task.Run(async () =>
{ {
var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), bodyArray); try
await onRespond(connectionReference, deliveryTag, response); {
} var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), bodyArray);
catch await onRespond(connectionReference, deliveryTag, response);
{ }
await onRespond(connectionReference, deliveryTag, ConsumeResult.Error); catch
} {
await onRespond(connectionReference, deliveryTag, ConsumeResult.Error);
}
});
} }
} }
} }

3
Tapeti/Connection/TapetiClient.cs
View File

@ -777,8 +777,7 @@ namespace Tapeti.Connection
Password = connectionParams.Password, Password = connectionParams.Password,
AutomaticRecoveryEnabled = false, AutomaticRecoveryEnabled = false,
TopologyRecoveryEnabled = false, TopologyRecoveryEnabled = false,
RequestedHeartbeat = TimeSpan.FromSeconds(30), RequestedHeartbeat = TimeSpan.FromSeconds(30)
DispatchConsumersAsync = true
}; };
if (connectionParams.ClientProperties != null) if (connectionParams.ClientProperties != null)