SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Enhancements for using azure service bus with MassTransit activities (#457)

Browse source 
* Added AzureServiceBus support for MassTransit activities

Fixes #455
This commit is contained in:
cvijayak 2020-11-13 02:09:32 +05:30 committed by GitHub
parent 2934bfce72
commit a4d9abdd75
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
56 changed files with 1416 additions and 203 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
Samples.sln
View file

@ -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}

48
src/activities/Elsa.Activities.MassTransit/DefaultAzureServiceBusMassTransitBuilder.cs Normal file
View file

@ -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<AzureServiceBusOptions>
{
protected override IBusControl CreateBus(IServiceProvider serviceProvider)
{
return Bus.Factory.CreateUsingAzureServiceBus(bus =>
{
var options = serviceProvider.GetRequiredService<IOptions<AzureServiceBusOptions>>().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<Type> MessageTypes { get; set; }
}
}

57
src/activities/Elsa.Activities.MassTransit/DefaultRabbitMqMassTransitBuilder.cs Normal file
View file

@ -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<RabbitMqOptions>
{
protected override IBusControl CreateBus(IServiceProvider serviceProvider)
{
return Bus.Factory.CreateUsingRabbitMq(bus =>
{
var options = serviceProvider.GetRequiredService<IOptions<RabbitMqOptions>>().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<Type> MessageTypes { get; set; }
}
}

1
src/activities/Elsa.Activities.MassTransit/Elsa.Activities.MassTransit.csproj
View file

@ -29,6 +29,7 @@
<ItemGroup>
<PackageReference Include="MassTransit.AspNetCore" Version="6.0.1" />
<PackageReference Include="MassTransit.Azure.ServiceBus.Core" Version="6.0.1" />
<PackageReference Include="MassTransit.Extensions.DependencyInjection" Version="6.0.1" />
<PackageReference Include="MassTransit.Quartz" Version="6.0.1" />
<PackageReference Include="MassTransit.RabbitMQ" Version="6.0.1" />

112
src/activities/Elsa.Activities.MassTransit/Extensions/ServiceCollectionExtensions.cs
View file

@ -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<TOptions>(this IServiceCollection services,
IMassTransitBuilder<TOptions> massTransitBuilder,
Action<OptionsBuilder<MessageScheduleOptions>> options) where TOptions : class
{
if (services == null)
{
throw new ArgumentNullException(nameof(services));
}
return services
.AddActivity<PublishMassTransitMessage>()
.AddActivity<ReceiveMassTransitMessage>()
.AddActivity<SendMassTransitMessage>();
}
public static IServiceCollection AddMassTransitSchedulingActivities(this IServiceCollection services, Action<MessageScheduleOptions> configureOptions)
{
if (services == null)
{
throw new ArgumentNullException(nameof(services));
}
if (configureOptions == null)
{
throw new ArgumentNullException(nameof(configureOptions));
}
var optionsBuilder = services.AddOptions<MessageScheduleOptions>();
options?.Invoke(optionsBuilder);
services.AddMassTransitActivities()
.AddActivity<CancelScheduledMassTransitMessage>()
.AddActivity<ScheduleSendMassTransitMessage>();
services.Configure(configureOptions);
massTransitBuilder.Build(services);
return services;
}
public static IServiceCollection AddRabbitMqActivities(this IServiceCollection services, Action<OptionsBuilder<RabbitMqOptions>> options = null, params Type[] messageTypes)
public static IServiceCollection AddMassTransitActivities<TOptions>(this IServiceCollection services,
IMassTransitBuilder<TOptions> massTransitBuilder) where TOptions : class
{
var optionsBuilder = services.AddOptions<RabbitMqOptions>();
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<TMessage>(
@ -93,51 +53,17 @@ namespace Elsa.Activities.MassTransit.Extensions
return configurator.AddConsumer(configure);
}
private static IBusControl CreateUsingRabbitMq(IServiceProvider sp, IEnumerable<Type> messageTypes)
private static IServiceCollection AddMassTransitActivities(this IServiceCollection services)
{
return Bus.Factory.CreateUsingRabbitMq(
bus =>
{
var options = sp.GetRequiredService<IOptions<RabbitMqOptions>>().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<PublishMassTransitMessage>()
.AddActivity<ReceiveMassTransitMessage>()
.AddActivity<SendMassTransitMessage>();
}
}
}

21
src/activities/Elsa.Activities.MassTransit/Extensions/TypeExtensions.cs Normal file
View file

@ -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);
}
}
}

9
src/activities/Elsa.Activities.MassTransit/IMassTransitBuilder.cs Normal file
View file

@ -0,0 +1,9 @@
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Activities.MassTransit
{
public interface IMassTransitBuilder<TOptions> where TOptions : class
{
IServiceCollection Build(IServiceCollection services);
}
}

27
src/activities/Elsa.Activities.MassTransit/MassTransitBuilderBase.cs Normal file
View file

@ -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<TOptions> : IMassTransitBuilder<TOptions> where TOptions : class
{
protected MassTransitBuilderBase() { }
public IServiceCollection Build(IServiceCollection services)
{
var optionsBuilder = services.AddOptions<TOptions>();
Options?.Invoke(optionsBuilder);
return services.AddMassTransit(CreateBus, ConfigureMassTransit);
}
protected abstract void ConfigureMassTransit(IServiceCollectionConfigurator configurator);
protected abstract IBusControl CreateBus(IServiceProvider serviceProvider);
public Action<OptionsBuilder<TOptions>> Options { get; set; }
}
}

