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

Add support for pub-sub consumers and implement Rebus distributed cache signaling

Browse source
This commit is contained in:
Sipke Schoorstra 2021-05-22 11:22:02 +02:00
parent 01839b19e5
commit b5453b23a2
23 changed files with 219 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
Elsa.sln
View file

@ -289,6 +289,10 @@ EndProjectSection
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.SignalApi", "src\samples\aspnet\Elsa.Samples.SignalApi\Elsa.Samples.SignalApi.csproj", "{3E2423CF-50E6-4D2B-8749-17B1EF540FE4}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "caching", "caching", "{9937FB02-72D9-4FCD-B31E-8E60CFF1B37F}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Caching.Rebus", "src\caching\Elsa.Caching.Rebus\Elsa.Caching.Rebus.csproj", "{61C16CA0-B190-4642-A81A-5C03705CA2C1}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -679,6 +683,10 @@ Global
{3E2423CF-50E6-4D2B-8749-17B1EF540FE4}.Debug|Any CPU.Build.0 = Debug|Any CPU
{3E2423CF-50E6-4D2B-8749-17B1EF540FE4}.Release|Any CPU.ActiveCfg = Release|Any CPU
{3E2423CF-50E6-4D2B-8749-17B1EF540FE4}.Release|Any CPU.Build.0 = Release|Any CPU
{61C16CA0-B190-4642-A81A-5C03705CA2C1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{61C16CA0-B190-4642-A81A-5C03705CA2C1}.Debug|Any CPU.Build.0 = Debug|Any CPU
{61C16CA0-B190-4642-A81A-5C03705CA2C1}.Release|Any CPU.ActiveCfg = Release|Any CPU
{61C16CA0-B190-4642-A81A-5C03705CA2C1}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@ -810,6 +818,8 @@ Global
{28C5C63F-2DCF-46BE-9576-1F50A7CFE8C4} = {164EB38A-D6D2-4092-8835-C944ECEE357C}
{5F81C84A-4C04-48FF-981C-23E48A7ACA6C} = {28C5C63F-2DCF-46BE-9576-1F50A7CFE8C4}
{3E2423CF-50E6-4D2B-8749-17B1EF540FE4} = {22E75696-6FE9-436A-9097-EE21C603F818}
{9937FB02-72D9-4FCD-B31E-8E60CFF1B37F} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925}
{61C16CA0-B190-4642-A81A-5C03705CA2C1} = {9937FB02-72D9-4FCD-B31E-8E60CFF1B37F}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {8B0975FD-7050-48B0-88C5-48C33378E158}

8
src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs
View file

@ -38,10 +38,10 @@ namespace Elsa.Activities.AzureServiceBus.Extensions
.AddBookmarkProvider<TopicMessageReceivedBookmarkProvider>()
;
options.AddConsumer<RestartServiceBusQueuesConsumer, WorkflowDefinitionPublished>();
options.AddConsumer<RestartServiceBusQueuesConsumer, WorkflowDefinitionRetracted>();
options.AddConsumer<RestartServiceBusTopicsConsumer, WorkflowDefinitionPublished>();
options.AddConsumer<RestartServiceBusTopicsConsumer, WorkflowDefinitionRetracted>();
options.AddCompetingConsumer<RestartServiceBusQueuesConsumer, WorkflowDefinitionPublished>();
options.AddCompetingConsumer<RestartServiceBusQueuesConsumer, WorkflowDefinitionRetracted>();
options.AddCompetingConsumer<RestartServiceBusTopicsConsumer, WorkflowDefinitionPublished>();
options.AddCompetingConsumer<RestartServiceBusTopicsConsumer, WorkflowDefinitionRetracted>();
options
.AddActivity<AzureServiceBusQueueMessageReceived>()

2
src/activities/Elsa.Activities.Telnyx/Extensions/SetupExtensions.cs
View file

@ -39,7 +39,7 @@ namespace Elsa.Activities.Telnyx.Extensions
// Activities.
elsaOptions
.AddActivitiesFrom<AnswerCall>()
.AddConsumer<TriggerWorkflows, TelnyxWebhookReceived>();
.AddCompetingConsumer<TriggerWorkflows, TelnyxWebhookReceived>();
// Services.
services

