-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Track messages that successfully completed the message or error pipel…
…ine but failed to get acknowledged due to expired leases in receiveonly mode (#1046) * Track messages that successfully completed the message or error pipeline but failed to get acknowledged due to expired leases in receiveonly mode (#1034) # Conflicts: # src/AcceptanceTests/NServiceBus.Transport.AzureServiceBus.AcceptanceTests.csproj # src/CommandLine/NServiceBus.Transport.AzureServiceBus.CommandLine.csproj # src/CommandLineTests/NServiceBus.Transport.AzureServiceBus.CommandLine.Tests.csproj # src/Tests/FakeReceiver.cs # src/Tests/NServiceBus.Transport.AzureServiceBus.Tests.csproj # src/Transport/NServiceBus.Transport.AzureServiceBus.csproj # src/TransportTests/NServiceBus.Transport.AzureServiceBus.TransportTests.csproj * Update src/AcceptanceTests/Receiving/When_message_visibility_expired.cs Co-authored-by: Travis Nickels <[email protected]> --------- Co-authored-by: Travis Nickels <[email protected]>
- Loading branch information
1 parent
624a16b
commit 8304add
Showing
13 changed files
with
487 additions
and
121 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
115 changes: 115 additions & 0 deletions
115
src/AcceptanceTests/Receiving/When_message_visibility_expired.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,115 @@ | ||
| namespace NServiceBus.Transport.AzureServiceBus.AcceptanceTests | ||
| { | ||
| using System; | ||
| using System.Linq; | ||
| using System.Threading.Tasks; | ||
| using AcceptanceTesting; | ||
| using Azure.Messaging.ServiceBus; | ||
| using NServiceBus.AcceptanceTests; | ||
| using NServiceBus.AcceptanceTests.EndpointTemplates; | ||
| using NUnit.Framework; | ||
|
|
||
| public class When_message_visibility_expired : NServiceBusAcceptanceTest | ||
| { | ||
| [Test] | ||
| public async Task Should_complete_message_on_next_receive_when_pipeline_successful() | ||
| { | ||
| var ctx = await Scenario.Define<Context>() | ||
| .WithEndpoint<Receiver>(b => | ||
| { | ||
| b.CustomConfig(c => | ||
| { | ||
| // Limiting the concurrency for this test to make sure messages that are made available again are | ||
| // not concurrently processed. This is not necessary for the test to pass but it makes | ||
| // reasoning about the test easier. | ||
| c.LimitMessageProcessingConcurrencyTo(1); | ||
| }); | ||
| b.When((session, _) => session.SendLocal(new MyMessage())); | ||
| }) | ||
| .Done(c => c.NativeMessageId is not null && c.Logs.Any(l => WasMarkedAsSuccessfullyCompleted(l, c))) | ||
| .Run(); | ||
|
|
||
| var items = ctx.Logs.Where(l => WasMarkedAsSuccessfullyCompleted(l, ctx)).ToArray(); | ||
|
|
||
| Assert.That(items, Is.Not.Empty); | ||
| } | ||
|
|
||
| [Test] | ||
| public async Task Should_complete_message_on_next_receive_when_error_pipeline_handled_the_message() | ||
| { | ||
| var ctx = await Scenario.Define<Context>(c => | ||
| { | ||
| c.ShouldThrow = true; | ||
| }) | ||
| .WithEndpoint<Receiver>(b => | ||
| { | ||
| b.DoNotFailOnErrorMessages(); | ||
| b.CustomConfig(c => | ||
| { | ||
| var recoverability = c.Recoverability(); | ||
| recoverability.AddUnrecoverableException<InvalidOperationException>(); | ||
|
|
||
| // Limiting the concurrency for this test to make sure messages that are made available again are | ||
| // not concurrently processed. This is not necessary for the test to pass but it makes | ||
| // reasoning about the test easier. | ||
| c.LimitMessageProcessingConcurrencyTo(1); | ||
| }); | ||
| b.When((session, _) => session.SendLocal(new MyMessage())); | ||
| }) | ||
| .Done(c => c.NativeMessageId is not null && c.Logs.Any(l => WasMarkedAsSuccessfullyCompleted(l, c))) | ||
| .Run(); | ||
|
|
||
| var items = ctx.Logs.Where(l => WasMarkedAsSuccessfullyCompleted(l, ctx)).ToArray(); | ||
|
|
||
| Assert.That(items, Is.Not.Empty); | ||
| } | ||
|
|
||
| static bool WasMarkedAsSuccessfullyCompleted(ScenarioContext.LogItem l, Context c) | ||
| => l.Message.StartsWith($"Received message with id '{c.NativeMessageId}' was marked as successfully completed"); | ||
|
|
||
| class Context : ScenarioContext | ||
| { | ||
| public bool ShouldThrow { get; set; } | ||
|
|
||
| public string NativeMessageId { get; set; } | ||
| } | ||
|
|
||
| class Receiver : EndpointConfigurationBuilder | ||
| { | ||
| public Receiver() => EndpointSetup<DefaultServer>(c => | ||
| { | ||
| var transport = c.ConfigureTransport<AzureServiceBusTransport>(); | ||
| // Explicitly setting the transport transaction mode to ReceiveOnly because the message | ||
| // tracking only is implemented for this mode. | ||
| transport.TransportTransactionMode = TransportTransactionMode.ReceiveOnly; | ||
| }); | ||
| } | ||
|
|
||
| public class MyMessage : IMessage | ||
| { | ||
| } | ||
|
|
||
| class MyMessageHandler : IHandleMessages<MyMessage> | ||
| { | ||
| readonly Context _testContext; | ||
|
|
||
| public MyMessageHandler(Context testContext) => _testContext = testContext; | ||
|
|
||
| public async Task Handle(MyMessage message, IMessageHandlerContext context) | ||
| { | ||
| var messageEventArgs = context.Extensions.Get<ProcessMessageEventArgs>(); | ||
| // By abandoning the message, the message will be "immediately available" for retrieval again and effectively the message pump | ||
| // has lost the message visibility timeout because any Complete or Abandon will be rejected by the azure service bus. | ||
| var serviceBusReceivedMessage = context.Extensions.Get<ServiceBusReceivedMessage>(); | ||
| await messageEventArgs.AbandonMessageAsync(serviceBusReceivedMessage); | ||
|
|
||
| _testContext.NativeMessageId = serviceBusReceivedMessage.MessageId; | ||
|
|
||
| if (_testContext.ShouldThrow) | ||
| { | ||
| throw new InvalidOperationException("Simulated exception"); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
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
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.