10
src/activities/Elsa.Activities.MassTransit/Options/AzureServiceBusOptions.cs Normal file
View file

@ -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; }
}
}

6
src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs
View file

@ -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<string, (string Name, Uri Uri)> collectionUris;
public DocumentDbStorage(IOptions<DocumentDbStorageOptions> 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);

8
src/samples/README.md
View file

@ -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).
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.

4
src/samples/Sample08/Sample08.csproj
View file

@ -4,10 +4,6 @@
<TargetFramework>netcoreapp3.1</TargetFramework>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="MassTransit.AspNetCore" Version="7.0.6" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\activities\Elsa.Activities.MassTransit\Elsa.Activities.MassTransit.csproj" />
<ProjectReference Include="..\..\core\Elsa\Elsa.csproj" />

19
src/samples/Sample08/Startup.cs
View file

@ -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<CreateOrderWorkflow>()
.AddWorkflow<HandleOrderWorkflow>();
}

97
src/samples/Sample21/RabbitMqSchedulerMassTransitBuilder.cs Normal file
View file

@ -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<RabbitMqSchedulerOptions>
{
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<IOptions<RabbitMqSchedulerOptions>>().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<CartRemovedConsumer>(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<CartCreated>(serviceProvider);
ep.ConfigureWorkflowConsumer<CartItemAdded>(serviceProvider);
ep.ConfigureWorkflowConsumer<OrderSubmitted>(serviceProvider);
ep.ConfigureWorkflowConsumer<CartExpiredEvent>(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<ScheduleMessageConsumer>(serviceProvider);
e.ConfigureConsumer<CancelScheduledMessageConsumer>(serviceProvider);
});
});
scheduler.JobFactory = new MassTransitJobFactory(bus);
return bus;
}
protected override void ConfigureMassTransit(IServiceCollectionConfigurator configurator)
{
// Configure scheduler service consumers.
configurator.AddConsumer<ScheduleMessageConsumer>();
configurator.AddConsumer<CancelScheduledMessageConsumer>();
// configure workflow consumers
configurator.AddWorkflowConsumer<CartCreated>();
configurator.AddWorkflowConsumer<CartItemAdded>();
configurator.AddWorkflowConsumer<OrderSubmitted>();
configurator.AddWorkflowConsumer<CartExpiredEvent>();
// host fake service consumers
configurator.AddConsumer<CartRemovedConsumer>();
}
}
}