19
src/caching/Elsa.Caching.Rebus/Consumers/TriggerCacheSignalConsumer.cs Normal file
View file

@ -0,0 +1,19 @@
using System.Threading.Tasks;
using Elsa.Caching.Rebus.Messages;
using Rebus.Handlers;
namespace Elsa.Caching.Rebus.Consumers
{
public class TriggerCacheSignalConsumer : IHandleMessages<TriggerCacheSignal>
{
private readonly ICacheSignal _cacheSignal;
public TriggerCacheSignalConsumer(ICacheSignal cacheSignal) => _cacheSignal = cacheSignal;
public Task Handle(TriggerCacheSignal message)
{
_cacheSignal.TriggerToken(message.Key);
return Task.CompletedTask;
}
}
}

11
src/caching/Elsa.Caching.Rebus/Elsa.Caching.Rebus.csproj Normal file
View file

@ -0,0 +1,11 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net5.0</TargetFramework>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\core\Elsa.Core\Elsa.Core.csproj" />
</ItemGroup>
</Project>

19
src/caching/Elsa.Caching.Rebus/Extensions/ServiceCollectionExtensions.cs Normal file
View file

@ -0,0 +1,19 @@
using Elsa.Caching.Rebus.Consumers;
using Elsa.Caching.Rebus.Messages;
using Elsa.Caching.Rebus.Services;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Caching.Rebus.Extensions
{
public static class ServiceCollectionExtensions
{
public static ElsaOptionsBuilder AddRebusCacheSignal(this ElsaOptionsBuilder builder)
{
var services = builder.Services;
services.Decorate<ICacheSignal, RebusCacheSignal>();
builder.AddPubSubConsumer<TriggerCacheSignalConsumer, TriggerCacheSignal>();
return builder;
}
}
}

4
src/caching/Elsa.Caching.Rebus/Messages/TriggerCacheSignal.cs Normal file
View file

@ -0,0 +1,4 @@
namespace Elsa.Caching.Rebus.Messages
{
public record TriggerCacheSignal(string Key);
}

29
src/caching/Elsa.Caching.Rebus/Services/RebusCacheSignal.cs Normal file
View file

@ -0,0 +1,29 @@
using System.Threading.Tasks;
using Elsa.Caching.Rebus.Messages;
using Elsa.Services;
using Microsoft.Extensions.Primitives;
namespace Elsa.Caching.Rebus.Services
{
public class RebusCacheSignal : ICacheSignal
{
private readonly ICacheSignal _cacheSignal;
private readonly IEventPublisher _eventPublisher;
public RebusCacheSignal(ICacheSignal cacheSignal, IEventPublisher eventPublisher)
{
_cacheSignal = cacheSignal;
_eventPublisher = eventPublisher;
}
public IChangeToken GetToken(string key) => _cacheSignal.GetToken(key);
public void TriggerToken(string key) => _cacheSignal.TriggerToken(key);
public async ValueTask TriggerTokenAsync(string key)
{
await _cacheSignal.TriggerTokenAsync(key);
await _eventPublisher.PublishAsync(new TriggerCacheSignal(key));
}
}
}

8
src/core/Elsa.Abstractions/Caching/ISignal.cs → src/core/Elsa.Abstractions/Caching/ICacheSignal.cs
View file

@ -1,13 +1,15 @@
using Microsoft.Extensions.Primitives;
using System.Threading.Tasks;
using Microsoft.Extensions.Primitives;
namespace Elsa.Caching
{
/// <summary>
/// Provides change tokens for memory caches, allowing code to evict cache entries by triggering a signal.
/// </summary>
public interface ISignal
public interface ICacheSignal
{
IChangeToken GetToken(string key);
void Trigger(string key);
void TriggerToken(string key);
ValueTask TriggerTokenAsync(string key);
}
}

7
src/core/Elsa.Abstractions/Services/IContainerNameAccessor.cs Normal file
View file

@ -0,0 +1,7 @@
namespace Elsa.Services
{
public interface IContainerNameAccessor
{
string GetContainerName();
}
}

