feat: expose terminal consumer failure hook
Allow consumers to react once after all processing retries are exhausted while preserving the base acknowledgement policy. Ref: IT-1033
This commit is contained in:
@@ -82,6 +82,20 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Handles a terminal processing failure after all message retries are exhausted.
|
||||
/// The default implementation is a no-op. Implementations should avoid throwing;
|
||||
/// failures from this hook are logged and the original message is still nacked.
|
||||
/// </summary>
|
||||
protected virtual Task HandleMessageRetriesExhaustedAsync(
|
||||
TMessage message,
|
||||
RabbitMqMessageContext context,
|
||||
Exception exception,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await EnsureConnectionAsync(stoppingToken);
|
||||
@@ -155,6 +169,10 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
||||
await _channel!.BasicAckAsync(args.DeliveryTag, multiple: false, cancellationToken: cancellationToken);
|
||||
return;
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex) when (attempt < MaxRetries)
|
||||
{
|
||||
_logger.LogWarning(ex,
|
||||
@@ -169,6 +187,23 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
||||
"All {Max} attempts exhausted for message on queue {Queue} [CorrelationId={CorrelationId}] — nacking without requeue",
|
||||
MaxRetries, QueueName, payloadCorrelationId);
|
||||
|
||||
try
|
||||
{
|
||||
await HandleMessageRetriesExhaustedAsync(
|
||||
message,
|
||||
context,
|
||||
ex,
|
||||
cancellationToken);
|
||||
}
|
||||
catch (Exception hookException)
|
||||
{
|
||||
_logger.LogError(
|
||||
hookException,
|
||||
"Terminal failure hook failed for message on queue {Queue} [CorrelationId={CorrelationId}]",
|
||||
QueueName,
|
||||
payloadCorrelationId);
|
||||
}
|
||||
|
||||
await NackWithoutRequeueAsync(args.DeliveryTag, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user