9
src/samples/Sample21/RabbitMqSchedulerOptions.cs Normal file
View file

@ -0,0 +1,9 @@
using Elsa.Activities.MassTransit.Options;
namespace Sample21
{
public class RabbitMqSchedulerOptions : RabbitMqOptions
{
public MessageScheduleOptions MessageSchedule { get; set; }
}
}

6
src/samples/Sample21/Sample21.csproj
View file

@ -1,14 +1,10 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<TargetFramework>netcoreapp3.1</TargetFramework>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="MassTransit.AspNetCore" Version="6.0.1" />
<PackageReference Include="MassTransit.Extensions.DependencyInjection" Version="6.0.1" />
<PackageReference Include="MassTransit.Quartz" Version="6.0.1" />
<PackageReference Include="MassTransit.RabbitMQ" Version="6.0.1" />
<PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="3.1.9" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="3.1.9">
<PrivateAssets>all</PrivateAssets>

92
src/samples/Sample21/Startup.cs
View file

@ -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<CartTrackingWorkflow>()
.AddScoped<ICarts, Carts>()
.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<IHostedService, QuartzHostedService>();
void ConfigureMassTransit(IServiceCollectionConfigurator configurator)
{
// Configure scheduler service consumers.
configurator.AddConsumer<ScheduleMessageConsumer>();
configurator.AddConsumer<CancelScheduledMessageConsumer>();
// configure workflow consumers
configurator.AddWorkflowConsumer<CartCreated>();
configurator.AddWorkflowConsumer<CartItemAdded>();
configurator.AddWorkflowConsumer<OrderSubmitted>();
configurator.AddWorkflowConsumer<CartExpiredEvent>();
// host fake service consumers
configurator.AddConsumer<CartRemovedConsumer>();
}
// 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<CartRemovedConsumer>(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<CartCreated>(serviceProvider);
ep.ConfigureWorkflowConsumer<CartItemAdded>(serviceProvider);
ep.ConfigureWorkflowConsumer<OrderSubmitted>(serviceProvider);
ep.ConfigureWorkflowConsumer<CartExpiredEvent>(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<ScheduleMessageConsumer>(serviceProvider);
e.ConfigureConsumer<CancelScheduledMessageConsumer>(serviceProvider);
});
});
scheduler.JobFactory = new MassTransitJobFactory(bus);
return bus;
}
.AddSingleton<IHostedService, QuartzHostedService>(); // 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);
}

12
src/samples/Sample21/Workflows/CartTrackingWorkflow.cs
View file

@ -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<RabbitMqSchedulerOptions> options)
{
this.options = options.Value;
}
public void Build(IWorkflowBuilder builder)
{
builder.StartWith<ReceiveMassTransitMessage>(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<DateTime>("plus(LastUpdateTimestamp, durationFromSeconds(10)).ToDateTimeUtc()");
activity.Message = new JavaScriptExpression<CartExpiredEvent>("return { correlationId: correlationId(), cartId: correlationId() }");
}).WithName("ScheduleExpire")
@ -62,7 +70,7 @@ namespace Sample21.Workflows
.Then<SendMassTransitMessage>(activity =>
{
activity.Message = new JavaScriptExpression<CartRemovedEvent>("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");

10
src/samples/Sample21/appsettings.json
View file

@ -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"
}
}
}
}

13
src/samples/Sample26/Messages/CreateOrder.cs Normal file
View file

@ -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; }
}
}

13
src/samples/Sample26/Messages/OrderShipped.cs Normal file
View file

@ -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; }
}
}

8
src/samples/Sample26/Models/Customer.cs Normal file
View file

@ -0,0 +1,8 @@
namespace Sample26.Models
{
public class Customer
{
public string Name { get; set; }
public string Email { get; set; }
}
}

10
src/samples/Sample26/Models/Order.cs Normal file
View file

@ -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; }
}
}

14
src/samples/Sample26/Program.cs Normal file
View file