1
src/core/Elsa.Abstractions/Services/IServiceBusFactory.cs
View file

@ -7,6 +7,7 @@ namespace Elsa.Services
{
public interface IServiceBusFactory
{
Task<IBus> GetServiceBusAsync(Type messageType, string queueName, CancellationToken cancellationToken = default);
Task<IBus> GetServiceBusAsync(Type messageType, CancellationToken cancellationToken = default);
}
}

18
src/core/Elsa.Core/Caching/Signal.cs → src/core/Elsa.Core/Caching/CacheSignal.cs
View file

@ -1,14 +1,15 @@
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Primitives;
namespace Elsa.Caching
{
public class Signal : ISignal
public class CacheSignal : ICacheSignal
{
private readonly ConcurrentDictionary<string, ChangeTokenInfo> _changeTokens;
public Signal()
public CacheSignal()
{
_changeTokens = new ConcurrentDictionary<string, ChangeTokenInfo>();
}
@ -25,12 +26,19 @@ namespace Elsa.Caching
}).ChangeToken;
}
public void Trigger(string key)
public void TriggerToken(string key)
{
if (_changeTokens.TryRemove(key, out var changeTokenInfo)) changeTokenInfo.TokenSource.Cancel();
if (_changeTokens.TryRemove(key, out var changeTokenInfo))
changeTokenInfo.TokenSource.Cancel();
}
private struct ChangeTokenInfo
public ValueTask TriggerTokenAsync(string key)
{
TriggerToken(key);
return new ValueTask();
}
private readonly struct ChangeTokenInfo
{
public ChangeTokenInfo(IChangeToken changeToken, CancellationTokenSource tokenSource)
{

10
src/core/Elsa.Core/Decorators/CachingWorkflowRegistry.cs
View file

@ -19,13 +19,13 @@ namespace Elsa.Decorators
private const string CacheKey = "WorkflowRegistry";
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IMemoryCache _memoryCache;
private readonly ISignal _signal;
private readonly ICacheSignal _cacheSignal;
public CachingWorkflowRegistry(IWorkflowRegistry workflowRegistry, IMemoryCache memoryCache, ISignal signal)
public CachingWorkflowRegistry(IWorkflowRegistry workflowRegistry, IMemoryCache memoryCache, ICacheSignal cacheSignal)
{
_workflowRegistry = workflowRegistry;
_memoryCache = memoryCache;
_signal = signal;
_cacheSignal = cacheSignal;
}
public async Task<IEnumerable<IWorkflowBlueprint>> ListAsync(CancellationToken cancellationToken) => await GetWorkflowBlueprints(cancellationToken);
@ -50,14 +50,14 @@ namespace Elsa.Decorators
{
return await _memoryCache.GetOrCreateAsync(CacheKey, async entry =>
{
entry.Monitor(_signal.GetToken(CacheKey));
entry.Monitor(_cacheSignal.GetToken(CacheKey));
return await _workflowRegistry.ListAsync(cancellationToken).ToList();
});
}
Task INotificationHandler<WorkflowDefinitionSaved>.Handle(WorkflowDefinitionSaved notification, CancellationToken cancellationToken)
{
_signal.Trigger(CacheKey);
_cacheSignal.TriggerTokenAsync(CacheKey);
return Task.CompletedTask;
}
}

8
src/core/Elsa.Core/ElsaOptions.cs
View file

@ -30,7 +30,6 @@ namespace Elsa
WorkflowInstanceDispatcherFactory = sp => ActivatorUtilities.CreateInstance<QueuingWorkflowDispatcher>(sp);
CorrelatingWorkflowDispatcherFactory = sp => ActivatorUtilities.CreateInstance<QueuingWorkflowDispatcher>(sp);
StorageFactory = sp => Storage.Net.StorageFactory.Blobs.InMemory();
SignalFactory = sp => new Signal();
JsonSerializerConfigurer = (sp, serializer) => { };
DistributedLockingOptions = new DistributedLockingOptions();
ConfigureServiceBusEndpoint = ConfigureInMemoryServiceBusEndpoint;
@ -42,13 +41,15 @@ namespace Elsa
return serializer;
};
}
public string ContainerName { get; set; } = Environment.MachineName;
public ServiceFactory<IActivity> ActivityFactory { get; } = new();
public ServiceFactory<IWorkflow> WorkflowFactory { get; } = new();
public IEnumerable<Type> ActivityTypes => ActivityFactory.Types;
public IList<Type> WorkflowTypes { get; } = new List<Type>();
public IList<Type> MessageTypes { get; } = new List<Type>();
public IList<Type> CompetingMessageTypes { get; } = new List<Type>();
public IList<Type> PubSubMessageTypes { get; } = new List<Type>();
public ServiceBusOptions ServiceBusOptions { get; } = new();
public DistributedLockingOptions DistributedLockingOptions { get; set; }
@ -62,7 +63,6 @@ namespace Elsa
internal Func<IServiceProvider, IWorkflowInstanceStore> WorkflowInstanceStoreFactory { get; set; }
internal Func<IServiceProvider, IWorkflowExecutionLogStore> WorkflowExecutionLogStoreFactory { get; set; }
internal Func<IServiceProvider, IBookmarkStore> WorkflowTriggerStoreFactory { get; set; }
internal Func<IServiceProvider, ISignal> SignalFactory { get; set; }
internal Func<IServiceProvider, JsonSerializer> CreateJsonSerializer { get; set; }
internal Action<IServiceProvider, JsonSerializer> JsonSerializerConfigurer { get; set; }
internal Func<IServiceProvider, IWorkflowDefinitionDispatcher> WorkflowDefinitionDispatcherFactory { get; set; }

