From c0e83bdb779a1a7d9f55dc883cbb26c8e900bfb5 Mon Sep 17 00:00:00 2001 From: Anatolii Hrynchuk Date: Sun, 2 Aug 2026 22:56:57 +0300 Subject: [PATCH] feat: expose RabbitMQ delivery context Add backward-compatible context-aware handling and permanent validation hooks. Ref: IT-1033 --- .../RabbitMqConsumerBaseExtensionTests.cs | 134 ++++++++++++++++++ Hrynco.RabbitMq/README.md | 14 +- Hrynco.RabbitMq/RabbitMqConsumerBase.cs | 99 ++++++++++++- Hrynco.RabbitMq/RabbitMqMessageContext.cs | 23 +++ 4 files changed, 264 insertions(+), 6 deletions(-) create mode 100644 Hrynco.RabbitMq.Tests/RabbitMqConsumerBaseExtensionTests.cs create mode 100644 Hrynco.RabbitMq/RabbitMqMessageContext.cs diff --git a/Hrynco.RabbitMq.Tests/RabbitMqConsumerBaseExtensionTests.cs b/Hrynco.RabbitMq.Tests/RabbitMqConsumerBaseExtensionTests.cs new file mode 100644 index 0000000..09c8623 --- /dev/null +++ b/Hrynco.RabbitMq.Tests/RabbitMqConsumerBaseExtensionTests.cs @@ -0,0 +1,134 @@ +namespace Hrynco.RabbitMq.Tests; + +using System; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Xunit; + +public sealed class RabbitMqConsumerBaseExtensionTests +{ + private static readonly RabbitMqMessageContext Context = new() + { + QueueName = "messages.test", + Exchange = string.Empty, + RoutingKey = "messages.test", + MessageId = "message-123", + MessageType = "Example.Message.v1", + CorrelationId = "correlation-123", + ReplyTo = "messages.reply", + ContentType = "application/json", + Redelivered = true + }; + + [Fact] + public async Task ContextAwareOverload_ReceivesTransportMetadata() + { + var consumer = new ContextAwareConsumer(CreateOptions()); + var message = CreateMessage(); + + await consumer.InvokeAsync(message, Context); + + consumer.HandledMessage.Should().BeSameAs(message); + consumer.HandledContext.Should().BeSameAs(Context); + } + + [Fact] + public async Task ContextAwareOverload_DelegatesToLegacyOverride() + { + var consumer = new LegacyConsumer(CreateOptions()); + var message = CreateMessage(); + + await consumer.InvokeAsync(message, Context); + + consumer.HandledMessage.Should().BeSameAs(message); + } + + [Fact] + public void ValidationHook_IsValidByDefault() + { + var consumer = new LegacyConsumer(CreateOptions()); + + bool isValid = consumer.Validate(CreateMessage(), Context, out string? error); + + isValid.Should().BeTrue(); + error.Should().BeNull(); + } + + private static TestMessage CreateMessage() => new() + { + CorrelationContext = new CorrelationContext { CorrelationId = "payload-correlation" }, + Data = "payload" + }; + + private static IOptionsMonitor CreateOptions() + { + return new TestOptionsMonitor(new RabbitMqSettings + { + Host = "localhost", + User = "guest", + Password = "guest" + }); + } + + private sealed class ContextAwareConsumer(IOptionsMonitor options) + : RabbitMqConsumerBase(options, NullLogger.Instance) + { + protected override string QueueName => "messages.test"; + public TestMessage? HandledMessage { get; private set; } + public RabbitMqMessageContext? HandledContext { get; private set; } + + protected override Task HandleMessageAsync( + TestMessage message, + RabbitMqMessageContext context, + CancellationToken cancellationToken) + { + HandledMessage = message; + HandledContext = context; + return Task.CompletedTask; + } + + public Task InvokeAsync(TestMessage message, RabbitMqMessageContext context) + { + return HandleMessageAsync(message, context, CancellationToken.None); + } + } + + private sealed class LegacyConsumer(IOptionsMonitor options) + : RabbitMqConsumerBase(options, NullLogger.Instance) + { + protected override string QueueName => "messages.test"; + public TestMessage? HandledMessage { get; private set; } + + protected override Task HandleMessageAsync(TestMessage message, CancellationToken cancellationToken) + { + HandledMessage = message; + return Task.CompletedTask; + } + + public Task InvokeAsync(TestMessage message, RabbitMqMessageContext context) + { + return HandleMessageAsync(message, context, CancellationToken.None); + } + + public bool Validate(TestMessage message, RabbitMqMessageContext context, out string? error) + { + return TryValidateMessage(message, context, out error); + } + } + + private sealed record TestMessage : IRabbitMqMessage + { + public CorrelationContext CorrelationContext { get; set; } = null!; + public string Data { get; set; } = string.Empty; + } + + private sealed class TestOptionsMonitor(RabbitMqSettings settings) : IOptionsMonitor + { + public RabbitMqSettings CurrentValue => settings; + public RabbitMqSettings Get(string? name) => settings; + public IDisposable? OnChange(Action listener) => null; + } +} diff --git a/Hrynco.RabbitMq/README.md b/Hrynco.RabbitMq/README.md index 25da421..bd1d96d 100644 --- a/Hrynco.RabbitMq/README.md +++ b/Hrynco.RabbitMq/README.md @@ -6,10 +6,22 @@ RabbitMQ publisher and consumer base for HrynCo applications. - `RabbitMqSettings` — connection settings record (host, port, user, password, virtual host) - `IRabbitMqPublisher` / `RabbitMqPublisher` — publishes JSON-serialized messages to a named queue -- `RabbitMqConsumerBase` — abstract background service base for consumers, with retry + dead-letter support +- `RabbitMqConsumerBase` — background service base with connection management, manual ACK/NACK, retry, permanent-validation rejection, and backward-compatible context-aware handling +- `RabbitMqMessageContext` — neutral AMQP delivery metadata (`MessageId`, type, correlation, routing, headers, and redelivery state) - `IRabbitMqMessage` — message contract interface - `CorrelationContext` — correlation ID carrier ## Packaging This package is intended for reuse through NuGet. The test project is excluded from packing. + +## Consumer extension points + +Existing consumers can keep overriding `HandleMessageAsync(message, cancellationToken)`. +Consumers that need AMQP metadata can instead override +`HandleMessageAsync(message, context, cancellationToken)`. The base class retains +ownership of acknowledgements and retries. + +Override `TryValidateMessage(...)` for application-specific permanent validation. +Returning `false` nacks the delivery without requeue before retry processing begins. +Keep validation errors free of credentials and sensitive payload values. diff --git a/Hrynco.RabbitMq/RabbitMqConsumerBase.cs b/Hrynco.RabbitMq/RabbitMqConsumerBase.cs index 6b6e90e..96bd3f2 100644 --- a/Hrynco.RabbitMq/RabbitMqConsumerBase.cs +++ b/Hrynco.RabbitMq/RabbitMqConsumerBase.cs @@ -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 : BackgroundS private RabbitMqSettings Settings => _options.Get(SettingsName); - protected abstract Task HandleMessageAsync(TMessage message, CancellationToken cancellationToken); + /// + /// Handles a deserialized message. Existing consumers can continue overriding this overload. + /// New consumers that need transport metadata can override the context-aware overload instead. + /// + protected virtual Task HandleMessageAsync(TMessage message, CancellationToken cancellationToken) + { + throw new NotSupportedException( + $"{GetType().Name} must override a HandleMessageAsync overload."); + } + + /// + /// Handles a deserialized message together with neutral RabbitMQ delivery metadata. + /// The default implementation preserves compatibility by delegating to the original overload. + /// + protected virtual Task HandleMessageAsync( + TMessage message, + RabbitMqMessageContext context, + CancellationToken cancellationToken) + { + return HandleMessageAsync(message, cancellationToken); + } + + /// + /// 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. + /// + 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 : BackgroundS return; } + RabbitMqMessageContext context = CreateMessageContext(args); + string? payloadCorrelationId = message.CorrelationContext?.CorrelationId; + + using IDisposable? scope = _logger.BeginScope(new Dictionary + { + ["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 : 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 : 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 headers = args.BasicProperties.Headers is null + ? new Dictionary() + : new Dictionary(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; diff --git a/Hrynco.RabbitMq/RabbitMqMessageContext.cs b/Hrynco.RabbitMq/RabbitMqMessageContext.cs new file mode 100644 index 0000000..65a9faf --- /dev/null +++ b/Hrynco.RabbitMq/RabbitMqMessageContext.cs @@ -0,0 +1,23 @@ +namespace Hrynco.RabbitMq; + +using System.Collections.Generic; + +/// +/// Transport metadata supplied by RabbitMQ for a delivered message. +/// Application-specific consumers can use this context for validation, +/// correlation, and diagnostics without taking ownership of acknowledgements. +/// +public sealed record RabbitMqMessageContext +{ + public required string QueueName { get; init; } + public required string Exchange { get; init; } + public required string RoutingKey { get; init; } + public string? MessageId { get; init; } + public string? MessageType { get; init; } + public string? CorrelationId { get; init; } + public string? ReplyTo { get; init; } + public string? ContentType { get; init; } + public bool Redelivered { get; init; } + public IReadOnlyDictionary Headers { get; init; } + = new Dictionary(); +} -- 2.52.0