diff --git a/Samples.sln b/Samples.sln index b2e63dedd..da1b27ebc 100644 --- a/Samples.sln +++ b/Samples.sln @@ -121,9 +121,13 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sample22", "src\samples\Sam EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sample23", "src\samples\Sample23\Sample23.csproj", "{5F16E2EC-7162-47F8-B8F1-81CACB05A5D9}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sample24", "src\samples\Sample24\Sample24.csproj", "{C7B1C36D-7297-4AA2-8AFE-31FC5CB99AA0}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sample24", "src\samples\Sample24\Sample24.csproj", "{C7B1C36D-7297-4AA2-8AFE-31FC5CB99AA0}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sample25", "src\samples\Sample25\Sample25.csproj", "{ADDA6EB3-32F0-4802-B595-B12090DC4A1C}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sample25", "src\samples\Sample25\Sample25.csproj", "{ADDA6EB3-32F0-4802-B595-B12090DC4A1C}" +EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sample26", "src\samples\Sample26\Sample26.csproj", "{61B0A047-844E-430C-A83F-DCF2C3E2BD7D}" +EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sample27", "src\samples\Sample27\Sample27.csproj", "{AF45B023-759B-4CAF-BA88-EB0505B03447}" EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution @@ -319,6 +323,14 @@ Global {ADDA6EB3-32F0-4802-B595-B12090DC4A1C}.Debug|Any CPU.Build.0 = Debug|Any CPU {ADDA6EB3-32F0-4802-B595-B12090DC4A1C}.Release|Any CPU.ActiveCfg = Release|Any CPU {ADDA6EB3-32F0-4802-B595-B12090DC4A1C}.Release|Any CPU.Build.0 = Release|Any CPU + {61B0A047-844E-430C-A83F-DCF2C3E2BD7D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {61B0A047-844E-430C-A83F-DCF2C3E2BD7D}.Debug|Any CPU.Build.0 = Debug|Any CPU + {61B0A047-844E-430C-A83F-DCF2C3E2BD7D}.Release|Any CPU.ActiveCfg = Release|Any CPU + {61B0A047-844E-430C-A83F-DCF2C3E2BD7D}.Release|Any CPU.Build.0 = Release|Any CPU + {AF45B023-759B-4CAF-BA88-EB0505B03447}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {AF45B023-759B-4CAF-BA88-EB0505B03447}.Debug|Any CPU.Build.0 = Debug|Any CPU + {AF45B023-759B-4CAF-BA88-EB0505B03447}.Release|Any CPU.ActiveCfg = Release|Any CPU + {AF45B023-759B-4CAF-BA88-EB0505B03447}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -377,6 +389,8 @@ Global {5F16E2EC-7162-47F8-B8F1-81CACB05A5D9} = {5E5E1E84-DDBC-40D6-B891-0D563A15A44A} {C7B1C36D-7297-4AA2-8AFE-31FC5CB99AA0} = {5E5E1E84-DDBC-40D6-B891-0D563A15A44A} {ADDA6EB3-32F0-4802-B595-B12090DC4A1C} = {5E5E1E84-DDBC-40D6-B891-0D563A15A44A} + {61B0A047-844E-430C-A83F-DCF2C3E2BD7D} = {5E5E1E84-DDBC-40D6-B891-0D563A15A44A} + {AF45B023-759B-4CAF-BA88-EB0505B03447} = {5E5E1E84-DDBC-40D6-B891-0D563A15A44A} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {8B0975FD-7050-48B0-88C5-48C33378E158} diff --git a/src/activities/Elsa.Activities.MassTransit/DefaultAzureServiceBusMassTransitBuilder.cs b/src/activities/Elsa.Activities.MassTransit/DefaultAzureServiceBusMassTransitBuilder.cs new file mode 100644 index 000000000..dfa892eb6 --- /dev/null +++ b/src/activities/Elsa.Activities.MassTransit/DefaultAzureServiceBusMassTransitBuilder.cs @@ -0,0 +1,48 @@ +using System; +using MassTransit; +using MassTransit.Azure.ServiceBus.Core; +using MassTransit.ExtensionsDependencyInjectionIntegration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Elsa.Activities.MassTransit.Options; +using Elsa.Activities.MassTransit.Extensions; +using System.Collections.Generic; + +namespace Elsa.Activities.MassTransit +{ + public sealed class DefaultAzureServiceBusMassTransitBuilder : MassTransitBuilderBase + { + protected override IBusControl CreateBus(IServiceProvider serviceProvider) + { + return Bus.Factory.CreateUsingAzureServiceBus(bus => + { + var options = serviceProvider.GetRequiredService>().Value; + + bus.Host(options.ConnectionString); + + foreach (var messageType in MessageTypes) + { + var queueName = messageType.Name; + var consumerType = messageType.CreateConsumerType(); + + bus.ReceiveEndpoint(queueName, endpoint => + { + endpoint.PrefetchCount = 16; + endpoint.ConfigureConsumer(serviceProvider, consumerType); + messageType.MapEndpointConvention(endpoint.InputAddress); + }); + } + }); + } + + protected override void ConfigureMassTransit(IServiceCollectionConfigurator configurator) + { + foreach (var messageType in MessageTypes) + { + configurator.AddConsumer(messageType.CreateConsumerType()); + } + } + + public IEnumerable MessageTypes { get; set; } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.MassTransit/DefaultRabbitMqMassTransitBuilder.cs b/src/activities/Elsa.Activities.MassTransit/DefaultRabbitMqMassTransitBuilder.cs new file mode 100644 index 000000000..b03c4bac3 --- /dev/null +++ b/src/activities/Elsa.Activities.MassTransit/DefaultRabbitMqMassTransitBuilder.cs @@ -0,0 +1,57 @@ +using System; +using System.Collections.Generic; +using Elsa.Activities.MassTransit.Extensions; +using Elsa.Activities.MassTransit.Options; +using MassTransit; +using MassTransit.ExtensionsDependencyInjectionIntegration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Elsa.Activities.MassTransit +{ + public sealed class DefaultRabbitMqMassTransitBuilder : MassTransitBuilderBase + { + protected override IBusControl CreateBus(IServiceProvider serviceProvider) + { + return Bus.Factory.CreateUsingRabbitMq(bus => + { + var options = serviceProvider.GetRequiredService>().Value; + var host = bus.Host(new Uri(options.Host), h => + { + if (!string.IsNullOrEmpty(options.Username)) + { + h.Username(options.Username); + + if (!string.IsNullOrEmpty(options.Password)) + { + h.Password(options.Password); + } + } + }); + + foreach (var messageType in MessageTypes) + { + var queueName = messageType.Name; + var consumerType = messageType.CreateConsumerType(); + + bus.ReceiveEndpoint(queueName, endpoint => + { + endpoint.PrefetchCount = 16; + endpoint.ConfigureConsumer(serviceProvider, consumerType); + messageType.MapEndpointConvention(endpoint.InputAddress); + }); + } + }); + } + + protected override void ConfigureMassTransit(IServiceCollectionConfigurator configurator) + { + foreach (var messageType in MessageTypes) + { + configurator.AddConsumer(messageType.CreateConsumerType()); + } + } + + public IEnumerable MessageTypes { get; set; } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.MassTransit/Elsa.Activities.MassTransit.csproj b/src/activities/Elsa.Activities.MassTransit/Elsa.Activities.MassTransit.csproj index 7ee16406b..4001e9b6a 100644 --- a/src/activities/Elsa.Activities.MassTransit/Elsa.Activities.MassTransit.csproj +++ b/src/activities/Elsa.Activities.MassTransit/Elsa.Activities.MassTransit.csproj @@ -29,6 +29,7 @@ + diff --git a/src/activities/Elsa.Activities.MassTransit/Extensions/ServiceCollectionExtensions.cs b/src/activities/Elsa.Activities.MassTransit/Extensions/ServiceCollectionExtensions.cs index efee3f37d..d96390c98 100644 --- a/src/activities/Elsa.Activities.MassTransit/Extensions/ServiceCollectionExtensions.cs +++ b/src/activities/Elsa.Activities.MassTransit/Extensions/ServiceCollectionExtensions.cs @@ -1,12 +1,9 @@ using System; -using System.Collections.Generic; using Elsa.Activities.MassTransit.Activities; using Elsa.Activities.MassTransit.Consumers; using Elsa.Activities.MassTransit.Options; using MassTransit; -using MassTransit.AspNetCoreIntegration; using MassTransit.ConsumeConfigurators; -using MassTransit.ExtensionsDependencyInjectionIntegration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; @@ -14,64 +11,27 @@ namespace Elsa.Activities.MassTransit.Extensions { public static class ServiceCollectionExtensions { - public static IServiceCollection AddMassTransitActivities(this IServiceCollection services) + public static IServiceCollection AddMassTransitSchedulingActivities(this IServiceCollection services, + IMassTransitBuilder massTransitBuilder, + Action> options) where TOptions : class { - if (services == null) - { - throw new ArgumentNullException(nameof(services)); - } - - return services - .AddActivity() - .AddActivity() - .AddActivity(); - } - - public static IServiceCollection AddMassTransitSchedulingActivities(this IServiceCollection services, Action configureOptions) - { - if (services == null) - { - throw new ArgumentNullException(nameof(services)); - } - - if (configureOptions == null) - { - throw new ArgumentNullException(nameof(configureOptions)); - } + var optionsBuilder = services.AddOptions(); + options?.Invoke(optionsBuilder); services.AddMassTransitActivities() .AddActivity() .AddActivity(); - services.Configure(configureOptions); + massTransitBuilder.Build(services); return services; } - public static IServiceCollection AddRabbitMqActivities(this IServiceCollection services, Action> options = null, params Type[] messageTypes) + public static IServiceCollection AddMassTransitActivities(this IServiceCollection services, + IMassTransitBuilder massTransitBuilder) where TOptions : class { - var optionsBuilder = services.AddOptions(); - options?.Invoke(optionsBuilder); - - services - .AddMassTransitActivities() - .AddMassTransit(CreateBus, ConfigureMassTransit); - + services.AddMassTransitActivities(); + massTransitBuilder.Build(services); return services; - - // local function to configure consumers - void ConfigureMassTransit(IServiceCollectionConfigurator configurator) - { - foreach (var messageType in messageTypes) - { - configurator.AddConsumer(CreateConsumerType(messageType)); - } - } - - // local function to create the bus - IBusControl CreateBus(IServiceProvider sp) - { - return CreateUsingRabbitMq(sp, messageTypes); - } } public static void ConfigureWorkflowConsumer( @@ -93,51 +53,17 @@ namespace Elsa.Activities.MassTransit.Extensions return configurator.AddConsumer(configure); } - private static IBusControl CreateUsingRabbitMq(IServiceProvider sp, IEnumerable messageTypes) + private static IServiceCollection AddMassTransitActivities(this IServiceCollection services) { - return Bus.Factory.CreateUsingRabbitMq( - bus => - { - var options = sp.GetRequiredService>().Value; - var host = bus.Host(new Uri(options.Host), h => - { - if (!string.IsNullOrEmpty(options.Username)) - { - h.Username(options.Username); + if (services == null) + { + throw new ArgumentNullException(nameof(services)); + } - if (!string.IsNullOrEmpty(options.Password)) - h.Password(options.Password); - } - }); - - - foreach (var messageType in messageTypes) - { - var queueName = messageType.Name; - var consumerType = CreateConsumerType(messageType); - - bus.ReceiveEndpoint( - queueName, - endpoint => - { - endpoint.PrefetchCount = 16; - endpoint.ConfigureConsumer(sp, consumerType); - MapEndpointConvention(messageType, endpoint.InputAddress); - }); - } - }); - } - - private static void MapEndpointConvention(Type messageType, Uri destinationAddress) - { - var method = typeof(EndpointConvention).GetMethod("Map", new[]{ typeof(Uri) }); - var generic = method.MakeGenericMethod(messageType); - generic.Invoke(null, new object[]{destinationAddress}); - } - - private static Type CreateConsumerType(Type messageType) - { - return typeof(WorkflowConsumer<>).MakeGenericType(messageType); + return services + .AddActivity() + .AddActivity() + .AddActivity(); } } } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.MassTransit/Extensions/TypeExtensions.cs b/src/activities/Elsa.Activities.MassTransit/Extensions/TypeExtensions.cs new file mode 100644 index 000000000..596971088 --- /dev/null +++ b/src/activities/Elsa.Activities.MassTransit/Extensions/TypeExtensions.cs @@ -0,0 +1,21 @@ +using Elsa.Activities.MassTransit.Consumers; +using MassTransit; +using System; + +namespace Elsa.Activities.MassTransit.Extensions +{ + internal static class TypeExtensions + { + public static void MapEndpointConvention(this Type messageType, Uri destinationAddress) + { + var method = typeof(EndpointConvention).GetMethod("Map", new[] { typeof(Uri) }); + var generic = method.MakeGenericMethod(messageType); + generic.Invoke(null, new object[] { destinationAddress }); + } + + public static Type CreateConsumerType(this Type messageType) + { + return typeof(WorkflowConsumer<>).MakeGenericType(messageType); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.MassTransit/IMassTransitBuilder.cs b/src/activities/Elsa.Activities.MassTransit/IMassTransitBuilder.cs new file mode 100644 index 000000000..abed4c97f --- /dev/null +++ b/src/activities/Elsa.Activities.MassTransit/IMassTransitBuilder.cs @@ -0,0 +1,9 @@ +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Activities.MassTransit +{ + public interface IMassTransitBuilder where TOptions : class + { + IServiceCollection Build(IServiceCollection services); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.MassTransit/MassTransitBuilderBase.cs b/src/activities/Elsa.Activities.MassTransit/MassTransitBuilderBase.cs new file mode 100644 index 000000000..d6e61ee78 --- /dev/null +++ b/src/activities/Elsa.Activities.MassTransit/MassTransitBuilderBase.cs @@ -0,0 +1,27 @@ +using System; +using MassTransit; +using MassTransit.AspNetCoreIntegration; +using MassTransit.ExtensionsDependencyInjectionIntegration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Elsa.Activities.MassTransit +{ + public abstract class MassTransitBuilderBase : IMassTransitBuilder where TOptions : class + { + protected MassTransitBuilderBase() { } + + public IServiceCollection Build(IServiceCollection services) + { + var optionsBuilder = services.AddOptions(); + Options?.Invoke(optionsBuilder); + + return services.AddMassTransit(CreateBus, ConfigureMassTransit); + } + + protected abstract void ConfigureMassTransit(IServiceCollectionConfigurator configurator); + protected abstract IBusControl CreateBus(IServiceProvider serviceProvider); + + public Action> Options { get; set; } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.MassTransit/Options/AzureServiceBusOptions.cs b/src/activities/Elsa.Activities.MassTransit/Options/AzureServiceBusOptions.cs new file mode 100644 index 000000000..b9e33faf9 --- /dev/null +++ b/src/activities/Elsa.Activities.MassTransit/Options/AzureServiceBusOptions.cs @@ -0,0 +1,10 @@ +using Microsoft.Extensions.Hosting; + +namespace Elsa.Activities.MassTransit.Options +{ + public class AzureServiceBusOptions + { + public string Host { get; set; } + public string ConnectionString { get; set; } + } +} \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs b/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs index 7a4961b0c..9cb5dbd45 100644 --- a/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs +++ b/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs @@ -1,4 +1,4 @@ -using Microsoft.Azure.Documents; +using Microsoft.Azure.Documents; using Microsoft.Azure.Documents.Client; using Microsoft.Extensions.Options; using Newtonsoft.Json; @@ -17,7 +17,7 @@ namespace Elsa.Persistence.DocumentDb { private readonly SemaphoreSlim clientInstanceLock; private readonly DocumentDbStorageOptions options; - private DocumentClient client; + private readonly DocumentClient client; private Dictionary collectionUris; public DocumentDbStorage(IOptions options) @@ -69,8 +69,6 @@ namespace Elsa.Persistence.DocumentDb Id = options.DatabaseName }); - - var tasks = options.CollectionNames.Select(async collectionName => { var databaseUri = UriFactory.CreateDatabaseUri(database.Resource.Id); diff --git a/src/samples/README.md b/src/samples/README.md index bab3f92a5..517cb928a 100644 --- a/src/samples/README.md +++ b/src/samples/README.md @@ -9,7 +9,7 @@ This list of sample implementations showcase a variety of workflows using ELSA. 5. [Recurring](Sample05) - A minimal workflows program defined in code with a strongly-typed workflow class. 6. [HttpRequest/HttpResponse](Sample06) - Example of a workflow that intercepts a HttpRequest and handles the HttpResponse. 7. [Document Approval](Sample07) - Example of a Document Approval workflow. -8. [Create and Handle Orders](Sample08) - Example of a workflow that creates and handles orders. +8. [Create and Handle Orders](Sample08) - Example of a workflow that creates and handles orders. Uses Mass Transit and RabbitMq. 9. [Calculator: Defined in Json](Sample9) - A simple console program that loads & executes a workflow designed with the HTML5 workflow designer. 10. [Hellow World: YesSql](Sample10) - A simple demonstration of using YesSql persistence providers. 11. [Correlation](Sample11) - Demonstrates workflow correlation. @@ -22,8 +22,10 @@ This list of sample implementations showcase a variety of workflows using ELSA. 18. [Email Reminder: Console Application](Sample18) - Example of a workflow that sends email reminders on a timer until a condition is met. 19. [ELSA Dashboard: Web App, SqlLite Persistence](Sample19) - Example of created a web application that hosts a ELSA Dashboard. 20. [Reflection Activities](Sample20) - Demonstrates Reflection activities. -21. [Shopping Cart: Mass Transit & Quartz Scheduling](Sample21) - Example of a workflow that tracks a shopping cart. Uses Mass Transit and Quartz scheduling. +21. [Shopping Cart: Mass Transit & Quartz Scheduling](Sample21) - Example of a workflow that tracks a shopping cart. Uses Mass Transit, RabbitMq and Quartz scheduling. 22. [Hellow World: SQL Server](Sample22) - A simple demonstration of using Entity Framework Core persistence providers with SQL Server. 23. [Hello World Custom Schema: Sqllite](Sample23) - A simple demonstration of using Entity Framework Core persistence providers with SqlLite and Custom Schema. 24. [Hello World: MySql with migration](Sample24) - A simple demonstration of using Entity Framework Core MySql persistence provider. A migration is also demonstrated. -24. [Data processing](Sample25) - A simple demonstration of flowing data through a pipeline of processing activities (https://github.com/elsa-workflows/elsa-core/issues/405). \ No newline at end of file +25. [Data processing](Sample25) - A simple demonstration of flowing data through a pipeline of processing activities (https://github.com/elsa-workflows/elsa-core/issues/405). +26. [Create and Handle Orders](Sample26) - Example of a workflow that creates and handles orders. Uses Mass Transit and Azure Service Bus. +27. [Shopping Cart: Mass Transit & Quartz Scheduling](Sample27) - Example of a workflow that tracks a shopping cart. Uses Mass Transit, Azure Service Bus and Quartz scheduling. \ No newline at end of file diff --git a/src/samples/Sample08/Sample08.csproj b/src/samples/Sample08/Sample08.csproj index 6f1e10515..53a6c0441 100644 --- a/src/samples/Sample08/Sample08.csproj +++ b/src/samples/Sample08/Sample08.csproj @@ -4,10 +4,6 @@ netcoreapp3.1 - - - - diff --git a/src/samples/Sample08/Startup.cs b/src/samples/Sample08/Startup.cs index 0cfbe4595..4847e69fb 100644 --- a/src/samples/Sample08/Startup.cs +++ b/src/samples/Sample08/Startup.cs @@ -1,6 +1,8 @@ -using Elsa.Activities.Email.Extensions; +using Elsa.Activities.Email.Extensions; using Elsa.Activities.Http.Extensions; +using Elsa.Activities.MassTransit; using Elsa.Activities.MassTransit.Extensions; +using Elsa.Activities.MassTransit.Options; using Elsa.Activities.Timers.Extensions; using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.Configuration; @@ -23,15 +25,22 @@ namespace Sample08 public void ConfigureServices(IServiceCollection services) { + var massTransitBuilder = new DefaultRabbitMqMassTransitBuilder + { + MessageTypes = new[] + { + typeof(CreateOrder), + typeof(OrderShipped) + }, + Options = o => o.Bind(Configuration.GetSection("MassTransit:RabbitMq")) + }; + services .AddElsa() .AddHttpActivities() .AddTimerActivities(options => options.Configure(x => x.SweepInterval = Duration.FromSeconds(10))) .AddEmailActivities(options => options.Bind(Configuration.GetSection("Smtp"))) - .AddRabbitMqActivities( - options => options.Bind(Configuration.GetSection("MassTransit:RabbitMq")), - typeof(CreateOrder), - typeof(OrderShipped)) + .AddMassTransitActivities(massTransitBuilder) .AddWorkflow() .AddWorkflow(); } diff --git a/src/samples/Sample21/RabbitMqSchedulerMassTransitBuilder.cs b/src/samples/Sample21/RabbitMqSchedulerMassTransitBuilder.cs new file mode 100644 index 000000000..f2ffc5a5c --- /dev/null +++ b/src/samples/Sample21/RabbitMqSchedulerMassTransitBuilder.cs @@ -0,0 +1,97 @@ +using System; +using Elsa.Activities.MassTransit.Extensions; +using GreenPipes; +using MassTransit; +using MassTransit.ExtensionsDependencyInjectionIntegration; +using MassTransit.QuartzIntegration; +using Quartz; +using Sample21.Consumers; +using Sample21.Messages; +using Elsa.Activities.MassTransit; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Sample21 +{ + internal class RabbitMqSchedulerMassTransitBuilder : MassTransitBuilderBase + { + private readonly IScheduler scheduler; + + public RabbitMqSchedulerMassTransitBuilder(IScheduler scheduler) + { + this.scheduler = scheduler; + } + + protected override IBusControl CreateBus(IServiceProvider serviceProvider) + { + var bus = Bus.Factory.CreateUsingRabbitMq(cfg => + { + var options = serviceProvider.GetRequiredService>().Value; + var host = cfg.Host(new Uri(options.Host), h => + { + if (!string.IsNullOrEmpty(options.Username)) + { + h.Username(options.Username); + + if (!string.IsNullOrEmpty(options.Password)) + { + h.Password(options.Password); + } + } + }); + + cfg.UseMessageScheduler(options.MessageSchedule.SchedulerAddress); + + cfg.ReceiveEndpoint("shopping_cart_service", ep => + { + ep.ConfigureConsumer(serviceProvider); + }); + + cfg.ReceiveEndpoint("shopping_cart_state", ep => + { + ep.PrefetchCount = 16; + + ep.UseMessageRetry(r => r.Interval(2, 100)); + + // Consume all workflow messages from the same queue. + ep.ConfigureWorkflowConsumer(serviceProvider); + ep.ConfigureWorkflowConsumer(serviceProvider); + ep.ConfigureWorkflowConsumer(serviceProvider); + ep.ConfigureWorkflowConsumer(serviceProvider); + }); + + // Should use external process scheduler service + // https://github.com/MassTransit/MassTransit/tree/develop/src/Samples/MassTransit.QuartzService + cfg.ReceiveEndpoint("sample_quartz_scheduler", e => + { + // For MT4.0, prefetch must be set for Quartz prior to anything else + e.PrefetchCount = 1; + cfg.UseMessageScheduler(e.InputAddress); + + e.ConfigureConsumer(serviceProvider); + e.ConfigureConsumer(serviceProvider); + }); + }); + + scheduler.JobFactory = new MassTransitJobFactory(bus); + + return bus; + } + + protected override void ConfigureMassTransit(IServiceCollectionConfigurator configurator) + { + // Configure scheduler service consumers. + configurator.AddConsumer(); + configurator.AddConsumer(); + + // configure workflow consumers + configurator.AddWorkflowConsumer(); + configurator.AddWorkflowConsumer(); + configurator.AddWorkflowConsumer(); + configurator.AddWorkflowConsumer(); + + // host fake service consumers + configurator.AddConsumer(); + } + } +} \ No newline at end of file diff --git a/src/samples/Sample21/RabbitMqSchedulerOptions.cs b/src/samples/Sample21/RabbitMqSchedulerOptions.cs new file mode 100644 index 000000000..331656081 --- /dev/null +++ b/src/samples/Sample21/RabbitMqSchedulerOptions.cs @@ -0,0 +1,9 @@ +using Elsa.Activities.MassTransit.Options; + +namespace Sample21 +{ + public class RabbitMqSchedulerOptions : RabbitMqOptions + { + public MessageScheduleOptions MessageSchedule { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample21/Sample21.csproj b/src/samples/Sample21/Sample21.csproj index 2b5108984..509f79ee3 100644 --- a/src/samples/Sample21/Sample21.csproj +++ b/src/samples/Sample21/Sample21.csproj @@ -1,14 +1,10 @@ - + netcoreapp3.1 - - - - all diff --git a/src/samples/Sample21/Startup.cs b/src/samples/Sample21/Startup.cs index 1832f0ef6..7cf6c1520 100644 --- a/src/samples/Sample21/Startup.cs +++ b/src/samples/Sample21/Startup.cs @@ -4,11 +4,6 @@ using Elsa.Activities.Email.Extensions; using Elsa.Activities.Http.Extensions; using Elsa.Activities.MassTransit.Extensions; using Elsa.Activities.Timers.Extensions; -using GreenPipes; -using MassTransit; -using MassTransit.ExtensionsDependencyInjectionIntegration; -using MassTransit.AspNetCoreIntegration; -using MassTransit.QuartzIntegration; using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; @@ -17,9 +12,7 @@ using NodaTime; using Quartz; using Quartz.Impl; using Quartz.Logging; -using Sample21.Consumers; using Sample21.Controllers; -using Sample21.Messages; using Sample21.Services; using Sample21.Workflows; @@ -40,92 +33,21 @@ namespace Sample21 .AddControllers(); var scheduler = CreateScheduler(); + var massTransitBuilder = new RabbitMqSchedulerMassTransitBuilder(scheduler) + { + Options = o => o.Bind(Configuration.GetSection("MassTransit:RabbitMq")) + }; services .AddElsa() .AddHttpActivities() .AddTimerActivities(options => options.Configure(x => x.SweepInterval = Duration.FromSeconds(10))) .AddEmailActivities(options => options.Bind(Configuration.GetSection("Smtp"))) - .AddMassTransitSchedulingActivities(options => - { - options.SchedulerAddress = new Uri("rabbitmq://localhost/sample_quartz_scheduler"); - }) + .AddMassTransitSchedulingActivities(massTransitBuilder, options => options.Bind(Configuration.GetSection("MassTransit:RabbitMq:MessageSchedule"))) .AddWorkflow() - .AddScoped() - .AddSingleton(scheduler) - - // configures MassTransit to integrate with the built-in dependency injection - .AddMassTransit(CreateBus, ConfigureMassTransit) - - // Add a hosted service to stat and stop the quartz scheduler - .AddSingleton(); - - void ConfigureMassTransit(IServiceCollectionConfigurator configurator) - { - // Configure scheduler service consumers. - configurator.AddConsumer(); - configurator.AddConsumer(); - - // configure workflow consumers - configurator.AddWorkflowConsumer(); - configurator.AddWorkflowConsumer(); - configurator.AddWorkflowConsumer(); - configurator.AddWorkflowConsumer(); - - // host fake service consumers - configurator.AddConsumer(); - } - - // local function to create the bus - IBusControl CreateBus(IServiceProvider serviceProvider) - { - var bus = Bus.Factory.CreateUsingRabbitMq(cfg => - { - var host = cfg.Host(new Uri("rabbitmq://localhost"), h => - { - h.Username("guest"); - h.Password("guest"); - }); - - cfg.UseMessageScheduler(new Uri("rabbitmq://localhost/sample_quartz_scheduler")); - - cfg.ReceiveEndpoint("shopping_cart_service", ep => - { - ep.ConfigureConsumer(serviceProvider); - }); - - cfg.ReceiveEndpoint("shopping_cart_state", ep => - { - ep.PrefetchCount = 16; - - ep.UseMessageRetry(r => r.Interval(2, 100)); - - // Consume all workflow messages from the same queue. - ep.ConfigureWorkflowConsumer(serviceProvider); - ep.ConfigureWorkflowConsumer(serviceProvider); - ep.ConfigureWorkflowConsumer(serviceProvider); - ep.ConfigureWorkflowConsumer(serviceProvider); - }); - - // Should use external process scheduler service - // https://github.com/MassTransit/MassTransit/tree/develop/src/Samples/MassTransit.QuartzService - cfg.ReceiveEndpoint("sample_quartz_scheduler", e => - { - // For MT4.0, prefetch must be set for Quartz prior to anything else - e.PrefetchCount = 1; - cfg.UseMessageScheduler(e.InputAddress); - - e.ConfigureConsumer(serviceProvider); - e.ConfigureConsumer(serviceProvider); - }); - }); - - scheduler.JobFactory = new MassTransitJobFactory(bus); - - return bus; - } + .AddSingleton(); // Add a hosted service to stat and stop the quartz scheduler IScheduler CreateScheduler() { @@ -166,7 +88,7 @@ namespace Sample21 { return (level, func, exception, parameters) => { - if (level >= Quartz.Logging.LogLevel.Debug && func != null) + if (level >= LogLevel.Debug && func != null) { Console.WriteLine("[" + DateTime.Now.ToLongTimeString() + "] [" + level + "] " + func(), parameters); } diff --git a/src/samples/Sample21/Workflows/CartTrackingWorkflow.cs b/src/samples/Sample21/Workflows/CartTrackingWorkflow.cs index a3294ebff..471e733a4 100644 --- a/src/samples/Sample21/Workflows/CartTrackingWorkflow.cs +++ b/src/samples/Sample21/Workflows/CartTrackingWorkflow.cs @@ -5,6 +5,7 @@ using Elsa.Activities.MassTransit.Activities; using Elsa.Scripting.JavaScript; using Elsa.Services; using Elsa.Services.Models; +using Microsoft.Extensions.Options; using NodaTime; using Sample21.Messages; @@ -12,6 +13,13 @@ namespace Sample21.Workflows { public class CartTrackingWorkflow : IWorkflow { + private readonly RabbitMqSchedulerOptions options; + + public CartTrackingWorkflow(IOptions options) + { + this.options = options.Value; + } + public void Build(IWorkflowBuilder builder) { builder.StartWith(activity => activity.MessageType = typeof(CartCreated)) @@ -34,7 +42,7 @@ namespace Sample21.Workflows activity => { activity.MessageType = typeof(CartExpiredEvent); - activity.EndpointAddress = new Uri("rabbitmq://localhost/shopping_cart_state"); + activity.EndpointAddress = new Uri($"{options.Host}/shopping_cart_state"); activity.ScheduledTime = new JavaScriptExpression("plus(LastUpdateTimestamp, durationFromSeconds(10)).ToDateTimeUtc()"); activity.Message = new JavaScriptExpression("return { correlationId: correlationId(), cartId: correlationId() }"); }).WithName("ScheduleExpire") @@ -62,7 +70,7 @@ namespace Sample21.Workflows .Then(activity => { activity.Message = new JavaScriptExpression("return { cartId: correlationId() };"); - activity.EndpointAddress = new Uri("rabbitmq://localhost/shopping_cart_service"); + activity.EndpointAddress = new Uri($"{options.Host}/shopping_cart_service"); activity.MessageType = typeof(CartRemovedEvent); }) .Then("Join"); diff --git a/src/samples/Sample21/appsettings.json b/src/samples/Sample21/appsettings.json index a39cd9b13..5c2cf0a17 100644 --- a/src/samples/Sample21/appsettings.json +++ b/src/samples/Sample21/appsettings.json @@ -8,5 +8,15 @@ "Smtp": { "Host": "localhost", "Port": "2525" + }, + "MassTransit": { + "RabbitMq": { + "Host": "rabbitmq://localhost:5672", + "Username": "guest", + "Password": "guest", + "MessageSchedule": { + "SchedulerAddress": "rabbitmq://localhost:5672/sample_quartz_scheduler" + } + } } } diff --git a/src/samples/Sample26/Messages/CreateOrder.cs b/src/samples/Sample26/Messages/CreateOrder.cs new file mode 100644 index 000000000..621988cd0 --- /dev/null +++ b/src/samples/Sample26/Messages/CreateOrder.cs @@ -0,0 +1,13 @@ +using System; +using Sample26.Models; + +namespace Sample26.Messages +{ + public class CreateOrder + { + // Convention based property for message correlation + // https://masstransit-project.com/MassTransit/usage/messages.html#message-correlation + public Guid CorrelationId { get; set; } + public Order Order { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample26/Messages/OrderShipped.cs b/src/samples/Sample26/Messages/OrderShipped.cs new file mode 100644 index 000000000..d4f94d94d --- /dev/null +++ b/src/samples/Sample26/Messages/OrderShipped.cs @@ -0,0 +1,13 @@ +using System; +using Sample26.Models; + +namespace Sample26.Messages +{ + public class OrderShipped + { + // Convention based property for message correlation + // https://masstransit-project.com/MassTransit/usage/messages.html#message-correlation + public Guid CorrelationId { get; set; } + public Order Order { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample26/Models/Customer.cs b/src/samples/Sample26/Models/Customer.cs new file mode 100644 index 000000000..67216e6d4 --- /dev/null +++ b/src/samples/Sample26/Models/Customer.cs @@ -0,0 +1,8 @@ +namespace Sample26.Models +{ + public class Customer + { + public string Name { get; set; } + public string Email { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample26/Models/Order.cs b/src/samples/Sample26/Models/Order.cs new file mode 100644 index 000000000..15535e709 --- /dev/null +++ b/src/samples/Sample26/Models/Order.cs @@ -0,0 +1,10 @@ +namespace Sample26.Models +{ + public class Order + { + public string Id { get; set; } + public Customer Customer { get; set; } + public string Product { get; set; } + public decimal Amount { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample26/Program.cs b/src/samples/Sample26/Program.cs new file mode 100644 index 000000000..86c542a45 --- /dev/null +++ b/src/samples/Sample26/Program.cs @@ -0,0 +1,14 @@ +using Microsoft.AspNetCore.Hosting; +using Microsoft.Extensions.Hosting; + +namespace Sample26 +{ + public static class Program + { + public static void Main(string[] args) => CreateHostBuilder(args).Build().Run(); + + public static IHostBuilder CreateHostBuilder(string[] args) => + Host.CreateDefaultBuilder(args) + .ConfigureWebHostDefaults(webBuilder => { webBuilder.UseStartup(); }); + } +} \ No newline at end of file diff --git a/src/samples/Sample26/Properties/launchSettings.json b/src/samples/Sample26/Properties/launchSettings.json new file mode 100644 index 000000000..d8aa116de --- /dev/null +++ b/src/samples/Sample26/Properties/launchSettings.json @@ -0,0 +1,30 @@ +{ + "$schema": "http://json.schemastore.org/launchsettings.json", + "iisSettings": { + "windowsAuthentication": false, + "anonymousAuthentication": true, + "iisExpress": { + "applicationUrl": "http://localhost:59862", + "sslPort": 44347 + } + }, + "profiles": { + "IIS Express": { + "commandName": "IISExpress", + "launchBrowser": true, + "launchUrl": "api/values", + "environmentVariables": { + "ASPNETCORE_ENVIRONMENT": "Development" + } + }, + "Sample26": { + "commandName": "Project", + "launchBrowser": true, + "launchUrl": "api/values", + "applicationUrl": "http://localhost:59862", + "environmentVariables": { + "ASPNETCORE_ENVIRONMENT": "Development" + } + } + } +} \ No newline at end of file diff --git a/src/samples/Sample26/Sample26.csproj b/src/samples/Sample26/Sample26.csproj new file mode 100644 index 000000000..8ed3f1bde --- /dev/null +++ b/src/samples/Sample26/Sample26.csproj @@ -0,0 +1,13 @@ + + + + netcoreapp3.1 + + + + + + + + + diff --git a/src/samples/Sample26/Sample26.http b/src/samples/Sample26/Sample26.http new file mode 100644 index 000000000..85c395505 --- /dev/null +++ b/src/samples/Sample26/Sample26.http @@ -0,0 +1,18 @@ +// Use VS Code Rest Client +// https://marketplace.visualstudio.com/items?itemName=humao.rest-client + +// Click Send Requst below: +POST http://localhost:59862/orders +Content-Type: application/json + +{ + "id": "order-1", + "customer": { + "name": "Jimmy Drop Tables", + "email": "drop-tables@where.com" + }, + "product": "MassTransit", + "amount": 99.95 +} + +### diff --git a/src/samples/Sample26/Startup.cs b/src/samples/Sample26/Startup.cs new file mode 100644 index 000000000..4f0cfe1e0 --- /dev/null +++ b/src/samples/Sample26/Startup.cs @@ -0,0 +1,56 @@ +using Elsa.Activities.Email.Extensions; +using Elsa.Activities.Http.Extensions; +using Elsa.Activities.MassTransit; +using Elsa.Activities.MassTransit.Extensions; +using Elsa.Activities.MassTransit.Options; +using Elsa.Activities.Timers.Extensions; +using Microsoft.AspNetCore.Builder; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using NodaTime; +using Sample26.Messages; +using Sample26.Workflows; + +namespace Sample26 +{ + public class Startup + { + public Startup(IConfiguration configuration) + { + Configuration = configuration; + } + + private IConfiguration Configuration { get; } + + public void ConfigureServices(IServiceCollection services) + { + var massTransitBuilder = new DefaultAzureServiceBusMassTransitBuilder + { + MessageTypes = new[] + { + typeof(CreateOrder), + typeof(OrderShipped) + }, + Options = o => o.Bind(Configuration.GetSection("MassTransit:AzureServiceBus")) + }; + + services + .AddElsa() + .AddHttpActivities() + .AddTimerActivities(options => options.Configure(x => x.SweepInterval = Duration.FromSeconds(10))) + .AddEmailActivities(options => options.Bind(Configuration.GetSection("Smtp"))) + .AddMassTransitActivities(massTransitBuilder) + .AddWorkflow() + .AddWorkflow(); + } + + public void Configure(IApplicationBuilder app, IHostEnvironment env) + { + if (env.IsDevelopment()) + app.UseDeveloperExceptionPage(); + + app.UseHttpActivities(); + } + } +} \ No newline at end of file diff --git a/src/samples/Sample26/Workflows/CreateOrderWorkflow.cs b/src/samples/Sample26/Workflows/CreateOrderWorkflow.cs new file mode 100644 index 000000000..5757695e4 --- /dev/null +++ b/src/samples/Sample26/Workflows/CreateOrderWorkflow.cs @@ -0,0 +1,80 @@ +using System; +using System.Net; +using System.Net.Http; +using Elsa.Activities; +using Elsa.Activities.Email.Activities; +using Elsa.Activities.Http.Activities; +using Elsa.Activities.MassTransit.Activities; +using Elsa.Activities.ControlFlow.Activities; +using Elsa.Activities.Workflows.Activities; +using Elsa.Expressions; +using Elsa.Scripting.JavaScript; +using Elsa.Services; +using Elsa.Services.Models; +using Sample26.Messages; + +namespace Sample26.Workflows +{ + public class CreateOrderWorkflow : IWorkflow + { + public void Build(IWorkflowBuilder builder) + { + builder + .StartWith( + activity => + { + activity.Method = HttpMethod.Post.Method; + activity.Path = new Uri("/orders", UriKind.RelativeOrAbsolute); + activity.ReadContent = true; + } + ) + .Then( + activity => + { + activity.VariableName = "order"; + activity.ValueExpression = new JavaScriptExpression("lastResult().Body"); + } + ) + // Need to ensure that the correlation ID is the same string format that is used by the WorkflowConsumer + // MassTransit will always use Guid values for correlation ID's, so need to ensure the same string format is used. + .Then(activity => activity.ValueExpression = new JavaScriptExpression("newGuid()")) + .Then(activity => + { + activity.Message = new JavaScriptExpression("return { correlationId: correlationId(), order: order};"); + activity.MessageType = typeof(CreateOrder); + } + ) + .Then( + activity => activity.Branches = new[] { "Write-Response", "Await-Shipment" }, + fork => + { + fork + .When("Write-Response") + .Then( + activity => + { + activity.Content = new LiteralExpression("

