Files
hrynco-rabbitmq/Hrynco.RabbitMq.Tests/RabbitMqConsumerBaseExtensionTests.cs
agrynco 33f4b80ff8 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
2026-08-04 11:38:24 +03:00

207 lines
6.8 KiB
C#

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;
}
}