@ -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<Startup>(); });
}
}

30
src/samples/Sample26/Properties/launchSettings.json Normal file
View file

@ -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"
}
}
}
}

13
src/samples/Sample26/Sample26.csproj Normal file
View file

@ -0,0 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<TargetFramework>netcoreapp3.1</TargetFramework>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\activities\Elsa.Activities.MassTransit\Elsa.Activities.MassTransit.csproj" />
<ProjectReference Include="..\..\core\Elsa\Elsa.csproj" />
</ItemGroup>
</Project>

18
src/samples/Sample26/Sample26.http Normal file
View file

@ -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
}
###

56
src/samples/Sample26/Startup.cs Normal file
View file

@ -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<CreateOrderWorkflow>()
.AddWorkflow<HandleOrderWorkflow>();
}
public void Configure(IApplicationBuilder app, IHostEnvironment env)
{
if (env.IsDevelopment())
app.UseDeveloperExceptionPage();
app.UseHttpActivities();
}
}
}

80
src/samples/Sample26/Workflows/CreateOrderWorkflow.cs Normal file
View file

@ -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<ReceiveHttpRequest>(
activity =>
{
activity.Method = HttpMethod.Post.Method;
activity.Path = new Uri("/orders", UriKind.RelativeOrAbsolute);
activity.ReadContent = true;
}
)
.Then<SetVariable>(
activity =>
{
activity.VariableName = "order";
activity.ValueExpression = new JavaScriptExpression<object>("lastResult().Body");
}
)
// Need to ensure that the correlation ID is the same string format that is used by the WorkflowConsumer<T>
// MassTransit will always use Guid values for correlation ID's, so need to ensure the same string format is used.
.Then<Correlate>(activity => activity.ValueExpression = new JavaScriptExpression<string>("newGuid()"))
.Then<SendMassTransitMessage>(activity =>
{
activity.Message = new JavaScriptExpression<CreateOrder>("return { correlationId: correlationId(), order: order};");
activity.MessageType = typeof(CreateOrder);
}
)
.Then<Fork>(
activity => activity.Branches = new[] { "Write-Response", "Await-Shipment" },
fork =>
{
fork
.When("Write-Response")
.Then<WriteHttpResponse>(
activity =>
{
activity.Content = new LiteralExpression("<h1>Order Received</h1><p>Your order has been received. Waiting for shipment.</p>");
activity.ContentType = "text/html";
activity.StatusCode = HttpStatusCode.Accepted;
}
);
fork
.When("Await-Shipment")
.Then<ReceiveMassTransitMessage>(activity => activity.MessageType = typeof(OrderShipped))
.Then<SendEmail>(
activity =>
{
activity.From = new LiteralExpression("shipment@acme.com");
activity.To = new JavaScriptExpression<string>("order.customer.email");
activity.Subject = new JavaScriptExpression<string>("`Your order with ID #${order.id} has been shipped!`");
activity.Body = new JavaScriptExpression<string>(
"`Dear ${order.customer.name}, your order has shipped!`"
);
}
);
}
);
}
}
}

35
src/samples/Sample26/Workflows/HandleOrderWorkflow.cs Normal file
View file

@ -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<ReceiveMassTransitMessage>(activity => activity.MessageType = typeof(CreateOrder))
.Then<SetVariable>(
activity =>
{
activity.VariableName = "order";
activity.ValueExpression = new JavaScriptExpression<object>("lastResult().Order");
}
)
.Then<TimerEvent>(activity => activity.TimeoutExpression = new LiteralExpression<TimeSpan>("00:00:05"))
.Then<PublishMassTransitMessage>(activity =>
{
activity.Message = new JavaScriptExpression<OrderShipped>("return { correlationId: correlationId(), order: order}");
activity.MessageType = typeof(OrderShipped);
}
);
}
}
}

6
src/samples/Sample26/app.config Normal file
View file

