forked from MassTransit/MassTransit
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add initial commit for adding dynamoDB
Completely refactored the original PR to comply with updated standards, and included a container test and configuration - PhatBoyG Updated DynamoDb with concurrency test and proper condition validation for Add/Insert and Update
- Loading branch information
Showing
26 changed files
with
1,471 additions
and
12 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| namespace MassTransit.TestFramework.Sagas | ||
| { | ||
| namespace ChoirConcurrency | ||
| { | ||
| using System; | ||
|
|
||
|
|
||
| public class RehearsalBegins | ||
| { | ||
| public Guid CorrelationId { get; set; } | ||
| } | ||
|
|
||
|
|
||
| public class Bass | ||
| { | ||
| public Guid CorrelationId { get; set; } | ||
| public string Name { get; set; } | ||
| } | ||
|
|
||
|
|
||
| public class Baritone | ||
| { | ||
| public Guid CorrelationId { get; set; } | ||
| public string Name { get; set; } | ||
| } | ||
|
|
||
|
|
||
| public class Tenor | ||
| { | ||
| public Guid CorrelationId { get; set; } | ||
| public string Name { get; set; } | ||
| } | ||
|
|
||
|
|
||
| public class Countertenor | ||
| { | ||
| public Guid CorrelationId { get; set; } | ||
| public string Name { get; set; } | ||
| } | ||
|
|
||
|
|
||
| public class ChoirState : | ||
| SagaStateMachineInstance, | ||
| ISagaVersion | ||
| { | ||
| public string CurrentState { get; set; } | ||
| public int Harmony { get; set; } | ||
|
|
||
| public string BassName { get; set; } | ||
| public string BaritoneName { get; set; } | ||
| public string TenorName { get; set; } | ||
| public string CountertenorName { get; set; } | ||
| public int Version { get; set; } | ||
|
|
||
| public Guid CorrelationId { get; set; } | ||
| } | ||
|
|
||
|
|
||
| public class ChoirStateMachine : | ||
| MassTransitStateMachine<ChoirState> | ||
| { | ||
| public ChoirStateMachine() | ||
| { | ||
| InstanceState(x => x.CurrentState); | ||
|
|
||
| Event(() => RehearsalStarts, x => x.CorrelateById(context => context.Message.CorrelationId)); | ||
| Event(() => BassStarts, x => x.CorrelateById(context => context.Message.CorrelationId)); | ||
| Event(() => BaritoneStarts, x => x.CorrelateById(context => context.Message.CorrelationId)); | ||
| Event(() => TenorStarts, x => x.CorrelateById(context => context.Message.CorrelationId)); | ||
| Event(() => CountertenorStarts, x => x.CorrelateById(context => context.Message.CorrelationId)); | ||
|
|
||
| CompositeEvent(() => AllSinging, x => x.Harmony, BassStarts, BaritoneStarts, TenorStarts, CountertenorStarts); | ||
|
|
||
| Initially( | ||
| When(RehearsalStarts) | ||
| .Then(context => Console.WriteLine("Rehearsal Started!!")) | ||
| .TransitionTo(Warmup)); | ||
|
|
||
| During(Warmup, | ||
| When(BassStarts) | ||
| .Then(context => Console.WriteLine("Bass Started!!")) | ||
| .Then(context => context.Saga.BassName = context.Message.Name), | ||
| When(BaritoneStarts) | ||
| .Then(context => Console.WriteLine("Baritone Started!!")) | ||
| .Then(context => context.Saga.BaritoneName = context.Message.Name), | ||
| When(TenorStarts) | ||
| .Then(context => Console.WriteLine("Tenor Started!!")) | ||
| .Then(context => context.Saga.TenorName = context.Message.Name), | ||
| When(CountertenorStarts) | ||
| .Then(context => Console.WriteLine("CounterTenor Started!!")) | ||
| .Then(context => context.Saga.CountertenorName = context.Message.Name), | ||
| When(AllSinging) | ||
| .Then(context => Console.WriteLine("Harmony Reached!!")) | ||
| .TransitionTo(Harmony)); | ||
| } | ||
|
|
||
| // | ||
| // ReSharper disable UnassignedGetOnlyAutoProperty | ||
| // ReSharper disable MemberCanBePrivate.Global | ||
| public Event<RehearsalBegins> RehearsalStarts { get; } | ||
| public Event<Bass> BassStarts { get; } | ||
| public Event<Baritone> BaritoneStarts { get; } | ||
| public Event<Tenor> TenorStarts { get; } | ||
| public Event<Countertenor> CountertenorStarts { get; } | ||
| public Event AllSinging { get; } | ||
| public State Warmup { get; } | ||
| public State Harmony { get; } | ||
| } | ||
| } | ||
| } |
59 changes: 59 additions & 0 deletions
59
...sit.DynamoDbIntegration/Configuration/Configuration/DynamoDbSagaRepositoryConfigurator.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,59 @@ | ||
| namespace MassTransit.Configuration | ||
| { | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using Amazon.DynamoDBv2.DataModel; | ||
| using DynamoDbIntegration.Saga; | ||
| using Microsoft.Extensions.DependencyInjection.Extensions; | ||
| using Saga; | ||
|
|
||
|
|
||
| public class DynamoDbSagaRepositoryConfigurator<TSaga> : | ||
| IDynamoDbSagaRepositoryConfigurator<TSaga>, | ||
| ISpecification | ||
| where TSaga : class, ISagaVersion | ||
| { | ||
| Func<IServiceProvider, IDynamoDBContext> _contextFactory; | ||
|
|
||
| public DynamoDbSagaRepositoryConfigurator() | ||
| { | ||
| LockTimeout = TimeSpan.FromMinutes(30); | ||
| } | ||
|
|
||
| public string TableName { get; set; } | ||
| public string LockSuffix { get; set; } | ||
| public TimeSpan LockTimeout { get; set; } | ||
| public TimeSpan? Expiration { get; set; } | ||
|
|
||
| public void ContextFactory(Func<IDynamoDBContext> contextFactory) | ||
| { | ||
| _contextFactory = provider => contextFactory(); | ||
| } | ||
|
|
||
| public void ContextFactory(Func<IServiceProvider, IDynamoDBContext> contextFactory) | ||
| { | ||
| _contextFactory = contextFactory; | ||
| } | ||
|
|
||
| public IEnumerable<ValidationResult> Validate() | ||
| { | ||
| if (_contextFactory == null) | ||
| yield return this.Failure("ContextFactory", "must be specified"); | ||
| if (string.IsNullOrWhiteSpace(TableName)) | ||
| yield return this.Failure("TableName", "must be specified"); | ||
| if (LockTimeout <= TimeSpan.Zero) | ||
| yield return this.Failure("LockTimeout", "Must be > TimeSpan.Zero"); | ||
| if (Expiration < TimeSpan.FromSeconds(30)) | ||
| yield return this.Failure("Expiration", "If specified, must be > 30 seconds"); | ||
| } | ||
|
|
||
| public void Register<T>(ISagaRepositoryRegistrationConfigurator<T> configurator) | ||
| where T : class, ISagaVersion | ||
| { | ||
| configurator.TryAddSingleton(_contextFactory); | ||
| configurator.TryAddSingleton(new DynamoDbSagaRepositoryOptions<T>(TableName, Expiration)); | ||
| configurator.RegisterSagaRepository<T, DatabaseContext<T>, SagaConsumeContextFactory<DatabaseContext<T>, T>, | ||
| DynamoDbSagaRepositoryContextFactory<T>>(); | ||
| } | ||
| } | ||
| } |
60 changes: 60 additions & 0 deletions
60
...moDbIntegration/Configuration/Configuration/DynamoDbSagaRepositoryRegistrationProvider.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,60 @@ | ||
| namespace MassTransit.Configuration | ||
| { | ||
| using System; | ||
| using Internals; | ||
|
|
||
|
|
||
| public class DynamoDbSagaRepositoryRegistrationProvider : | ||
| ISagaRepositoryRegistrationProvider | ||
| { | ||
| readonly Action<IDynamoDbSagaRepositoryConfigurator> _configure; | ||
|
|
||
| public DynamoDbSagaRepositoryRegistrationProvider(Action<IDynamoDbSagaRepositoryConfigurator> configure) | ||
| { | ||
| _configure = configure; | ||
| } | ||
|
|
||
| void ISagaRepositoryRegistrationProvider.Configure<TSaga>(ISagaRegistrationConfigurator<TSaga> configurator) | ||
| where TSaga : class | ||
| { | ||
| if (typeof(TSaga).HasInterface<ISagaVersion>()) | ||
| { | ||
| var proxy = (IProxy)Activator.CreateInstance(typeof(Proxy<>).MakeGenericType(typeof(TSaga)), configurator); | ||
|
|
||
| proxy.Configure(this); | ||
| } | ||
| } | ||
|
|
||
| protected virtual void Configure<TSaga>(ISagaRegistrationConfigurator<TSaga> configurator) | ||
| where TSaga : class, ISagaVersion | ||
| { | ||
| configurator.DynamoDbRepository(r => _configure?.Invoke(r)); | ||
| } | ||
|
|
||
|
|
||
| interface IProxy | ||
| { | ||
| public void Configure<T>(T provider) | ||
| where T : DynamoDbSagaRepositoryRegistrationProvider; | ||
| } | ||
|
|
||
|
|
||
| class Proxy<TSaga> : | ||
| IProxy | ||
| where TSaga : class, ISagaVersion | ||
| { | ||
| readonly ISagaRegistrationConfigurator<TSaga> _configurator; | ||
|
|
||
| public Proxy(ISagaRegistrationConfigurator<TSaga> configurator) | ||
| { | ||
| _configurator = configurator; | ||
| } | ||
|
|
||
| public void Configure<T>(T provider) | ||
| where T : DynamoDbSagaRepositoryRegistrationProvider | ||
| { | ||
| provider.Configure(_configurator); | ||
| } | ||
| } | ||
| } | ||
| } |
41 changes: 41 additions & 0 deletions
41
...Transit.DynamoDbIntegration/Configuration/DynamoDbSagaRepositoryRegistrationExtensions.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,41 @@ | ||
| namespace MassTransit | ||
| { | ||
| using System; | ||
| using Configuration; | ||
|
|
||
|
|
||
| public static class DynamoDbSagaRepositoryRegistrationExtensions | ||
| { | ||
| /// <summary> | ||
| /// Adds a DynamoDb saga repository to the registration | ||
| /// </summary> | ||
| /// <param name="configurator"></param> | ||
| /// <param name="configure"></param> | ||
| /// <typeparam name="T"></typeparam> | ||
| /// <returns></returns> | ||
| public static ISagaRegistrationConfigurator<T> DynamoDbRepository<T>(this ISagaRegistrationConfigurator<T> configurator, | ||
| Action<IDynamoDbSagaRepositoryConfigurator<T>> configure = null) | ||
| where T : class, ISagaVersion | ||
| { | ||
| var repositoryConfigurator = new DynamoDbSagaRepositoryConfigurator<T>(); | ||
|
|
||
| configure?.Invoke(repositoryConfigurator); | ||
|
|
||
| repositoryConfigurator.Validate().ThrowIfContainsFailure("The DynamoDb saga repository configuration is invalid:"); | ||
|
|
||
| configurator.Repository(x => repositoryConfigurator.Register(x)); | ||
|
|
||
| return configurator; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Use the DynamoDb saga repository for sagas configured by type (without a specific generic call to AddSaga/AddSagaStateMachine) | ||
| /// </summary> | ||
| /// <param name="configurator"></param> | ||
| /// <param name="configure"></param> | ||
| public static void SetDynamoDbSagaRepositoryProvider(this IRegistrationConfigurator configurator, Action<IDynamoDbSagaRepositoryConfigurator> configure) | ||
| { | ||
| configurator.SetSagaRepositoryProvider(new DynamoDbSagaRepositoryRegistrationProvider(configure)); | ||
| } | ||
| } | ||
| } |
33 changes: 33 additions & 0 deletions
33
...ence/MassTransit.DynamoDbIntegration/Configuration/IDynamoDbSagaRepositoryConfigurator.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,33 @@ | ||
| namespace MassTransit | ||
| { | ||
| using System; | ||
| using Amazon.DynamoDBv2.DataModel; | ||
|
|
||
|
|
||
| public interface IDynamoDbSagaRepositoryConfigurator | ||
| { | ||
| string TableName { set; } | ||
| string LockSuffix { set; } | ||
| TimeSpan LockTimeout { set; } | ||
| TimeSpan? Expiration { set; } | ||
|
|
||
| /// <summary> | ||
| /// Factory method to get the DynamoDb context | ||
| /// </summary> | ||
| /// <param name="contextFactory"></param> | ||
| void ContextFactory(Func<IDynamoDBContext> contextFactory); | ||
|
|
||
| /// <summary> | ||
| /// Use the container to build the DynamoDb context | ||
| /// </summary> | ||
| /// <param name="contextFactory"></param> | ||
| void ContextFactory(Func<IServiceProvider, IDynamoDBContext> contextFactory); | ||
| } | ||
|
|
||
|
|
||
| public interface IDynamoDbSagaRepositoryConfigurator<TSaga> : | ||
| IDynamoDbSagaRepositoryConfigurator | ||
| where TSaga : class, ISagaVersion | ||
| { | ||
| } | ||
| } |
21 changes: 21 additions & 0 deletions
21
src/Persistence/MassTransit.DynamoDbIntegration/DynamoDbIntegration/Saga/DatabaseContext.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,21 @@ | ||
| namespace MassTransit.DynamoDbIntegration.Saga | ||
| { | ||
| using System; | ||
| using System.Threading.Tasks; | ||
|
|
||
|
|
||
| public interface DatabaseContext<TSaga> : | ||
| IDisposable | ||
| where TSaga : class, ISagaVersion | ||
| { | ||
| Task Add(SagaConsumeContext<TSaga> context); | ||
|
|
||
| Task Insert(TSaga instance); | ||
|
|
||
| Task<TSaga> Load(Guid correlationId); | ||
|
|
||
| Task Update(SagaConsumeContext<TSaga> context); | ||
|
|
||
| Task Delete(SagaConsumeContext<TSaga> context); | ||
| } | ||
| } |
Oops, something went wrong.