Compare commits
3 Commits
09c4b8b4d2
...
IT-1033
| Author | SHA1 | Date | |
|---|---|---|---|
| 81d35b5645 | |||
| 33f4b80ff8 | |||
| c0e83bdb77 |
@@ -0,0 +1,26 @@
|
||||
# HrynCo.RabbitMq Agent Rules
|
||||
|
||||
## Git workflow
|
||||
|
||||
- Treat `main` as the only default and integration branch for this repository.
|
||||
- Start every task branch from the latest `origin/main`.
|
||||
- Before creating a task branch, fetch the remote, switch to `main`, and fast-forward it from `origin/main`.
|
||||
- Create the task branch only after confirming that local `main` matches `origin/main`.
|
||||
- Open pull requests from the task branch into `main`.
|
||||
- Do not use `development` as the base or pull-request target for new work in this repository.
|
||||
- Do not commit, push, publish a NuGet package, or merge unless the repository owner explicitly requests that step.
|
||||
|
||||
Recommended branch preparation:
|
||||
|
||||
```text
|
||||
git fetch origin
|
||||
git switch main
|
||||
git pull --ff-only origin main
|
||||
git switch -c <task-branch>
|
||||
```
|
||||
|
||||
## Delivery
|
||||
|
||||
- Keep changes focused and backward compatible where practical because this repository produces a shared NuGet package.
|
||||
- Run the solution tests before handoff.
|
||||
- Use Conventional Commit messages with a lowercase subject and finish the commit body with `Ref: <issue-id>` when an issue exists.
|
||||
@@ -0,0 +1,206 @@
|
||||
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();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RetriesExhaustedHook_ReceivesMessageContextAndTerminalException()
|
||||
{
|
||||
var consumer = new TerminalFailureConsumer(CreateOptions());
|
||||
var message = CreateMessage();
|
||||
var exception = new InvalidOperationException("terminal failure");
|
||||
|
||||
await consumer.InvokeRetriesExhaustedAsync(message, Context, exception);
|
||||
|
||||
consumer.FailedMessage.Should().BeSameAs(message);
|
||||
consumer.FailedContext.Should().BeSameAs(Context);
|
||||
consumer.TerminalException.Should().BeSameAs(exception);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RetriesExhaustedHook_IsNoOpByDefault()
|
||||
{
|
||||
var consumer = new LegacyConsumer(CreateOptions());
|
||||
|
||||
Func<Task> act = () => consumer.InvokeRetriesExhaustedAsync(
|
||||
CreateMessage(),
|
||||
Context,
|
||||
new InvalidOperationException("terminal failure"));
|
||||
|
||||
await act.Should().NotThrowAsync();
|
||||
}
|
||||
|
||||
private static TestMessage CreateMessage() => new()
|
||||
{
|
||||
CorrelationContext = new CorrelationContext { CorrelationId = "payload-correlation" },
|
||||
Data = "payload"
|
||||
};
|
||||
|
||||
private static IOptionsMonitor<RabbitMqSettings> CreateOptions()
|
||||
{
|
||||
return new TestOptionsMonitor(new RabbitMqSettings
|
||||
{
|
||||
Host = "localhost",
|
||||
User = "guest",
|
||||
Password = "guest"
|
||||
});
|
||||
}
|
||||
|
||||
private sealed class ContextAwareConsumer(IOptionsMonitor<RabbitMqSettings> options)
|
||||
: RabbitMqConsumerBase<TestMessage, string>(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<RabbitMqSettings> options)
|
||||
: RabbitMqConsumerBase<TestMessage, string>(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);
|
||||
}
|
||||
|
||||
public Task InvokeRetriesExhaustedAsync(
|
||||
TestMessage message,
|
||||
RabbitMqMessageContext context,
|
||||
Exception exception)
|
||||
{
|
||||
return HandleMessageRetriesExhaustedAsync(
|
||||
message,
|
||||
context,
|
||||
exception,
|
||||
CancellationToken.None);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class TerminalFailureConsumer(IOptionsMonitor<RabbitMqSettings> options)
|
||||
: RabbitMqConsumerBase<TestMessage, string>(options, NullLogger.Instance)
|
||||
{
|
||||
protected override string QueueName => "messages.test";
|
||||
public TestMessage? FailedMessage { get; private set; }
|
||||
public RabbitMqMessageContext? FailedContext { get; private set; }
|
||||
public Exception? TerminalException { get; private set; }
|
||||
|
||||
protected override Task HandleMessageRetriesExhaustedAsync(
|
||||
TestMessage message,
|
||||
RabbitMqMessageContext context,
|
||||
Exception exception,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
FailedMessage = message;
|
||||
FailedContext = context;
|
||||
TerminalException = exception;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task InvokeRetriesExhaustedAsync(
|
||||
TestMessage message,
|
||||
RabbitMqMessageContext context,
|
||||
Exception exception)
|
||||
{
|
||||
return HandleMessageRetriesExhaustedAsync(
|
||||
message,
|
||||
context,
|
||||
exception,
|
||||
CancellationToken.None);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed record TestMessage : IRabbitMqMessage<string>
|
||||
{
|
||||
public CorrelationContext CorrelationContext { get; set; } = null!;
|
||||
public string Data { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
private sealed class TestOptionsMonitor(RabbitMqSettings settings) : IOptionsMonitor<RabbitMqSettings>
|
||||
{
|
||||
public RabbitMqSettings CurrentValue => settings;
|
||||
public RabbitMqSettings Get(string? name) => settings;
|
||||
public IDisposable? OnChange(Action<RabbitMqSettings, string?> listener) => null;
|
||||
}
|
||||
}
|
||||
@@ -6,10 +6,29 @@ 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<TMessage, TMessageData>` — abstract background service base for consumers, with retry + dead-letter support
|
||||
- `RabbitMqConsumerBase<TMessage, TMessageData>` — 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<TMessageData>` — 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.
|
||||
|
||||
Override `HandleMessageRetriesExhaustedAsync(...)` when a consumer must react once to a
|
||||
terminal processing failure, for example by publishing a neutral failure result to the
|
||||
requesting client. The hook runs after the final handler exception and before the original
|
||||
delivery is nacked without requeue. Hook failures are logged and do not replace the
|
||||
original failure or acknowledgement policy. Application shutdown cancellation does not
|
||||
invoke the hook.
|
||||
|
||||
@@ -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,55 @@ 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;
|
||||
}
|
||||
|
||||
/// <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)
|
||||
{
|
||||
@@ -88,19 +137,47 @@ 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;
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex) when (attempt < MaxRetries)
|
||||
{
|
||||
_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 +185,60 @@ 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);
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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;
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
namespace Hrynco.RabbitMq;
|
||||
|
||||
using System.Collections.Generic;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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<string, object?> Headers { get; init; }
|
||||
= new Dictionary<string, object?>();
|
||||
}
|
||||
Reference in New Issue
Block a user