-
Notifications
You must be signed in to change notification settings - Fork 647
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 #6094 from Particular/sc-retry-notifications
Send retry confirmation to SC
- Loading branch information
Showing
11 changed files
with
535 additions
and
38 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
110 changes: 110 additions & 0 deletions
110
...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,110 @@ | ||
| namespace NServiceBus.AcceptanceTests.Recoverability | ||
| { | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using AcceptanceTesting; | ||
| using AcceptanceTesting.Customization; | ||
| using EndpointTemplates; | ||
| using Features; | ||
| using Microsoft.Extensions.DependencyInjection; | ||
| 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 = DateTimeOffsetHelper.ToDateTimeOffset(context.RetryProcessingTimestamp); | ||
| Assert.That(processingTime, Is.EqualTo(DateTimeOffset.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.GetRequiredService<IMessageDispatcher>())); | ||
| } | ||
| } | ||
|
|
||
| class ControlMessageSender : FeatureStartupTask | ||
| { | ||
| IMessageDispatcher dispatcher; | ||
|
|
||
| public ControlMessageSender(IMessageDispatcher dispatcher) | ||
| { | ||
| this.dispatcher = dispatcher; | ||
| } | ||
|
|
||
| protected override async Task OnStart(IMessageSession session, CancellationToken cancellationToken = default) | ||
| { | ||
| 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(), cancellationToken); | ||
| } | ||
|
|
||
| protected override Task OnStop(IMessageSession session, CancellationToken cancellationToken = default) => Task.CompletedTask; | ||
| } | ||
| } | ||
|
|
||
| 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"]; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.