-
Notifications
You must be signed in to change notification settings - Fork 648
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Introduce a recoverability pipeline (#6238)
* Introduce a dispatch pipline to outgoing errors * Temp * Naming * Revert * Revert * Adjust context * Split executors * FIrst draft of a pipeline * Fix filename * Fix names * Invoke policy before pipeline * Move errorhandle result to the recoverability action * Pull recoverability events out to a behavior * Make transport adjustments in one place * Make satellites use the same executor * Make actions return the transport operations * Remove unused file * Fix logging * Add the satellite executor back * Make static headers work again * Move events to terminator * Cleanup * Add test to show alternate recoverability actions * Make actions extendable * Make test more realistic * Cleanup test * Better name * Approvals * Allow the actions to return the notification with an internal API * Make things protected internal to enforce normal creation via the provided methods but allow inheritance * Fix the notifications * Switch to routing context design * Fix tests and approvals * Remove the dispatch method from error context and partially undo the extensibility changes * Remove the closure allocations from the recoverability action invocations * Additional comment * Update src/NServiceBus.Core/Recoverability/IRecoverabilityActionContext.cs Co-authored-by: Andreas Öhlund <andreas.ohlund@particular.net> * Proper typing for the executors Co-authored-by: danielmarbach <daniel.marbach@openplace.net> Co-authored-by: Daniel Marbach <daniel.marbach@nservicebus.com>
- Loading branch information
3 people
authored
Feb 8, 2022
1 parent
846a2b4
commit 7044043
Showing
50 changed files
with
1,438 additions
and
996 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
133 changes: 133 additions & 0 deletions
133
src/NServiceBus.AcceptanceTests/Core/Recoverability/When_applying_message_recoverability.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| namespace NServiceBus.AcceptanceTests.Core.Recoverability | ||
| { | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Threading.Tasks; | ||
| using AcceptanceTesting; | ||
| using AcceptanceTesting.Customization; | ||
| using EndpointTemplates; | ||
| using NServiceBus.Pipeline; | ||
| using NUnit.Framework; | ||
|
|
||
| public class When_applying_message_recoverability : NServiceBusAcceptanceTest | ||
| { | ||
| [Test] | ||
| public async Task Should_allow_for_alternate_move_to_error_action() | ||
| { | ||
| var onMessageSentToErrorQueueTriggered = false; | ||
| var context = await Scenario.Define<Context>() | ||
| .WithEndpoint<EndpointWithFailingHandler>(b => b | ||
| .DoNotFailOnErrorMessages() | ||
| .CustomConfig(config => | ||
| { | ||
| config.Recoverability() | ||
| .Failed(f => f.OnMessageSentToErrorQueue((_, __) => | ||
| { | ||
| onMessageSentToErrorQueueTriggered = true; | ||
| return Task.CompletedTask; | ||
| })); | ||
| }) | ||
| .When((session, ctx) => session.SendLocal(new InitiatingMessage())) | ||
| ) | ||
| .WithEndpoint<ErrorSpy>() | ||
| .Done(c => c.MessageMovedToErrorQueue) | ||
| .Run(); | ||
|
|
||
| Assert.True(context.MessageBodyWasEmpty); | ||
| Assert.True(onMessageSentToErrorQueueTriggered); | ||
| } | ||
|
|
||
| class Context : ScenarioContext | ||
| { | ||
| public bool MessageMovedToErrorQueue { get; set; } | ||
| public bool MessageBodyWasEmpty { get; set; } | ||
| } | ||
|
|
||
| class EndpointWithFailingHandler : EndpointConfigurationBuilder | ||
| { | ||
| static readonly string ErrorQueueAddress = Conventions.EndpointNamingConvention(typeof(ErrorSpy)); | ||
|
|
||
| public EndpointWithFailingHandler() | ||
| { | ||
| EndpointSetup<DefaultServer>((config, context) => | ||
| { | ||
| config.SendFailedMessagesTo(ErrorQueueAddress); | ||
| config.Pipeline.Register(typeof(CustomRecoverabilityActionBehavior), "Applies a custom recoverability actions"); | ||
| }); | ||
| } | ||
|
|
||
| public class CustomRecoverabilityActionBehavior : Behavior<IRecoverabilityContext> | ||
| { | ||
| public override Task Invoke(IRecoverabilityContext context, Func<Task> next) | ||
| { | ||
| if (context.RecoverabilityAction is MoveToError) | ||
| { | ||
| //Here we could store the body, headers and error metadata elsewhere | ||
|
|
||
| context.RecoverabilityAction = new CustomOnErrorAction(context.RecoverabilityConfiguration.Failed.ErrorQueue); | ||
| } | ||
|
|
||
| return next(); | ||
| } | ||
|
|
||
| class CustomOnErrorAction : MoveToError | ||
| { | ||
| public CustomOnErrorAction(string errorQueue) : base(errorQueue) | ||
| { | ||
| } | ||
|
|
||
| public override IReadOnlyCollection<IRoutingContext> GetRoutingContexts(IRecoverabilityActionContext context) | ||
| { | ||
| var routingContexts = base.GetRoutingContexts(context); | ||
|
|
||
| // show how we just send an empty message with the message id to the error queue | ||
| // headers are preserved to make sure the necessary acceptance test infrastructure is still present | ||
| foreach (var routingContext in routingContexts) | ||
| { | ||
| routingContext.Message.UpdateBody(ReadOnlyMemory<byte>.Empty); | ||
| } | ||
|
|
||
| return routingContexts; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class InitiatingHandler : IHandleMessages<InitiatingMessage> | ||
| { | ||
| public Task Handle(InitiatingMessage initiatingMessage, IMessageHandlerContext context) | ||
| { | ||
| throw new SimulatedException("Some failure"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class ErrorSpy : EndpointConfigurationBuilder | ||
| { | ||
| public ErrorSpy() | ||
| { | ||
| EndpointSetup<DefaultServer>(c => c.Pipeline.Register(typeof(ErrorMessageDetector), "Detect incoming error messages")); | ||
| } | ||
|
|
||
| class ErrorMessageDetector : IBehavior<ITransportReceiveContext, ITransportReceiveContext> | ||
| { | ||
| public ErrorMessageDetector(Context testContext) | ||
| { | ||
| this.testContext = testContext; | ||
| } | ||
|
|
||
| public Task Invoke(ITransportReceiveContext context, Func<ITransportReceiveContext, Task> next) | ||
| { | ||
| testContext.MessageBodyWasEmpty = context.Message.Body.IsEmpty; | ||
| testContext.MessageMovedToErrorQueue = true; | ||
| return next(context); | ||
| } | ||
|
|
||
| Context testContext; | ||
| } | ||
| } | ||
|
|
||
| public class InitiatingMessage : IMessage | ||
| { | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
105 changes: 105 additions & 0 deletions
105
...rviceBus.AcceptanceTests/Core/Recoverability/When_message_is_dispatched_to_error_queue.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| namespace NServiceBus.AcceptanceTests.Core.Recoverability | ||
| { | ||
| using System; | ||
| using System.Threading.Tasks; | ||
| using AcceptanceTesting; | ||
| using AcceptanceTesting.Customization; | ||
| using EndpointTemplates; | ||
| using NServiceBus.Pipeline; | ||
| using NServiceBus.Routing; | ||
| using NUnit.Framework; | ||
|
|
||
| public class When_message_is_dispatched_to_error_queue : NServiceBusAcceptanceTest | ||
| { | ||
| [Test] | ||
| public async Task Should_allow_body_to_be_manipulated() | ||
| { | ||
| var context = await Scenario.Define<Context>() | ||
| .WithEndpoint<EndpointWithFailingHandler>(b => b | ||
| .DoNotFailOnErrorMessages() | ||
| .When((session, ctx) => session.SendLocal(new InitiatingMessage())) | ||
| ) | ||
| .WithEndpoint<ErrorSpy>() | ||
| .Done(c => c.MessageMovedToErrorQueue) | ||
| .Run(); | ||
|
|
||
| Assert.True(context.MessageBodyWasEmpty); | ||
| } | ||
|
|
||
| class Context : ScenarioContext | ||
| { | ||
| public bool MessageMovedToErrorQueue { get; set; } | ||
| public bool MessageBodyWasEmpty { get; internal set; } | ||
| } | ||
|
|
||
| class EndpointWithFailingHandler : EndpointConfigurationBuilder | ||
| { | ||
| static string errorQueueAddress = Conventions.EndpointNamingConvention(typeof(ErrorSpy)); | ||
|
|
||
| public EndpointWithFailingHandler() | ||
| { | ||
| EndpointSetup<DefaultServer>((config, context) => | ||
| { | ||
| config.SendFailedMessagesTo(errorQueueAddress); | ||
| config.Pipeline.Register(typeof(ErrorBodyStorageBehavior), "Simulate writing the body to a separate storage and pass a null body to the transport"); | ||
| }); | ||
| } | ||
|
|
||
| public class ErrorBodyStorageBehavior : Behavior<IDispatchContext> | ||
| { | ||
| public override Task Invoke(IDispatchContext context, Func<Task> next) | ||
| { | ||
| foreach (var operation in context.Operations) | ||
| { | ||
| var unicastAddress = operation.AddressTag as UnicastAddressTag; | ||
|
|
||
| if (unicastAddress?.Destination != errorQueueAddress) | ||
| { | ||
| continue; | ||
| } | ||
|
|
||
| operation.Message.UpdateBody(ReadOnlyMemory<byte>.Empty); | ||
| } | ||
| return next(); | ||
| } | ||
| } | ||
|
|
||
| class InitiatingHandler : IHandleMessages<InitiatingMessage> | ||
| { | ||
| public Task Handle(InitiatingMessage initiatingMessage, IMessageHandlerContext context) | ||
| { | ||
| throw new SimulatedException("Some failure"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class ErrorSpy : EndpointConfigurationBuilder | ||
| { | ||
| public ErrorSpy() | ||
| { | ||
| EndpointSetup<DefaultServer>(c => c.Pipeline.Register(typeof(ErrorMessageDetector), "Detect incoming error messages")); | ||
| } | ||
|
|
||
| class ErrorMessageDetector : IBehavior<ITransportReceiveContext, ITransportReceiveContext> | ||
| { | ||
| public ErrorMessageDetector(Context testContext) | ||
| { | ||
| this.testContext = testContext; | ||
| } | ||
|
|
||
| public Task Invoke(ITransportReceiveContext context, Func<ITransportReceiveContext, Task> next) | ||
| { | ||
| testContext.MessageBodyWasEmpty = context.Message.Body.IsEmpty; | ||
| testContext.MessageMovedToErrorQueue = true; | ||
| return next(context); | ||
| } | ||
|
|
||
| Context testContext; | ||
| } | ||
| } | ||
|
|
||
| public class InitiatingMessage : IMessage | ||
| { | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.