28
src/core/Elsa.Core/ElsaOptionsBuilder.cs
View file

@ -2,6 +2,7 @@ using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using Elsa.Activities.Signaling.Services;
using Elsa.Builders;
using Elsa.Caching;
using Elsa.Persistence;
@ -32,6 +33,7 @@ namespace Elsa
services.AddSingleton<InMemorySubscriberStore>();
services.AddSingleton<InMemDataStore>();
services.AddMemoryCache();
services.AddSingleton<ICacheSignal, CacheSignal>();
DistributedLockingOptionsBuilder = new DistributedLockingOptionsBuilder(this);
}
@ -48,6 +50,12 @@ namespace Elsa
return this;
}
public ElsaOptionsBuilder WithContainerName(string name)
{
ElsaOptions.ContainerName = name;
return this;
}
public ElsaOptionsBuilder AddActivity<T>() where T : IActivity => AddActivity(typeof(T));
public ElsaOptionsBuilder AddActivity(Type activityType)
@ -132,13 +140,21 @@ namespace Elsa
return this;
}
public ElsaOptionsBuilder AddMessageType(Type messageType)
public ElsaOptionsBuilder AddCompetingMessageType(Type messageType)
{
ElsaOptions.MessageTypes.Add(messageType);
ElsaOptions.CompetingMessageTypes.Add(messageType);
return this;
}
public ElsaOptionsBuilder AddMessageType<T>() => AddMessageType(typeof(T));
public ElsaOptionsBuilder AddCompetingMessageType<T>() => AddCompetingMessageType(typeof(T));
public ElsaOptionsBuilder AddPubSubMessageType(Type messageType)
{
ElsaOptions.PubSubMessageTypes.Add(messageType);
return this;
}
public ElsaOptionsBuilder AddPubSubMessageType<T>() => AddPubSubMessageType(typeof(T));
public ElsaOptionsBuilder ConfigureDistributedLockProvider(Action<DistributedLockingOptionsBuilder> configureOptions)
{
@ -146,12 +162,6 @@ namespace Elsa
return this;
}
public ElsaOptionsBuilder UseSignal(Func<IServiceProvider, ISignal> factory)
{
ElsaOptions.SignalFactory = factory;
return this;
}
public ElsaOptionsBuilder UseStorage(Func<IBlobStorage> factory) => UseStorage(_ => factory());
public ElsaOptionsBuilder UseStorage(Func<IServiceProvider, IBlobStorage> factory)

31
src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs
View file

