-
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.
Merge pull request #6101 from Particular/backport-sc-notifications
Backport ServiceControl Retry Notifications
- Loading branch information
Showing
10 changed files
with
516 additions
and
19 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
109 changes: 109 additions & 0 deletions
109
...rviceBus.AcceptanceTests/Recoverability/When_retrying_control_message_from_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,109 @@ | ||
| namespace NServiceBus.AcceptanceTests.Recoverability | ||
| { | ||
| using System; | ||
| using System.Threading.Tasks; | ||
| using AcceptanceTesting; | ||
| using AcceptanceTesting.Customization; | ||
| using EndpointTemplates; | ||
| using Extensibility; | ||
| using Features; | ||
| using NServiceBus.Pipeline; | ||
| using NServiceBus.Routing; | ||
| using NUnit.Framework; | ||
| using Transport; | ||
| using Unicast.Transport; | ||
|
|
||
| public class When_retrying_control_message_from_error_queue : NServiceBusAcceptanceTest | ||
| { | ||
| static readonly string RetryId = Guid.NewGuid().ToString("D"); | ||
|
|
||
| [Test] | ||
| public async Task Should_confirm_successful_processing() | ||
| { | ||
| Requires.MessageDrivenPubSub(); //required for subscription control message support | ||
|
|
||
| var context = await Scenario.Define<Context>() | ||
| .WithEndpoint<ProcessingEndpoint>() | ||
| .WithEndpoint<RetryAckSpy>() | ||
| .Done(c => c.ConfirmedRetryId != null) | ||
| .Run(); | ||
|
|
||
| Assert.AreEqual(RetryId, context.ConfirmedRetryId); | ||
| var processingTime = DateTimeExtensions.ToUtcDateTime(context.RetryProcessingTimestamp); | ||
| Assert.That(processingTime, Is.EqualTo(DateTime.UtcNow).Within(TimeSpan.FromMinutes(1))); | ||
| } | ||
|
|
||
| class Context : ScenarioContext | ||
| { | ||
| public string ConfirmedRetryId { get; set; } | ||
| public string RetryProcessingTimestamp { get; set; } | ||
| } | ||
|
|
||
| class ProcessingEndpoint : EndpointConfigurationBuilder | ||
| { | ||
| public ProcessingEndpoint() => EndpointSetup<DefaultServer>(c => | ||
| { | ||
| c.EnableFeature<ControlMessageFeature>(); | ||
| }); | ||
|
|
||
| class ControlMessageFeature : Feature | ||
| { | ||
| protected override void Setup(FeatureConfigurationContext context) | ||
| { | ||
| context.RegisterStartupTask(s => | ||
| new ControlMessageSender(s.Build<IDispatchMessages>())); | ||
| } | ||
| } | ||
|
|
||
| class ControlMessageSender : FeatureStartupTask | ||
| { | ||
| IDispatchMessages dispatcher; | ||
|
|
||
| public ControlMessageSender(IDispatchMessages dispatcher) | ||
| { | ||
| this.dispatcher = dispatcher; | ||
| } | ||
|
|
||
| protected override async Task OnStart(IMessageSession session) | ||
| { | ||
| var controlMessage = ControlMessageFactory.Create(MessageIntentEnum.Subscribe); | ||
| // set necessary subscription control message headers | ||
| controlMessage.Headers.Add(Headers.SubscriptionMessageType, typeof(object).AssemblyQualifiedName); | ||
| controlMessage.Headers.Add(Headers.ReplyToAddress, "TestSubscriberAddress"); | ||
| // set SC headers | ||
| controlMessage.Headers.Add("ServiceControl.Retry.UniqueMessageId", RetryId); | ||
| controlMessage.Headers.Add("ServiceControl.Retry.AcknowledgementQueue", Conventions.EndpointNamingConvention(typeof(RetryAckSpy))); | ||
| var messageOperation = new TransportOperation(controlMessage, new UnicastAddressTag(Conventions.EndpointNamingConvention(typeof(ProcessingEndpoint)))); | ||
| await dispatcher.Dispatch(new TransportOperations(messageOperation), new TransportTransaction(), new ContextBag()); | ||
| } | ||
|
|
||
| protected override Task OnStop(IMessageSession session) => Task.FromResult(0); | ||
| } | ||
| } | ||
|
|
||
| class RetryAckSpy : EndpointConfigurationBuilder | ||
| { | ||
| public RetryAckSpy() => EndpointSetup<DefaultServer>((e, r) => e.Pipeline.Register( | ||
| new ControlMessageBehavior(r.ScenarioContext as Context), | ||
| "Checks for confirmation control message")); | ||
|
|
||
| class ControlMessageBehavior : Behavior<IIncomingPhysicalMessageContext> | ||
| { | ||
| Context testContext; | ||
|
|
||
| public ControlMessageBehavior(Context testContext) | ||
| { | ||
| this.testContext = testContext; | ||
| } | ||
|
|
||
| public override async Task Invoke(IIncomingPhysicalMessageContext context, Func<Task> next) | ||
| { | ||
| await next(); | ||
|
|
||
| testContext.ConfirmedRetryId = context.MessageHeaders["ServiceControl.Retry.UniqueMessageId"]; | ||
| testContext.RetryProcessingTimestamp = context.MessageHeaders["ServiceControl.Retry.Successful"]; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
145 changes: 145 additions & 0 deletions
145
src/NServiceBus.AcceptanceTests/Recoverability/When_retrying_message_from_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,145 @@ | ||
| namespace NServiceBus.AcceptanceTests.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_retrying_message_from_error_queue : NServiceBusAcceptanceTest | ||
| { | ||
| [Test] | ||
| public async Task Should_confirm_successful_processing() | ||
| { | ||
| var retryId = Guid.NewGuid().ToString("D"); | ||
|
|
||
| var context = await Scenario.Define<Context>() | ||
| .WithEndpoint<ProcessingEndpoint>(e => e | ||
| .When(s => | ||
| { | ||
| var sendOptions = new SendOptions(); | ||
| sendOptions.RouteToThisEndpoint(); | ||
| // set SC retry header information | ||
| sendOptions.SetHeader("ServiceControl.Retry.UniqueMessageId", retryId); | ||
| sendOptions.SetHeader("ServiceControl.Retry.AcknowledgementQueue", Conventions.EndpointNamingConvention(typeof(RetryAckSpy))); | ||
| return s.Send(new FailedMessage(), sendOptions); | ||
| })) | ||
| .WithEndpoint<RetryAckSpy>() | ||
| .WithEndpoint<AuditSpy>() | ||
| .Done(c => c.ConfirmedRetryId != null && c.AuditHeaders != null) | ||
| .Run(); | ||
|
|
||
| Assert.IsTrue(context.MessageProcessed); | ||
| Assert.AreEqual(retryId, context.ConfirmedRetryId); | ||
| var processingTime = DateTimeExtensions.ToUtcDateTime(context.RetryProcessingTimestamp); | ||
| Assert.That(processingTime, Is.EqualTo(DateTime.UtcNow).Within(TimeSpan.FromMinutes(1))); | ||
| Assert.IsTrue(context.AuditHeaders.ContainsKey("ServiceControl.Retry.AcknowledgementSent")); | ||
| } | ||
|
|
||
| class Context : ScenarioContext | ||
| { | ||
| public string ConfirmedRetryId { get; set; } | ||
| public string RetryProcessingTimestamp { get; set; } | ||
| public bool MessageProcessed { get; set; } | ||
| public IReadOnlyDictionary<string, string> AuditHeaders { get; set; } | ||
| } | ||
|
|
||
| class ProcessingEndpoint : EndpointConfigurationBuilder | ||
| { | ||
| public ProcessingEndpoint() | ||
| { | ||
| EndpointSetup<DefaultServer>(c => | ||
| { | ||
| c.AuditProcessedMessagesTo<AuditSpy>(); | ||
| }); | ||
| } | ||
|
|
||
| class FailedMessageHandler : IHandleMessages<FailedMessage> | ||
| { | ||
| Context testContext; | ||
|
|
||
| public FailedMessageHandler(Context testContext) | ||
| { | ||
| this.testContext = testContext; | ||
| } | ||
|
|
||
| public Task Handle(FailedMessage message, IMessageHandlerContext context) | ||
| { | ||
| testContext.MessageProcessed = true; | ||
| return Task.FromResult(0); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class RetryAckSpy : EndpointConfigurationBuilder | ||
| { | ||
| public RetryAckSpy() => EndpointSetup<DefaultServer>((e, r) => e.Pipeline.Register( | ||
| new ControlMessageBehavior(r.ScenarioContext as Context), | ||
| "Checks for confirmation control message")); | ||
|
|
||
| class ControlMessageBehavior : Behavior<IIncomingPhysicalMessageContext> | ||
| { | ||
| Context testContext; | ||
|
|
||
| public ControlMessageBehavior(Context testContext) | ||
| { | ||
| this.testContext = testContext; | ||
| } | ||
|
|
||
| public override async Task Invoke(IIncomingPhysicalMessageContext context, Func<Task> next) | ||
| { | ||
| await next(); | ||
|
|
||
| testContext.ConfirmedRetryId = context.MessageHeaders["ServiceControl.Retry.UniqueMessageId"]; | ||
| testContext.RetryProcessingTimestamp = context.MessageHeaders["ServiceControl.Retry.Successful"]; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class AuditSpy : EndpointConfigurationBuilder | ||
| { | ||
| public AuditSpy() => EndpointSetup<DefaultServer>(); | ||
|
|
||
| class FailedMessageHandler : IHandleMessages<FailedMessage> | ||
| { | ||
| Context testContext; | ||
|
|
||
| public FailedMessageHandler(Context testContext) | ||
| { | ||
| this.testContext = testContext; | ||
| } | ||
|
|
||
| public Task Handle(FailedMessage message, IMessageHandlerContext context) | ||
| { | ||
| testContext.AuditHeaders = context.MessageHeaders; | ||
| return Task.FromResult(0); | ||
| } | ||
| } | ||
|
|
||
| class ControlMessageBehavior : Behavior<IIncomingPhysicalMessageContext> | ||
| { | ||
| Context testContext; | ||
|
|
||
| public ControlMessageBehavior(Context testContext) | ||
| { | ||
| this.testContext = testContext; | ||
| } | ||
|
|
||
| public override async Task Invoke(IIncomingPhysicalMessageContext context, Func<Task> next) | ||
| { | ||
| await next(); | ||
|
|
||
| testContext.ConfirmedRetryId = context.MessageHeaders["ServiceControl.Retry.UniqueMessageId"]; | ||
| testContext.RetryProcessingTimestamp = context.MessageHeaders["ServiceControl.Retry.Successful"]; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public class FailedMessage : 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.