Order Received

Your order has been received. Waiting for shipment.

"); + activity.ContentType = "text/html"; + activity.StatusCode = HttpStatusCode.Accepted; + } + ); + + fork + .When("Await-Shipment") + .Then(activity => activity.MessageType = typeof(OrderShipped)) + .Then( + activity => + { + activity.From = new LiteralExpression("shipment@acme.com"); + activity.To = new JavaScriptExpression("order.customer.email"); + activity.Subject = new JavaScriptExpression("`Your order with ID #${order.id} has been shipped!`"); + activity.Body = new JavaScriptExpression( + "`Dear ${order.customer.name}, your order has shipped!`" + ); + } + ); + } + ); + } + } +} \ No newline at end of file diff --git a/src/samples/Sample26/Workflows/HandleOrderWorkflow.cs b/src/samples/Sample26/Workflows/HandleOrderWorkflow.cs new file mode 100644 index 000000000..9d1b14c20 --- /dev/null +++ b/src/samples/Sample26/Workflows/HandleOrderWorkflow.cs @@ -0,0 +1,35 @@ +using System; +using Elsa.Activities; +using Elsa.Activities.MassTransit.Activities; +using Elsa.Activities.Timers.Activities; +using Elsa.Expressions; +using Elsa.Scripting.JavaScript; +using Elsa.Services; +using Elsa.Services.Models; +using Sample26.Messages; + +namespace Sample26.Workflows +{ + public class HandleOrderWorkflow : IWorkflow + { + public void Build(IWorkflowBuilder builder) + { + builder + .StartWith(activity => activity.MessageType = typeof(CreateOrder)) + .Then( + activity => + { + activity.VariableName = "order"; + activity.ValueExpression = new JavaScriptExpression("lastResult().Order"); + } + ) + .Then(activity => activity.TimeoutExpression = new LiteralExpression("00:00:05")) + .Then(activity => + { + activity.Message = new JavaScriptExpression("return { correlationId: correlationId(), order: order}"); + activity.MessageType = typeof(OrderShipped); + } + ); + } + } +} \ No newline at end of file diff --git a/src/samples/Sample26/app.config b/src/samples/Sample26/app.config new file mode 100644 index 000000000..a6ad82831 --- /dev/null +++ b/src/samples/Sample26/app.config @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/src/samples/Sample26/appsettings.Development.json b/src/samples/Sample26/appsettings.Development.json new file mode 100644 index 000000000..e203e9407 --- /dev/null +++ b/src/samples/Sample26/appsettings.Development.json @@ -0,0 +1,9 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Debug", + "System": "Information", + "Microsoft": "Information" + } + } +} diff --git a/src/samples/Sample26/appsettings.json b/src/samples/Sample26/appsettings.json new file mode 100644 index 000000000..d69f25dff --- /dev/null +++ b/src/samples/Sample26/appsettings.json @@ -0,0 +1,18 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Warning" + } + }, + "AllowedHosts": "*", + "Smtp": { + "Host": "localhost", + "Port": "2525" + }, + "MassTransit": { + "AzureServiceBus": { + "Host": "sb://AZURE_SERVICE_BUS_HOST_NAME", + "ConnectionString": "AZURE_SERVICE_BUS_CONNECTION_STRING" + } + } +} diff --git a/src/samples/Sample27/AzureServiceBusSchedulerMassTransitBuilder.cs b/src/samples/Sample27/AzureServiceBusSchedulerMassTransitBuilder.cs new file mode 100644 index 000000000..4465cd032 --- /dev/null +++ b/src/samples/Sample27/AzureServiceBusSchedulerMassTransitBuilder.cs @@ -0,0 +1,87 @@ +using System; +using Elsa.Activities.MassTransit.Extensions; +using GreenPipes; +using MassTransit; +using MassTransit.ExtensionsDependencyInjectionIntegration; +using MassTransit.QuartzIntegration; +using Quartz; +using Sample27.Consumers; +using Sample27.Messages; +using Elsa.Activities.MassTransit; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using MassTransit.Azure.ServiceBus.Core; + +namespace Sample27 +{ + internal class AzureServiceBusSchedulerMassTransitBuilder : MassTransitBuilderBase + { + private readonly IScheduler scheduler; + + public AzureServiceBusSchedulerMassTransitBuilder(IScheduler scheduler) + { + this.scheduler = scheduler; + } + + protected override IBusControl CreateBus(IServiceProvider serviceProvider) + { + var bus = Bus.Factory.CreateUsingAzureServiceBus(cfg => + { + var options = serviceProvider.GetRequiredService>().Value; + cfg.Host(options.ConnectionString); + + cfg.UseMessageScheduler(options.MessageSchedule.SchedulerAddress); + + cfg.ReceiveEndpoint("shopping_cart_service", ep => + { + ep.ConfigureConsumer(serviceProvider); + }); + + cfg.ReceiveEndpoint("shopping_cart_state", ep => + { + ep.PrefetchCount = 16; + + ep.UseMessageRetry(r => r.Interval(2, 100)); + + // Consume all workflow messages from the same queue. + ep.ConfigureWorkflowConsumer(serviceProvider); + ep.ConfigureWorkflowConsumer(serviceProvider); + ep.ConfigureWorkflowConsumer(serviceProvider); + ep.ConfigureWorkflowConsumer(serviceProvider); + }); + + // Should use external process scheduler service + // https://github.com/MassTransit/MassTransit/tree/develop/src/Samples/MassTransit.QuartzService + cfg.ReceiveEndpoint("sample_quartz_scheduler", e => + { + // For MT4.0, prefetch must be set for Quartz prior to anything else + e.PrefetchCount = 1; + cfg.UseMessageScheduler(e.InputAddress); + + e.ConfigureConsumer(serviceProvider); + e.ConfigureConsumer(serviceProvider); + }); + }); + + scheduler.JobFactory = new MassTransitJobFactory(bus); + + return bus; + } + + protected override void ConfigureMassTransit(IServiceCollectionConfigurator configurator) + { + // Configure scheduler service consumers. + configurator.AddConsumer(); + configurator.AddConsumer(); + + // configure workflow consumers + configurator.AddWorkflowConsumer(); + configurator.AddWorkflowConsumer(); + configurator.AddWorkflowConsumer(); + configurator.AddWorkflowConsumer(); + + // host fake service consumers + configurator.AddConsumer(); + } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/AzureServiceBusSchedulerOptions.cs b/src/samples/Sample27/AzureServiceBusSchedulerOptions.cs new file mode 100644 index 000000000..c2979f056 --- /dev/null +++ b/src/samples/Sample27/AzureServiceBusSchedulerOptions.cs @@ -0,0 +1,9 @@ +using Elsa.Activities.MassTransit.Options; + +namespace Sample27 +{ + public class AzureServiceBusSchedulerOptions : AzureServiceBusOptions + { + public MessageScheduleOptions MessageSchedule { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Consumers/CartRemovedConsumer.cs b/src/samples/Sample27/Consumers/CartRemovedConsumer.cs new file mode 100644 index 000000000..83a355dde --- /dev/null +++ b/src/samples/Sample27/Consumers/CartRemovedConsumer.cs @@ -0,0 +1,24 @@ +using System.Threading.Tasks; +using MassTransit; +using Sample27.Controllers; +using Sample27.Messages; + +namespace Sample27.Consumers +{ + public class CartRemovedConsumer : IConsumer + { + private readonly ICarts carts; + + public CartRemovedConsumer(ICarts carts) + { + this.carts = carts; + } + + public Task Consume(ConsumeContext context) + { + carts.Remove(context.Message.CartId); + + return Task.CompletedTask; + } + } +} diff --git a/src/samples/Sample27/Controllers/CartController.cs b/src/samples/Sample27/Controllers/CartController.cs new file mode 100644 index 000000000..90c6a47e7 --- /dev/null +++ b/src/samples/Sample27/Controllers/CartController.cs @@ -0,0 +1,51 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Mvc; +using Sample27.Models; + +namespace Sample27.Controllers +{ + [Route("api/[controller]")] + [ApiController] + public class CartController : ControllerBase + { + private const string Username = "sampleUser"; + + private readonly ICarts carts; + + public CartController(ICarts carts) + { + this.carts = carts; + } + + // GET: api/Cart + [HttpGet] + public async Task Get() + { + var cart = await carts.Cart(Username); + + return cart ?? new Cart + { + Id = Guid.Empty, + Username = Username, + Items = new List() + }; + } + + // POST: api/Cart + [HttpPost] + public async Task Post([FromBody] CartItem item) + { + await carts.AddItem(Username, item); + } + + // POST: api/cart/submit + [HttpPost] + [Route("submit")] + public async Task SubmitOrder() + { + await carts.Submit(Username); + } + } +} diff --git a/src/samples/Sample27/Controllers/Carts.cs b/src/samples/Sample27/Controllers/Carts.cs new file mode 100644 index 000000000..2a1b8c6eb --- /dev/null +++ b/src/samples/Sample27/Controllers/Carts.cs @@ -0,0 +1,99 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using MassTransit; +using Sample27.Messages; +using Sample27.Models; + +namespace Sample27.Controllers +{ + public class Carts : ICarts + { + private static readonly ConcurrentDictionary ActiveCarts = new ConcurrentDictionary(); + + private readonly ISendEndpointProvider sender; + + public Carts(ISendEndpointProvider sender) + { + this.sender = sender; + } + + public Task Cart(string username) + { + ActiveCarts.TryGetValue(username, out var cart); + + return Task.FromResult(cart); + } + + public async Task AddItem(string username, CartItem item) + { + var cart = await Cart(username); + + if (cart == null) + { + cart = new Cart + { + Id = Guid.NewGuid(), + Username = username, + Items = new List() + }; + + if (ActiveCarts.TryAdd(username, cart)) + { + await sender.Send(new CartCreated + { + CartId = cart.Id, + UserName = cart.Username, + Timestamp = DateTime.UtcNow + }); + } + else + { + ActiveCarts.TryGetValue(username, out cart); + } + } + + var exitingItem = cart.Items.FirstOrDefault(x => x.ProductSku == item.ProductSku); + if (exitingItem != null) + { + exitingItem.Quantity += item.Quantity; + } + else + { + cart.Items.Add(item); + } + + await sender.Send(new CartItemAdded + { + CartId = cart.Id, + UserName = cart.Username, + Timestamp = DateTime.UtcNow + }); + } + + public async Task Submit(string username) + { + if (ActiveCarts.TryRemove(username, out var cart)) + { + await sender.Send(new OrderSubmitted + { + OrderId = Guid.NewGuid(), + CartId = cart.Id, + UserName = username, + Timestamp = DateTime.UtcNow + }); + } + } + + public void Remove(Guid cartId) + { + var cart = ActiveCarts.Values.FirstOrDefault(x => x.Id == cartId); + if (cart != null) + { + ActiveCarts.TryRemove(cart.Username, out _); + } + } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Controllers/ICarts.cs b/src/samples/Sample27/Controllers/ICarts.cs new file mode 100644 index 000000000..611cb2456 --- /dev/null +++ b/src/samples/Sample27/Controllers/ICarts.cs @@ -0,0 +1,14 @@ +using System; +using System.Threading.Tasks; +using Sample27.Models; + +namespace Sample27.Controllers +{ + public interface ICarts + { + Task Cart(string username); + Task AddItem(string username, CartItem item); + Task Submit(string username); + void Remove(Guid cartId); + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Messages/CartCreated.cs b/src/samples/Sample27/Messages/CartCreated.cs new file mode 100644 index 000000000..30f59c6f4 --- /dev/null +++ b/src/samples/Sample27/Messages/CartCreated.cs @@ -0,0 +1,15 @@ +using System; +using MassTransit; + +namespace Sample27.Messages +{ + public class CartCreated : CorrelatedBy + { + public Guid CorrelationId => CartId; + + public Guid CartId { get; set; } + + public DateTime Timestamp { get; set; } + public string UserName { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Messages/CartExpiredEvent.cs b/src/samples/Sample27/Messages/CartExpiredEvent.cs new file mode 100644 index 000000000..bd8db1de9 --- /dev/null +++ b/src/samples/Sample27/Messages/CartExpiredEvent.cs @@ -0,0 +1,11 @@ +using System; +using MassTransit; + +namespace Sample27.Messages +{ + public class CartExpiredEvent : CorrelatedBy + { + public Guid CorrelationId { get; set; } + public Guid CartId { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Messages/CartItemAdded.cs b/src/samples/Sample27/Messages/CartItemAdded.cs new file mode 100644 index 000000000..ea4ffc36b --- /dev/null +++ b/src/samples/Sample27/Messages/CartItemAdded.cs @@ -0,0 +1,16 @@ +using System; +using MassTransit; + +namespace Sample27.Messages +{ + public class CartItemAdded : CorrelatedBy + { + public Guid CorrelationId => CartId; + + public Guid CartId { get; set; } + + public string UserName { get; set; } + + public DateTime Timestamp { get; set; } + } +} diff --git a/src/samples/Sample27/Messages/CartRemovedEvent.cs b/src/samples/Sample27/Messages/CartRemovedEvent.cs new file mode 100644 index 000000000..37c6673bb --- /dev/null +++ b/src/samples/Sample27/Messages/CartRemovedEvent.cs @@ -0,0 +1,12 @@ +using System; +using MassTransit; + +namespace Sample27.Messages +{ + public class CartRemovedEvent : CorrelatedBy + { + public Guid CorrelationId => CartId; + + public Guid CartId { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Messages/OrderSubmitted.cs b/src/samples/Sample27/Messages/OrderSubmitted.cs new file mode 100644 index 000000000..11745e476 --- /dev/null +++ b/src/samples/Sample27/Messages/OrderSubmitted.cs @@ -0,0 +1,18 @@ +using System; +using MassTransit; + +namespace Sample27.Messages +{ + public class OrderSubmitted : CorrelatedBy + { + public Guid CorrelationId => CartId; + + public Guid OrderId { get; set; } + + public DateTime Timestamp { get; set; } + + public Guid CartId { get; set; } + + public string UserName { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Models/Cart.cs b/src/samples/Sample27/Models/Cart.cs new file mode 100644 index 000000000..132d7a8a0 --- /dev/null +++ b/src/samples/Sample27/Models/Cart.cs @@ -0,0 +1,14 @@ +using System; +using System.Collections.Generic; + +namespace Sample27.Models +{ + public class Cart + { + public Guid Id { get; set; } + + public string Username { get; set; } + + public IList Items { get; set; } + } +} diff --git a/src/samples/Sample27/Models/CartItem.cs b/src/samples/Sample27/Models/CartItem.cs new file mode 100644 index 000000000..4ffff7e23 --- /dev/null +++ b/src/samples/Sample27/Models/CartItem.cs @@ -0,0 +1,9 @@ +namespace Sample27.Models +{ + public class CartItem + { + public int Quantity { get; set; } + + public string ProductSku { get; set; } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Program.cs b/src/samples/Sample27/Program.cs new file mode 100644 index 000000000..8ff41194e --- /dev/null +++ b/src/samples/Sample27/Program.cs @@ -0,0 +1,14 @@ +using Microsoft.AspNetCore.Hosting; +using Microsoft.Extensions.Hosting; + +namespace Sample27 +{ + public static class Program + { + public static void Main(string[] args) => CreateHostBuilder(args).Build().Run(); + + public static IHostBuilder CreateHostBuilder(string[] args) => + Host.CreateDefaultBuilder(args) + .ConfigureWebHostDefaults(webBuilder => { webBuilder.UseStartup(); }); + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Properties/launchSettings.json b/src/samples/Sample27/Properties/launchSettings.json new file mode 100644 index 000000000..ac86fe255 --- /dev/null +++ b/src/samples/Sample27/Properties/launchSettings.json @@ -0,0 +1,30 @@ +{ + "$schema": "http://json.schemastore.org/launchsettings.json", + "iisSettings": { + "windowsAuthentication": false, + "anonymousAuthentication": true, + "iisExpress": { + "applicationUrl": "http://localhost:59862", + "sslPort": 44347 + } + }, + "profiles": { + "IIS Express": { + "commandName": "IISExpress", + "launchBrowser": true, + "launchUrl": "api/cart", + "environmentVariables": { + "ASPNETCORE_ENVIRONMENT": "Development" + } + }, + "Sample27": { + "commandName": "Project", + "launchBrowser": true, + "launchUrl": "api/cart", + "applicationUrl": "http://localhost:59862", + "environmentVariables": { + "ASPNETCORE_ENVIRONMENT": "Development" + } + } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Sample27.csproj b/src/samples/Sample27/Sample27.csproj new file mode 100644 index 000000000..59dacc570 --- /dev/null +++ b/src/samples/Sample27/Sample27.csproj @@ -0,0 +1,21 @@ + + + + netcoreapp3.1 + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + + diff --git a/src/samples/Sample27/Sample27.http b/src/samples/Sample27/Sample27.http new file mode 100644 index 000000000..b2656acff --- /dev/null +++ b/src/samples/Sample27/Sample27.http @@ -0,0 +1,21 @@ +// Use VS Code Rest Client +// https://marketplace.visualstudio.com/items?itemName=humao.rest-client + + +GET http://localhost:59862/api/cart + +#### + +POST http://localhost:59862/api/cart +Content-Type: application/json + +{ + "productSku": "010-01787-00", + "quantity": 1 +} + +#### + +POST http://localhost:59862/api/cart/submit + +### \ No newline at end of file diff --git a/src/samples/Sample27/Services/QuartzHostedService.cs b/src/samples/Sample27/Services/QuartzHostedService.cs new file mode 100644 index 000000000..a10b16ade --- /dev/null +++ b/src/samples/Sample27/Services/QuartzHostedService.cs @@ -0,0 +1,27 @@ +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Hosting; +using Quartz; + +namespace Sample27.Services +{ + public class QuartzHostedService : IHostedService + { + private readonly IScheduler scheduler; + + public QuartzHostedService(IScheduler scheduler) + { + this.scheduler = scheduler; + } + + public async Task StartAsync(CancellationToken cancellationToken) + { + await scheduler.Start(cancellationToken); + } + + public async Task StopAsync(CancellationToken cancellationToken) + { + await scheduler.Shutdown(true, cancellationToken); + } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Startup.cs b/src/samples/Sample27/Startup.cs new file mode 100644 index 000000000..8425a1d07 --- /dev/null +++ b/src/samples/Sample27/Startup.cs @@ -0,0 +1,110 @@ +using System; +using System.Collections.Specialized; +using Elsa.Activities.Email.Extensions; +using Elsa.Activities.Http.Extensions; +using Elsa.Activities.MassTransit.Extensions; +using Elsa.Activities.Timers.Extensions; +using Microsoft.AspNetCore.Builder; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using NodaTime; +using Quartz; +using Quartz.Impl; +using Quartz.Logging; +using Sample27.Controllers; +using Sample27.Services; +using Sample27.Workflows; + +namespace Sample27 +{ + public class Startup + { + public Startup(IConfiguration configuration) + { + Configuration = configuration; + } + + private IConfiguration Configuration { get; } + + public void ConfigureServices(IServiceCollection services) + { + services + .AddControllers(); + + var scheduler = CreateScheduler(); + var massTransitBuilder = new AzureServiceBusSchedulerMassTransitBuilder(scheduler) + { + Options = o => o.Bind(Configuration.GetSection("MassTransit:AzureServiceBus")) + }; + + services + .AddElsa() + .AddHttpActivities() + .AddTimerActivities(options => options.Configure(x => x.SweepInterval = Duration.FromSeconds(10))) + .AddEmailActivities(options => options.Bind(Configuration.GetSection("Smtp"))) + .AddMassTransitSchedulingActivities(massTransitBuilder, options => options.Bind(Configuration.GetSection("MassTransit:AzureServiceBus:MessageSchedule"))) + .AddWorkflow() + .AddScoped() + .AddSingleton(scheduler) + .AddSingleton(); // Add a hosted service to stat and stop the quartz scheduler + + IScheduler CreateScheduler() + { + LogProvider.SetCurrentLogProvider(new QuartzConsoleLogProvider()); + + ISchedulerFactory schedulerFactory = new StdSchedulerFactory(new NameValueCollection() + { + {"quartz.scheduler.instanceName", "Sample-QuartzScheduler"}, + {"quartz.scheduler.instanceId", "AUTO"}, + {"quartz.threadPool.type", "Quartz.Simpl.SimpleThreadPool, Quartz"}, + {"quartz.threadPool.threadCount", "4"}, + {"quartz.jobStore.misfireThreshold", "60000"}, + {"quartz.jobStore.type", "Quartz.Simpl.RAMJobStore, Quartz"}, + }); + + var scheduler = schedulerFactory.GetScheduler().GetAwaiter().GetResult(); + + return scheduler; + } + } + + public void Configure(IApplicationBuilder app, IHostEnvironment env) + { + if (env.IsDevelopment()) + app.UseDeveloperExceptionPage(); + + app + .UseStaticFiles() + .UseHttpActivities() + .UseRouting() + .UseEndpoints(endpoints => endpoints.MapControllers()) + .UseWelcomePage(); + } + + private class QuartzConsoleLogProvider : ILogProvider + { + public Logger GetLogger(string name) + { + return (level, func, exception, parameters) => + { + if (level >= LogLevel.Debug && func != null) + { + Console.WriteLine("[" + DateTime.Now.ToLongTimeString() + "] [" + level + "] " + func(), parameters); + } + return true; + }; + } + + public IDisposable OpenNestedContext(string message) + { + throw new NotImplementedException(); + } + + public IDisposable OpenMappedContext(string key, string value) + { + throw new NotImplementedException(); + } + } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/Workflows/CartTrackingWorkflow.cs b/src/samples/Sample27/Workflows/CartTrackingWorkflow.cs new file mode 100644 index 000000000..9c3fca588 --- /dev/null +++ b/src/samples/Sample27/Workflows/CartTrackingWorkflow.cs @@ -0,0 +1,92 @@ +using System; +using System.Resources; +using Elsa.Activities; +using Elsa.Activities.ControlFlow.Activities; +using Elsa.Activities.MassTransit.Activities; +using Elsa.Scripting.JavaScript; +using Elsa.Services; +using Elsa.Services.Models; +using Microsoft.Extensions.Options; +using NodaTime; +using Sample27.Messages; + +namespace Sample27.Workflows +{ + public class CartTrackingWorkflow : IWorkflow + { + private readonly AzureServiceBusSchedulerOptions options; + + public CartTrackingWorkflow(IOptions options) + { + this.options = options.Value; + } + + public void Build(IWorkflowBuilder builder) + { + builder.StartWith(activity => activity.MessageType = typeof(CartCreated)) + .Then(activity => + { + activity.VariableName = "LastUpdateTimestamp"; + activity.ValueExpression = new JavaScriptExpression("instantFromDateTimeUtc(lastResult().Timestamp)"); + }) + .Then(activity => + { + activity.VariableName = "IsExpired"; + activity.ValueExpression = new JavaScriptExpression("false"); + }) + .Fork( + action => action.Branches = new [] {"Item-Added", "Cart-Expired", "Order-Submitted"}, + fork => + { + fork.When("Item-Added") + .Then( + activity => + { + activity.MessageType = typeof(CartExpiredEvent); + activity.EndpointAddress = new Uri($"{options.Host}/shopping_cart_state"); + activity.ScheduledTime = new JavaScriptExpression("plus(LastUpdateTimestamp, durationFromSeconds(10)).ToDateTimeUtc()"); + activity.Message = new JavaScriptExpression("return { correlationId: correlationId(), cartId: correlationId() }"); + }).WithName("ScheduleExpire") + .Then(activity => + { + activity.VariableName = "ScheduleTokenId"; + activity.ValueExpression = new JavaScriptExpression("lastResult()"); + }) + .Then(activity => activity.MessageType = typeof(CartItemAdded)) + .Then(activity => + { + activity.VariableName = "LastUpdateTimestamp"; + activity.ValueExpression = new JavaScriptExpression("instantFromDateTimeUtc(lastResult().Timestamp)"); + }) + .Then(activity => activity.TokenId = new JavaScriptExpression("return ScheduleTokenId")) + .Then("ScheduleExpire"); + + fork.When("Cart-Expired") + .Then(activity => activity.MessageType = typeof(CartExpiredEvent)) + .Then(activity => + { + activity.VariableName = "IsExpired"; + activity.ValueExpression = new JavaScriptExpression("true"); + }) + .Then(activity => + { + activity.Message = new JavaScriptExpression("return { cartId: correlationId() };"); + activity.EndpointAddress = new Uri($"{options.Host}/shopping_cart_service"); + activity.MessageType = typeof(CartRemovedEvent); + }) + .Then("Join"); + + fork.When("OrderSubmitted") + .Then(activity => activity.MessageType = typeof(OrderSubmitted)) + .Then(activity => + { + activity.VariableName = "LastUpdateTimestamp"; + activity.ValueExpression = new JavaScriptExpression("lastResult().Timestamp"); + }) + .Then(activity => activity.TokenId = new JavaScriptExpression("return ScheduleTokenId")) + .Then("Join"); + }) + .Join(x => x.Mode = Join.JoinMode.WaitAny).WithName("Join"); + } + } +} \ No newline at end of file diff --git a/src/samples/Sample27/app.config b/src/samples/Sample27/app.config new file mode 100644 index 000000000..52f8d6561 --- /dev/null +++ b/src/samples/Sample27/app.config @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/src/samples/Sample27/appsettings.Development.json b/src/samples/Sample27/appsettings.Development.json new file mode 100644 index 000000000..e203e9407 --- /dev/null +++ b/src/samples/Sample27/appsettings.Development.json @@ -0,0 +1,9 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Debug", + "System": "Information", + "Microsoft": "Information" + } + } +} diff --git a/src/samples/Sample27/appsettings.json b/src/samples/Sample27/appsettings.json new file mode 100644 index 000000000..87608ef3a --- /dev/null +++ b/src/samples/Sample27/appsettings.json @@ -0,0 +1,21 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Debug" + } + }, + "AllowedHosts": "*", + "Smtp": { + "Host": "localhost", + "Port": "2525" + }, + "MassTransit": { + "AzureServiceBus": { + "Host": "sb://AZURE_SERVICE_BUS_HOST_NAME", + "ConnectionString": "AZURE_SERVICE_BUS_CONNECTION_STRING", + "MessageSchedule": { + "SchedulerAddress": "sb://AZURE_SERVICE_BUS_HOST_NAME/sample_quartz_scheduler" + } + } + } +}