@ -64,7 +64,6 @@ namespace Microsoft.Extensions.DependencyInjection
.AddScoped(options.WorkflowExecutionLogStoreFactory)
.AddScoped(options.WorkflowTriggerStoreFactory)
.AddSingleton(options.DistributedLockingOptions.DistributedLockProviderFactory)
.AddSingleton(options.SignalFactory)
.AddSingleton(options.StorageFactory)
.AddSingleton(options.WorkflowDefinitionDispatcherFactory)
.AddSingleton(options.WorkflowInstanceDispatcherFactory)
@ -95,10 +94,25 @@ namespace Microsoft.Extensions.DependencyInjection
return elsaOptions.Services.AddHostedService<StartWorkflow<T>>();
}
public static ElsaOptionsBuilder AddConsumer<TConsumer, TMessage>(this ElsaOptionsBuilder elsaOptions) where TConsumer : class, IHandleMessages<TMessage>
/// <summary>
/// Registers a consumer and associated message type using the competing consumer pattern. With the competing consumer pattern, only the first consumer on a given node to obtain a message will handle that message.
/// This is in contrast to the Publisher-Subscriber pattern, where a message will be delivered to the consumer on all nodes in a cluster. To register a Publisher-Subscriber consumer, use <seealso cref="AddPubSubConsumer{TConsumer,TMessage}"/>
/// </summary>
/// <param name="elsaOptions"></param>
/// <typeparam name="TConsumer"></typeparam>
/// <typeparam name="TMessage"></typeparam>
/// <returns></returns>
public static ElsaOptionsBuilder AddCompetingConsumer<TConsumer, TMessage>(this ElsaOptionsBuilder elsaOptions) where TConsumer : class, IHandleMessages<TMessage>
{
elsaOptions.Services.AddTransient<IHandleMessages<TMessage>, TConsumer>();
elsaOptions.AddMessageType<TMessage>();
elsaOptions.AddCompetingMessageType<TMessage>();
return elsaOptions;
}
public static ElsaOptionsBuilder AddPubSubConsumer<TConsumer, TMessage>(this ElsaOptionsBuilder elsaOptions) where TConsumer : class, IHandleMessages<TMessage>
{
elsaOptions.Services.AddTransient<IHandleMessages<TMessage>, TConsumer>();
elsaOptions.AddPubSubMessageType<TMessage>();
return elsaOptions;
}
@ -116,6 +130,7 @@ namespace Microsoft.Extensions.DependencyInjection
services
.AddLogging()
.AddLocalization()
.AddSingleton<IContainerNameAccessor, OptionsContainerNameAccessor>()
.AddSingleton<IIdGenerator, IdGenerator>()
.AddScoped<IWorkflowRegistry, WorkflowRegistry>()
.AddSingleton<IActivityActivator, ActivityActivator>()
@ -202,11 +217,11 @@ namespace Microsoft.Extensions.DependencyInjection
.AddSingleton<IEventPublisher, EventPublisher>();
options
.AddConsumer<TriggerWorkflowsRequestConsumer, TriggerWorkflowsRequest>()
.AddConsumer<ExecuteWorkflowDefinitionRequestConsumer, ExecuteWorkflowDefinitionRequest>()
.AddConsumer<ExecuteWorkflowInstanceRequestConsumer, ExecuteWorkflowInstanceRequest>()
.AddConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionPublished>()
.AddConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionRetracted>();
.AddCompetingConsumer<TriggerWorkflowsRequestConsumer, TriggerWorkflowsRequest>()
.AddCompetingConsumer<ExecuteWorkflowDefinitionRequestConsumer, ExecuteWorkflowDefinitionRequest>()
.AddCompetingConsumer<ExecuteWorkflowInstanceRequestConsumer, ExecuteWorkflowInstanceRequest>()
.AddCompetingConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionPublished>()
.AddCompetingConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionRetracted>();
// AutoMapper.
services

9
src/core/Elsa.Core/Services/OptionsContainerNameAccessor.cs Normal file
View file

@ -0,0 +1,9 @@
namespace Elsa.Services
{
public class OptionsContainerNameAccessor : IContainerNameAccessor
{
private readonly ElsaOptions _elsaOptions;
public OptionsContainerNameAccessor(ElsaOptions elsaOptions) => _elsaOptions = elsaOptions;
public string GetContainerName() => _elsaOptions.ContainerName;
}
}