@ -0,0 +1,6 @@
<?xml version="1.0" encoding="utf-8" ?>
<configuration>
<runtime>
<gcServer enabled="true"/>
</runtime>
</configuration>

9
src/samples/Sample26/appsettings.Development.json Normal file
View file

@ -0,0 +1,9 @@
{
"Logging": {
"LogLevel": {
"Default": "Debug",
"System": "Information",
"Microsoft": "Information"
}
}
}

18
src/samples/Sample26/appsettings.json Normal file
View file

@ -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"
}
}
}

87
src/samples/Sample27/AzureServiceBusSchedulerMassTransitBuilder.cs Normal file
View file

@ -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<AzureServiceBusSchedulerOptions>
{
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<IOptions<AzureServiceBusSchedulerOptions>>().Value;
cfg.Host(options.ConnectionString);
cfg.UseMessageScheduler(options.MessageSchedule.SchedulerAddress);
cfg.ReceiveEndpoint("shopping_cart_service", ep =>
{
ep.ConfigureConsumer<CartRemovedConsumer>(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<CartCreated>(serviceProvider);
ep.ConfigureWorkflowConsumer<CartItemAdded>(serviceProvider);
ep.ConfigureWorkflowConsumer<OrderSubmitted>(serviceProvider);
ep.ConfigureWorkflowConsumer<CartExpiredEvent>(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<ScheduleMessageConsumer>(serviceProvider);
e.ConfigureConsumer<CancelScheduledMessageConsumer>(serviceProvider);
});
});
scheduler.JobFactory = new MassTransitJobFactory(bus);
return bus;
}
protected override void ConfigureMassTransit(IServiceCollectionConfigurator configurator)
{
// Configure scheduler service consumers.
configurator.AddConsumer<ScheduleMessageConsumer>();
configurator.AddConsumer<CancelScheduledMessageConsumer>();
// configure workflow consumers
configurator.AddWorkflowConsumer<CartCreated>();
configurator.AddWorkflowConsumer<CartItemAdded>();
configurator.AddWorkflowConsumer<OrderSubmitted>();
configurator.AddWorkflowConsumer<CartExpiredEvent>();
// host fake service consumers
configurator.AddConsumer<CartRemovedConsumer>();
}
}
}

9
src/samples/Sample27/AzureServiceBusSchedulerOptions.cs Normal file
View file

@ -0,0 +1,9 @@
using Elsa.Activities.MassTransit.Options;
namespace Sample27
{
public class AzureServiceBusSchedulerOptions : AzureServiceBusOptions
{
public MessageScheduleOptions MessageSchedule { get; set; }
}
}

24
src/samples/Sample27/Consumers/CartRemovedConsumer.cs Normal file
View file

@ -0,0 +1,24 @@
using System.Threading.Tasks;
using MassTransit;
using Sample27.Controllers;
using Sample27.Messages;
namespace Sample27.Consumers
{
public class CartRemovedConsumer : IConsumer<CartRemovedEvent>
{
private readonly ICarts carts;
public CartRemovedConsumer(ICarts carts)
{
this.carts = carts;
}
public Task Consume(ConsumeContext<CartRemovedEvent> context)
{
carts.Remove(context.Message.CartId);
return Task.CompletedTask;
}
}
}

51
src/samples/Sample27/Controllers/CartController.cs Normal file
View file

@ -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<Cart> Get()
{
var cart = await carts.Cart(Username);
return cart ?? new Cart
{
Id = Guid.Empty,
Username = Username,
Items = new List<CartItem>()
};
}
// 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);
}
}
}

99
src/samples/Sample27/Controllers/Carts.cs Normal file
View file

@ -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<string, Cart> ActiveCarts = new ConcurrentDictionary<string, Cart>();
private readonly ISendEndpointProvider sender;
public Carts(ISendEndpointProvider sender)
{
this.sender = sender;
}
public Task<Cart> 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<CartItem>()
};
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 _);
}
}
}
}

14
src/samples/Sample27/Controllers/ICarts.cs Normal file
View file

