Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8228328817 |
@@ -1,26 +0,0 @@
|
|||||||
# 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.
|
|
||||||
@@ -57,33 +57,6 @@ public sealed class RabbitMqConsumerBaseExtensionTests
|
|||||||
error.Should().BeNull();
|
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()
|
private static TestMessage CreateMessage() => new()
|
||||||
{
|
{
|
||||||
CorrelationContext = new CorrelationContext { CorrelationId = "payload-correlation" },
|
CorrelationContext = new CorrelationContext { CorrelationId = "payload-correlation" },
|
||||||
@@ -144,51 +117,6 @@ public sealed class RabbitMqConsumerBaseExtensionTests
|
|||||||
{
|
{
|
||||||
return TryValidateMessage(message, context, out 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>
|
private sealed record TestMessage : IRabbitMqMessage<string>
|
||||||
|
|||||||
@@ -25,10 +25,3 @@ ownership of acknowledgements and retries.
|
|||||||
Override `TryValidateMessage(...)` for application-specific permanent validation.
|
Override `TryValidateMessage(...)` for application-specific permanent validation.
|
||||||
Returning `false` nacks the delivery without requeue before retry processing begins.
|
Returning `false` nacks the delivery without requeue before retry processing begins.
|
||||||
Keep validation errors free of credentials and sensitive payload values.
|
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.
|
|
||||||
|
|||||||
@@ -82,20 +82,6 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
|||||||
return true;
|
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)
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||||
{
|
{
|
||||||
await EnsureConnectionAsync(stoppingToken);
|
await EnsureConnectionAsync(stoppingToken);
|
||||||
@@ -169,10 +155,6 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
|||||||
await _channel!.BasicAckAsync(args.DeliveryTag, multiple: false, cancellationToken: cancellationToken);
|
await _channel!.BasicAckAsync(args.DeliveryTag, multiple: false, cancellationToken: cancellationToken);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
|
||||||
{
|
|
||||||
throw;
|
|
||||||
}
|
|
||||||
catch (Exception ex) when (attempt < MaxRetries)
|
catch (Exception ex) when (attempt < MaxRetries)
|
||||||
{
|
{
|
||||||
_logger.LogWarning(ex,
|
_logger.LogWarning(ex,
|
||||||
@@ -187,23 +169,6 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
|
|||||||
"All {Max} attempts exhausted for message on queue {Queue} [CorrelationId={CorrelationId}] — nacking without requeue",
|
"All {Max} attempts exhausted for message on queue {Queue} [CorrelationId={CorrelationId}] — nacking without requeue",
|
||||||
MaxRetries, QueueName, payloadCorrelationId);
|
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);
|
await NackWithoutRequeueAsync(args.DeliveryTag, cancellationToken);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user