14
src/core/Elsa.Core/Services/ServiceBusFactory.cs
View file

@ -30,19 +30,21 @@ namespace Elsa.Services
_handlerActivator = new DependencyInjectionHandlerActivator(serviceProvider);
}
public async Task<IBus> GetServiceBusAsync(Type messageType, CancellationToken cancellationToken)
public Task<IBus> GetServiceBusAsync(Type messageType, CancellationToken cancellationToken = default) => GetServiceBusAsync(messageType, messageType.Name, cancellationToken);
public async Task<IBus> GetServiceBusAsync(Type messageType, string queueName, CancellationToken cancellationToken = default)
{
var queueName = FormatQueueName(messageType.Name);
var formattedQueueName = FormatQueueName(queueName);
await _semaphore.WaitAsync(cancellationToken);
try
{
if (_serviceBuses.TryGetValue(queueName, out var bus))
if (_serviceBuses.TryGetValue(formattedQueueName, out var bus))
return bus;
var configurer = Configure.With(_handlerActivator);
var map = new Dictionary<Type, string> { [messageType] = queueName };
var configureContext = new ServiceBusEndpointConfigurationContext(configurer, queueName, map, _serviceProvider);
var map = new Dictionary<Type, string> { [messageType] = formattedQueueName };
var configureContext = new ServiceBusEndpointConfigurationContext(configurer, formattedQueueName, map, _serviceProvider);
// Default options.
configurer
@ -55,7 +57,7 @@ namespace Elsa.Services
_elsaOptions.ConfigureServiceBusEndpoint(configureContext);
var newBus = configurer.Start();
_serviceBuses.TryAdd(queueName, newBus);
_serviceBuses.TryAdd(formattedQueueName, newBus);
return newBus;
}

20
src/core/Elsa.Core/StartupTasks/CreateSubscriptions.cs
View file

@ -9,23 +9,35 @@ namespace Elsa.StartupTasks
public class CreateSubscriptions : IStartupTask
{
private readonly IServiceBusFactory _serviceBusFactory;
private readonly IEnumerable<Type> _messageTypes;
private readonly IContainerNameAccessor _containerNameAccessor;
private readonly IEnumerable<Type> _competingMessageTypes;
private readonly IEnumerable<Type> _pubSubMessageTypes;
public CreateSubscriptions(IServiceBusFactory serviceBusFactory, ElsaOptions elsaOptions)
public CreateSubscriptions(IServiceBusFactory serviceBusFactory, ElsaOptions elsaOptions, IContainerNameAccessor containerNameAccessor)
{
_serviceBusFactory = serviceBusFactory;
_messageTypes = elsaOptions.MessageTypes;
_containerNameAccessor = containerNameAccessor;
_competingMessageTypes = elsaOptions.CompetingMessageTypes;
_pubSubMessageTypes = elsaOptions.PubSubMessageTypes;
}
public int Order => 900;
public async Task ExecuteAsync(CancellationToken cancellationToken)
{
foreach (var messageType in _messageTypes)
foreach (var messageType in _competingMessageTypes)
{
var bus = await _serviceBusFactory.GetServiceBusAsync(messageType, cancellationToken);
await bus.Subscribe(messageType);
}
var containerName = _containerNameAccessor.GetContainerName();
foreach (var messageType in _pubSubMessageTypes)
{
var queueName = $"{containerName}:{messageType.Name}";
var bus = await _serviceBusFactory.GetServiceBusAsync(messageType, queueName, cancellationToken);
await bus.Subscribe(messageType);
}
}
}
}

2
src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj
View file