@ -0,0 +1,14 @@
using System;
using System.Threading.Tasks;
using Sample27.Models;
namespace Sample27.Controllers
{
public interface ICarts
{
Task<Cart> Cart(string username);
Task AddItem(string username, CartItem item);
Task Submit(string username);
void Remove(Guid cartId);
}
}

15
src/samples/Sample27/Messages/CartCreated.cs Normal file
View file

@ -0,0 +1,15 @@
using System;
using MassTransit;
namespace Sample27.Messages
{
public class CartCreated : CorrelatedBy<Guid>
{
public Guid CorrelationId => CartId;
public Guid CartId { get; set; }
public DateTime Timestamp { get; set; }
public string UserName { get; set; }
}
}

11
src/samples/Sample27/Messages/CartExpiredEvent.cs Normal file
View file

@ -0,0 +1,11 @@
using System;
using MassTransit;
namespace Sample27.Messages
{
public class CartExpiredEvent : CorrelatedBy<Guid>
{
public Guid CorrelationId { get; set; }
public Guid CartId { get; set; }
}
}

16
src/samples/Sample27/Messages/CartItemAdded.cs Normal file
View file

@ -0,0 +1,16 @@
using System;
using MassTransit;
namespace Sample27.Messages
{
public class CartItemAdded : CorrelatedBy<Guid>
{
public Guid CorrelationId => CartId;
public Guid CartId { get; set; }
public string UserName { get; set; }
public DateTime Timestamp { get; set; }
}
}

12
src/samples/Sample27/Messages/CartRemovedEvent.cs Normal file
View file

@ -0,0 +1,12 @@
using System;
using MassTransit;
namespace Sample27.Messages
{
public class CartRemovedEvent : CorrelatedBy<Guid>
{
public Guid CorrelationId => CartId;
public Guid CartId { get; set; }
}
}

18
src/samples/Sample27/Messages/OrderSubmitted.cs Normal file
View file

@ -0,0 +1,18 @@
using System;
using MassTransit;
namespace Sample27.Messages
{
public class OrderSubmitted : CorrelatedBy<Guid>
{
public Guid CorrelationId => CartId;
public Guid OrderId { get; set; }
public DateTime Timestamp { get; set; }
public Guid CartId { get; set; }
public string UserName { get; set; }
}
}

14
src/samples/Sample27/Models/Cart.cs Normal file
View file

@ -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<CartItem> Items { get; set; }
}
}

9
src/samples/Sample27/Models/CartItem.cs Normal file
View file

@ -0,0 +1,9 @@
namespace Sample27.Models
{
public class CartItem
{
public int Quantity { get; set; }
public string ProductSku { get; set; }
}
}

14
src/samples/Sample27/Program.cs Normal file
View file

@ -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<Startup>(); });
}
}

30
src/samples/Sample27/Properties/launchSettings.json Normal file
View file

@ -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"
}
}
}
}

21
src/samples/Sample27/Sample27.csproj Normal file
View file

@ -0,0 +1,21 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<TargetFramework>netcoreapp3.1</TargetFramework>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="3.1.9" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="3.1.9">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\activities\Elsa.Activities.MassTransit\Elsa.Activities.MassTransit.csproj" />
<ProjectReference Include="..\..\core\Elsa\Elsa.csproj" />
</ItemGroup>
</Project>

21
src/samples/Sample27/Sample27.http Normal file
View file

@ -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
###

27
src/samples/Sample27/Services/QuartzHostedService.cs Normal file
View file

@ -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);
}
}
}

110
src/samples/Sample27/Startup.cs Normal file
View file

@ -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<CartTrackingWorkflow>()
.AddScoped<ICarts, Carts>()
.AddSingleton(scheduler)
.AddSingleton<IHostedService, QuartzHostedService>(); // 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();
}
}
}
}

92
src/samples/Sample27/Workflows/CartTrackingWorkflow.cs Normal file
View file

