-
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.
In-memory persister does not throw concurrency exceptions (#4149)
* Failing test for concurrently processing saga using in-memory persistence * InMemorySagaPersister.Save is now concurrency safe * Fix the acceptance test * Addressed review comments
- Loading branch information
1 parent
bf207b3
commit d061342
Showing
3 changed files
with
174 additions
and
10 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
150 changes: 150 additions & 0 deletions
150
src/NServiceBus.AcceptanceTests/Sagas/When_saga_started_concurrently.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,150 @@ | ||
| namespace NServiceBus.AcceptanceTests.Sagas | ||
| { | ||
| using System; | ||
| using System.Threading.Tasks; | ||
| using AcceptanceTesting; | ||
| using EndpointTemplates; | ||
| using NUnit.Framework; | ||
|
|
||
| public class When_saga_started_concurrently : NServiceBusAcceptanceTest | ||
| { | ||
| [Test] | ||
| public async Task Should_start_single_saga() | ||
| { | ||
| var context = await Scenario.Define<Context>(c => { c.SomeId = Guid.NewGuid().ToString(); }) | ||
| .WithEndpoint<ConcurrentHandlerEndpoint>(b => | ||
| { | ||
| b.When((session, ctx) => | ||
| { | ||
| var t1 = session.SendLocal(new StartMessageOne | ||
| { | ||
| SomeId = ctx.SomeId | ||
| }); | ||
| var t2 = session.SendLocal(new StartMessageTwo | ||
| { | ||
| SomeId = ctx.SomeId | ||
| }); | ||
| return Task.WhenAll(t1, t2); | ||
| }); | ||
| }) | ||
| .Done(c => c.PlacedSagaId != Guid.Empty && c.BilledSagaId != Guid.Empty) | ||
| .Run(); | ||
|
|
||
| Assert.AreNotEqual(Guid.Empty, context.PlacedSagaId); | ||
| Assert.AreNotEqual(Guid.Empty, context.BilledSagaId); | ||
| Assert.AreEqual(context.PlacedSagaId, context.BilledSagaId, "Both messages should have been handled by the same saga, but SagaIds don't match."); | ||
| } | ||
|
|
||
| class Context : ScenarioContext | ||
| { | ||
| public string SomeId { get; set; } | ||
| public Guid PlacedSagaId { get; set; } | ||
| public Guid BilledSagaId { get; set; } | ||
| public bool SagaCompleted { get; set; } | ||
| } | ||
|
|
||
| class ConcurrentHandlerEndpoint : EndpointConfigurationBuilder | ||
| { | ||
| public ConcurrentHandlerEndpoint() | ||
| { | ||
| EndpointSetup<DefaultServer>(b => | ||
| { | ||
| b.LimitMessageProcessingConcurrencyTo(2); | ||
| b.Recoverability().Immediate(immediate => immediate.NumberOfRetries(3)); | ||
| }); | ||
| } | ||
|
|
||
| class ConcurrentlyStartedSaga : Saga<ConcurrentlyStartedSagaData>, | ||
| IAmStartedByMessages<StartMessageTwo>, | ||
| IAmStartedByMessages<StartMessageOne> | ||
| { | ||
| public Context Context { get; set; } | ||
|
|
||
| public async Task Handle(StartMessageOne message, IMessageHandlerContext context) | ||
| { | ||
| Data.Placed = true; | ||
| await context.SendLocal(new SuccessfulProcessing | ||
| { | ||
| SagaId = Data.Id, | ||
| Type = nameof(StartMessageOne) | ||
| }); | ||
| CheckForCompletion(context); | ||
| } | ||
|
|
||
| public async Task Handle(StartMessageTwo message, IMessageHandlerContext context) | ||
| { | ||
| Data.Billed = true; | ||
| await context.SendLocal(new SuccessfulProcessing | ||
| { | ||
| SagaId = Data.Id, | ||
| Type = nameof(StartMessageTwo) | ||
| }); | ||
| CheckForCompletion(context); | ||
| } | ||
|
|
||
| protected override void ConfigureHowToFindSaga(SagaPropertyMapper<ConcurrentlyStartedSagaData> mapper) | ||
| { | ||
| mapper.ConfigureMapping<StartMessageOne>(msg => msg.SomeId).ToSaga(saga => saga.OrderId); | ||
| mapper.ConfigureMapping<StartMessageTwo>(msg => msg.SomeId).ToSaga(saga => saga.OrderId); | ||
| } | ||
|
|
||
| void CheckForCompletion(IMessageHandlerContext context) | ||
| { | ||
| if (!Data.Billed || !Data.Placed) | ||
| { | ||
| return; | ||
| } | ||
| MarkAsComplete(); | ||
| Context.SagaCompleted = true; | ||
| } | ||
| } | ||
|
|
||
| class ConcurrentlyStartedSagaData : ContainSagaData | ||
| { | ||
| public virtual string OrderId { get; set; } | ||
| public virtual bool Placed { get; set; } | ||
| public virtual bool Billed { get; set; } | ||
| } | ||
|
|
||
| // Intercepts the messages sent out by the saga | ||
| class LogSuccessfulHandler : IHandleMessages<SuccessfulProcessing> | ||
| { | ||
| public Context Context { get; set; } | ||
|
|
||
| public Task Handle(SuccessfulProcessing message, IMessageHandlerContext context) | ||
| { | ||
| if (message.Type == nameof(StartMessageOne)) | ||
| { | ||
| Context.PlacedSagaId = message.SagaId; | ||
| } | ||
| else if (message.Type == nameof(StartMessageTwo)) | ||
| { | ||
| Context.BilledSagaId = message.SagaId; | ||
| } | ||
| else | ||
| { | ||
| throw new Exception("Unknown type"); | ||
| } | ||
|
|
||
| return Task.FromResult(0); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class StartMessageOne : ICommand | ||
| { | ||
| public string SomeId { get; set; } | ||
| } | ||
|
|
||
| class StartMessageTwo : ICommand | ||
| { | ||
| public string SomeId { get; set; } | ||
| } | ||
|
|
||
| class SuccessfulProcessing : ICommand | ||
| { | ||
| public string Type { get; set; } | ||
| public Guid SagaId { get; set; } | ||
| } | ||
| } | ||
| } |
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