@ -13,6 +13,7 @@
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Temporal.Hangfire\Elsa.Activities.Temporal.Hangfire.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Temporal.Quartz\Elsa.Activities.Temporal.Quartz.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.UserTask\Elsa.Activities.UserTask.csproj" />
<ProjectReference Include="..\..\..\caching\Elsa.Caching.Rebus\Elsa.Caching.Rebus.csproj" />
<ProjectReference Include="..\..\..\persistence\Elsa.Persistence.EntityFramework\Elsa.Persistence.EntityFramework.PostgreSql\Elsa.Persistence.EntityFramework.PostgreSql.csproj" />
<ProjectReference Include="..\..\..\persistence\Elsa.Persistence.EntityFramework\Elsa.Persistence.EntityFramework.Sqlite\Elsa.Persistence.EntityFramework.Sqlite.csproj" />
<ProjectReference Include="..\..\..\persistence\Elsa.Persistence.EntityFramework\Elsa.Persistence.EntityFramework.SqlServer\Elsa.Persistence.EntityFramework.SqlServer.csproj" />
@ -20,6 +21,7 @@
<ProjectReference Include="..\..\..\server\Elsa.Server.Api\Elsa.Server.Api.csproj" />
<ProjectReference Include="..\..\..\server\Elsa.Server.Hangfire\Elsa.Server.Hangfire.csproj" />
<ProjectReference Include="..\..\..\server\Elsa.Server.Orleans\Elsa.Server.Orleans.csproj" />
<ProjectReference Include="..\..\..\servicebus\Elsa.Rebus.RabbitMq\Elsa.Rebus.RabbitMq.csproj" />
</ItemGroup>
<ItemGroup>

5
src/samples/server/Elsa.Samples.Server.Host/Startup.cs
View file

@ -1,9 +1,11 @@
using System;
using Elsa.Activities.UserTask.Extensions;
using Elsa.Caching.Rebus.Extensions;
using Elsa.Persistence.EntityFramework.Core.Extensions;
using Elsa.Persistence.EntityFramework.PostgreSql;
using Elsa.Persistence.EntityFramework.Sqlite;
using Elsa.Persistence.YesSql;
using Elsa.Rebus.RabbitMq.Extensions;
using Elsa.Samples.Server.Host.Activities;
using Hangfire;
using Microsoft.AspNetCore.Builder;
@ -36,7 +38,10 @@ namespace Elsa.Samples.Server.Host
.AddActivityPropertyOptionsProvider<VehicleActivity>()
.AddRuntimeSelectItemsProvider<VehicleActivity>()
.AddElsa(elsa => elsa
.WithContainerName(Configuration["ContainerName"] ?? System.Environment.MachineName)
.UseEntityFrameworkPersistence(ef => ef.UseSqlite())
.UseRabbitMq(Configuration.GetConnectionString("RabbitMq"))
.AddRebusCacheSignal()
.AddConsoleActivities()
.AddHttpActivities(elsaSection.GetSection("Http").Bind)
.AddEmailActivities(elsaSection.GetSection("Smtp").Bind)

3
src/samples/server/Elsa.Samples.Server.Host/appsettings.json
View file

@ -8,6 +8,9 @@
}
},
"AllowedHosts": "*",
"ConnectionStrings": {
"RabbitMq": "amqp://localhost:5672"
},
"Elsa": {
"Http": {
"BaseUrl": "https://localhost:11000",

4
src/servicebus/Elsa.Rebus.RabbitMq/Extensions/ElsaOptionsExtensions.cs
View file

@ -5,7 +5,7 @@ namespace Elsa.Rebus.RabbitMq.Extensions
{
public static class ElsaOptionsExtensions
{
public static ElsaOptionsBuilder UseRabbitMq(this ElsaOptionsBuilder elsaOptions, string connectionString) => elsaOptions.UseServiceBus(context => ConfigureAzureServiceBusEndpoint(context, connectionString));
private static void ConfigureAzureServiceBusEndpoint(ServiceBusEndpointConfigurationContext context, string connectionString) => context.Configurer.Transport(t => t.UseRabbitMq(connectionString, context.QueueName));
public static ElsaOptionsBuilder UseRabbitMq(this ElsaOptionsBuilder elsaOptions, string connectionString) => elsaOptions.UseServiceBus(context => ConfigureRabbitMqEndpoint(context, connectionString));
private static void ConfigureRabbitMqEndpoint(ServiceBusEndpointConfigurationContext context, string connectionString) => context.Configurer.Transport(t => t.UseRabbitMq(connectionString, context.QueueName));
}
}