@ -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<AzureServiceBusSchedulerOptions> options)
{
this.options = options.Value;
}
public void Build(IWorkflowBuilder builder)
{
builder.StartWith<ReceiveMassTransitMessage>(activity => activity.MessageType = typeof(CartCreated))
.Then<SetVariable>(activity =>
{
activity.VariableName = "LastUpdateTimestamp";
activity.ValueExpression = new JavaScriptExpression<Instant>("instantFromDateTimeUtc(lastResult().Timestamp)");
})
.Then<SetVariable>(activity =>
{
activity.VariableName = "IsExpired";
activity.ValueExpression = new JavaScriptExpression<bool>("false");
})
.Fork(
action => action.Branches = new [] {"Item-Added", "Cart-Expired", "Order-Submitted"},
fork =>
{
fork.When("Item-Added")
.Then<ScheduleSendMassTransitMessage>(
activity =>
{
activity.MessageType = typeof(CartExpiredEvent);
activity.EndpointAddress = new Uri($"{options.Host}/shopping_cart_state");
activity.ScheduledTime = new JavaScriptExpression<DateTime>("plus(LastUpdateTimestamp, durationFromSeconds(10)).ToDateTimeUtc()");
activity.Message = new JavaScriptExpression<CartExpiredEvent>("return { correlationId: correlationId(), cartId: correlationId() }");
}).WithName("ScheduleExpire")
.Then<SetVariable>(activity =>
{
activity.VariableName = "ScheduleTokenId";
activity.ValueExpression = new JavaScriptExpression<bool>("lastResult()");
})
.Then<ReceiveMassTransitMessage>(activity => activity.MessageType = typeof(CartItemAdded))
.Then<SetVariable>(activity =>
{
activity.VariableName = "LastUpdateTimestamp";
activity.ValueExpression = new JavaScriptExpression<Instant>("instantFromDateTimeUtc(lastResult().Timestamp)");
})
.Then<CancelScheduledMassTransitMessage>(activity => activity.TokenId = new JavaScriptExpression<Guid>("return ScheduleTokenId"))
.Then("ScheduleExpire");
fork.When("Cart-Expired")
.Then<ReceiveMassTransitMessage>(activity => activity.MessageType = typeof(CartExpiredEvent))
.Then<SetVariable>(activity =>
{
activity.VariableName = "IsExpired";
activity.ValueExpression = new JavaScriptExpression<bool>("true");
})
.Then<SendMassTransitMessage>(activity =>
{
activity.Message = new JavaScriptExpression<CartRemovedEvent>("return { cartId: correlationId() };");
activity.EndpointAddress = new Uri($"{options.Host}/shopping_cart_service");
activity.MessageType = typeof(CartRemovedEvent);
})
.Then("Join");
fork.When("OrderSubmitted")
.Then<ReceiveMassTransitMessage>(activity => activity.MessageType = typeof(OrderSubmitted))
.Then<SetVariable>(activity =>
{
activity.VariableName = "LastUpdateTimestamp";
activity.ValueExpression = new JavaScriptExpression<DateTime>("lastResult().Timestamp");
})
.Then<CancelScheduledMassTransitMessage>(activity => activity.TokenId = new JavaScriptExpression<Guid>("return ScheduleTokenId"))
.Then("Join");
})
.Join(x => x.Mode = Join.JoinMode.WaitAny).WithName("Join");
}
}
}

6
src/samples/Sample27/app.config Normal file
View file

@ -0,0 +1,6 @@
<?xml version="1.0" encoding="utf-8" ?>
<configuration>
<runtime>
<gcServer enabled="true"/>
</runtime>
</configuration>

9
src/samples/Sample27/appsettings.Development.json Normal file
View file

@ -0,0 +1,9 @@
{
"Logging": {
"LogLevel": {
"Default": "Debug",
"System": "Information",
"Microsoft": "Information"
}
}
}

21
src/samples/Sample27/appsettings.json Normal file
View file

@ -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"
}
}
}
}