feat: expose RabbitMQ delivery context
Add backward-compatible context-aware handling and permanent validation hooks. Ref: IT-1033
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
namespace Hrynco.RabbitMq;
|
||||
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using System.Threading;
|
||||
@@ -45,7 +46,41 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
||||
|
||||
private RabbitMqSettings Settings => _options.Get(SettingsName);
|
||||
|
||||
protected abstract Task HandleMessageAsync(TMessage message, CancellationToken cancellationToken);
|
||||
/// <summary>
|
||||
/// Handles a deserialized message. Existing consumers can continue overriding this overload.
|
||||
/// New consumers that need transport metadata can override the context-aware overload instead.
|
||||
/// </summary>
|
||||
protected virtual Task HandleMessageAsync(TMessage message, CancellationToken cancellationToken)
|
||||
{
|
||||
throw new NotSupportedException(
|
||||
$"{GetType().Name} must override a HandleMessageAsync overload.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Handles a deserialized message together with neutral RabbitMQ delivery metadata.
|
||||
/// The default implementation preserves compatibility by delegating to the original overload.
|
||||
/// </summary>
|
||||
protected virtual Task HandleMessageAsync(
|
||||
TMessage message,
|
||||
RabbitMqMessageContext context,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return HandleMessageAsync(message, cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Performs application-specific validation before retries begin.
|
||||
/// Return false for a permanently invalid message; it will be nacked without requeue.
|
||||
/// Validation errors should describe the rule without including sensitive payload values.
|
||||
/// </summary>
|
||||
protected virtual bool TryValidateMessage(
|
||||
TMessage message,
|
||||
RabbitMqMessageContext context,
|
||||
out string? validationError)
|
||||
{
|
||||
validationError = null;
|
||||
return true;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
@@ -88,11 +123,35 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
||||
return;
|
||||
}
|
||||
|
||||
RabbitMqMessageContext context = CreateMessageContext(args);
|
||||
string? payloadCorrelationId = message.CorrelationContext?.CorrelationId;
|
||||
|
||||
using IDisposable? scope = _logger.BeginScope(new Dictionary<string, object?>
|
||||
{
|
||||
["MessageId"] = context.MessageId,
|
||||
["MessageType"] = context.MessageType,
|
||||
["CorrelationId"] = payloadCorrelationId ?? context.CorrelationId,
|
||||
["BrokerCorrelationId"] = context.CorrelationId,
|
||||
["Queue"] = context.QueueName,
|
||||
["RoutingKey"] = context.RoutingKey,
|
||||
["Redelivered"] = context.Redelivered
|
||||
});
|
||||
|
||||
if (!TryValidateMessage(message, context, out string? validationError))
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"Rejected invalid message on queue {Queue}: {ValidationError} — nacking without requeue",
|
||||
QueueName,
|
||||
validationError ?? "No validation error was provided");
|
||||
await NackWithoutRequeueAsync(args.DeliveryTag, cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
for (int attempt = 1; attempt <= MaxRetries; attempt++)
|
||||
{
|
||||
try
|
||||
{
|
||||
await HandleMessageAsync(message, cancellationToken);
|
||||
await HandleMessageAsync(message, context, cancellationToken);
|
||||
await _channel!.BasicAckAsync(args.DeliveryTag, multiple: false, cancellationToken: cancellationToken);
|
||||
return;
|
||||
}
|
||||
@@ -100,7 +159,7 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
||||
{
|
||||
_logger.LogWarning(ex,
|
||||
"Attempt {Attempt}/{Max} failed for message on queue {Queue} [CorrelationId={CorrelationId}] — retrying in {Delay}s",
|
||||
attempt, MaxRetries, QueueName, message.CorrelationContext?.CorrelationId, RetryDelay.TotalSeconds);
|
||||
attempt, MaxRetries, QueueName, payloadCorrelationId, RetryDelay.TotalSeconds);
|
||||
|
||||
await Task.Delay(RetryDelay, cancellationToken);
|
||||
}
|
||||
@@ -108,13 +167,43 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
||||
{
|
||||
_logger.LogError(ex,
|
||||
"All {Max} attempts exhausted for message on queue {Queue} [CorrelationId={CorrelationId}] — nacking without requeue",
|
||||
MaxRetries, QueueName, message.CorrelationContext?.CorrelationId);
|
||||
MaxRetries, QueueName, payloadCorrelationId);
|
||||
|
||||
await _channel!.BasicNackAsync(args.DeliveryTag, multiple: false, requeue: false, cancellationToken: cancellationToken);
|
||||
await NackWithoutRequeueAsync(args.DeliveryTag, cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private RabbitMqMessageContext CreateMessageContext(BasicDeliverEventArgs args)
|
||||
{
|
||||
IReadOnlyDictionary<string, object?> headers = args.BasicProperties.Headers is null
|
||||
? new Dictionary<string, object?>()
|
||||
: new Dictionary<string, object?>(args.BasicProperties.Headers);
|
||||
|
||||
return new RabbitMqMessageContext
|
||||
{
|
||||
QueueName = QueueName,
|
||||
Exchange = args.Exchange,
|
||||
RoutingKey = args.RoutingKey,
|
||||
MessageId = args.BasicProperties.MessageId,
|
||||
MessageType = args.BasicProperties.Type,
|
||||
CorrelationId = args.BasicProperties.CorrelationId,
|
||||
ReplyTo = args.BasicProperties.ReplyTo,
|
||||
ContentType = args.BasicProperties.ContentType,
|
||||
Redelivered = args.Redelivered,
|
||||
Headers = headers
|
||||
};
|
||||
}
|
||||
|
||||
private Task NackWithoutRequeueAsync(ulong deliveryTag, CancellationToken cancellationToken)
|
||||
{
|
||||
return _channel!.BasicNackAsync(
|
||||
deliveryTag,
|
||||
multiple: false,
|
||||
requeue: false,
|
||||
cancellationToken: cancellationToken).AsTask();
|
||||
}
|
||||
|
||||
private async Task EnsureConnectionAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var s = Settings;
|
||||
|
||||
Reference in New Issue
Block a user