From b5453b23a2a785c8141bbd757e1a71eb8167459b Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Sat, 22 May 2021 11:22:02 +0200 Subject: [PATCH] Add support for pub-sub consumers and implement Rebus distributed cache signaling --- Elsa.sln | 10 ++++++ .../Extensions/ServiceCollectionExtensions.cs | 8 ++--- .../Extensions/SetupExtensions.cs | 2 +- .../Consumers/TriggerCacheSignalConsumer.cs | 19 ++++++++++++ .../Elsa.Caching.Rebus.csproj | 11 +++++++ .../Extensions/ServiceCollectionExtensions.cs | 19 ++++++++++++ .../Messages/TriggerCacheSignal.cs | 4 +++ .../Services/RebusCacheSignal.cs | 29 +++++++++++++++++ .../Caching/{ISignal.cs => ICacheSignal.cs} | 8 +++-- .../Services/IContainerNameAccessor.cs | 7 +++++ .../Services/IServiceBusFactory.cs | 1 + .../Caching/{Signal.cs => CacheSignal.cs} | 18 ++++++++--- .../Decorators/CachingWorkflowRegistry.cs | 10 +++--- src/core/Elsa.Core/ElsaOptions.cs | 8 ++--- src/core/Elsa.Core/ElsaOptionsBuilder.cs | 28 +++++++++++------ .../ElsaServiceCollectionExtensions.cs | 31 ++++++++++++++----- .../Services/OptionsContainerNameAccessor.cs | 9 ++++++ .../Elsa.Core/Services/ServiceBusFactory.cs | 14 +++++---- .../StartupTasks/CreateSubscriptions.cs | 20 +++++++++--- .../Elsa.Samples.Server.Host.csproj | 2 ++ .../Elsa.Samples.Server.Host/Startup.cs | 5 +++ .../Elsa.Samples.Server.Host/appsettings.json | 3 ++ .../Extensions/ElsaOptionsExtensions.cs | 4 +-- 23 files changed, 219 insertions(+), 51 deletions(-) create mode 100644 src/caching/Elsa.Caching.Rebus/Consumers/TriggerCacheSignalConsumer.cs create mode 100644 src/caching/Elsa.Caching.Rebus/Elsa.Caching.Rebus.csproj create mode 100644 src/caching/Elsa.Caching.Rebus/Extensions/ServiceCollectionExtensions.cs create mode 100644 src/caching/Elsa.Caching.Rebus/Messages/TriggerCacheSignal.cs create mode 100644 src/caching/Elsa.Caching.Rebus/Services/RebusCacheSignal.cs rename src/core/Elsa.Abstractions/Caching/{ISignal.cs => ICacheSignal.cs} (54%) create mode 100644 src/core/Elsa.Abstractions/Services/IContainerNameAccessor.cs rename src/core/Elsa.Core/Caching/{Signal.cs => CacheSignal.cs} (75%) create mode 100644 src/core/Elsa.Core/Services/OptionsContainerNameAccessor.cs diff --git a/Elsa.sln b/Elsa.sln index ea87e24c7..426f37125 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -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} diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs index cafa5476b..ff75f7c33 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs @@ -38,10 +38,10 @@ namespace Elsa.Activities.AzureServiceBus.Extensions .AddBookmarkProvider() ; - options.AddConsumer(); - options.AddConsumer(); - options.AddConsumer(); - options.AddConsumer(); + options.AddCompetingConsumer(); + options.AddCompetingConsumer(); + options.AddCompetingConsumer(); + options.AddCompetingConsumer(); options .AddActivity() diff --git a/src/activities/Elsa.Activities.Telnyx/Extensions/SetupExtensions.cs b/src/activities/Elsa.Activities.Telnyx/Extensions/SetupExtensions.cs index 9c6c7b8e6..fa772fd00 100644 --- a/src/activities/Elsa.Activities.Telnyx/Extensions/SetupExtensions.cs +++ b/src/activities/Elsa.Activities.Telnyx/Extensions/SetupExtensions.cs @@ -39,7 +39,7 @@ namespace Elsa.Activities.Telnyx.Extensions // Activities. elsaOptions .AddActivitiesFrom() - .AddConsumer(); + .AddCompetingConsumer(); // Services. services diff --git a/src/caching/Elsa.Caching.Rebus/Consumers/TriggerCacheSignalConsumer.cs b/src/caching/Elsa.Caching.Rebus/Consumers/TriggerCacheSignalConsumer.cs new file mode 100644 index 000000000..ffae21d72 --- /dev/null +++ b/src/caching/Elsa.Caching.Rebus/Consumers/TriggerCacheSignalConsumer.cs @@ -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 + { + private readonly ICacheSignal _cacheSignal; + + public TriggerCacheSignalConsumer(ICacheSignal cacheSignal) => _cacheSignal = cacheSignal; + + public Task Handle(TriggerCacheSignal message) + { + _cacheSignal.TriggerToken(message.Key); + return Task.CompletedTask; + } + } +} \ No newline at end of file diff --git a/src/caching/Elsa.Caching.Rebus/Elsa.Caching.Rebus.csproj b/src/caching/Elsa.Caching.Rebus/Elsa.Caching.Rebus.csproj new file mode 100644 index 000000000..578b8ffce --- /dev/null +++ b/src/caching/Elsa.Caching.Rebus/Elsa.Caching.Rebus.csproj @@ -0,0 +1,11 @@ + + + + net5.0 + + + + + + + diff --git a/src/caching/Elsa.Caching.Rebus/Extensions/ServiceCollectionExtensions.cs b/src/caching/Elsa.Caching.Rebus/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 000000000..78e5d1c48 --- /dev/null +++ b/src/caching/Elsa.Caching.Rebus/Extensions/ServiceCollectionExtensions.cs @@ -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(); + builder.AddPubSubConsumer(); + + return builder; + } + } +} \ No newline at end of file diff --git a/src/caching/Elsa.Caching.Rebus/Messages/TriggerCacheSignal.cs b/src/caching/Elsa.Caching.Rebus/Messages/TriggerCacheSignal.cs new file mode 100644 index 000000000..54b079b39 --- /dev/null +++ b/src/caching/Elsa.Caching.Rebus/Messages/TriggerCacheSignal.cs @@ -0,0 +1,4 @@ +namespace Elsa.Caching.Rebus.Messages +{ + public record TriggerCacheSignal(string Key); +} \ No newline at end of file diff --git a/src/caching/Elsa.Caching.Rebus/Services/RebusCacheSignal.cs b/src/caching/Elsa.Caching.Rebus/Services/RebusCacheSignal.cs new file mode 100644 index 000000000..d4685b013 --- /dev/null +++ b/src/caching/Elsa.Caching.Rebus/Services/RebusCacheSignal.cs @@ -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)); + } + } +} \ No newline at end of file diff --git a/src/core/Elsa.Abstractions/Caching/ISignal.cs b/src/core/Elsa.Abstractions/Caching/ICacheSignal.cs similarity index 54% rename from src/core/Elsa.Abstractions/Caching/ISignal.cs rename to src/core/Elsa.Abstractions/Caching/ICacheSignal.cs index a8c80c51f..eef1766c7 100644 --- a/src/core/Elsa.Abstractions/Caching/ISignal.cs +++ b/src/core/Elsa.Abstractions/Caching/ICacheSignal.cs @@ -1,13 +1,15 @@ -using Microsoft.Extensions.Primitives; +using System.Threading.Tasks; +using Microsoft.Extensions.Primitives; namespace Elsa.Caching { /// /// Provides change tokens for memory caches, allowing code to evict cache entries by triggering a signal. /// - public interface ISignal + public interface ICacheSignal { IChangeToken GetToken(string key); - void Trigger(string key); + void TriggerToken(string key); + ValueTask TriggerTokenAsync(string key); } } \ No newline at end of file diff --git a/src/core/Elsa.Abstractions/Services/IContainerNameAccessor.cs b/src/core/Elsa.Abstractions/Services/IContainerNameAccessor.cs new file mode 100644 index 000000000..dbc894166 --- /dev/null +++ b/src/core/Elsa.Abstractions/Services/IContainerNameAccessor.cs @@ -0,0 +1,7 @@ +namespace Elsa.Services +{ + public interface IContainerNameAccessor + { + string GetContainerName(); + } +} \ No newline at end of file diff --git a/src/core/Elsa.Abstractions/Services/IServiceBusFactory.cs b/src/core/Elsa.Abstractions/Services/IServiceBusFactory.cs index fe351994f..f0c60b08c 100644 --- a/src/core/Elsa.Abstractions/Services/IServiceBusFactory.cs +++ b/src/core/Elsa.Abstractions/Services/IServiceBusFactory.cs @@ -7,6 +7,7 @@ namespace Elsa.Services { public interface IServiceBusFactory { + Task GetServiceBusAsync(Type messageType, string queueName, CancellationToken cancellationToken = default); Task GetServiceBusAsync(Type messageType, CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Caching/Signal.cs b/src/core/Elsa.Core/Caching/CacheSignal.cs similarity index 75% rename from src/core/Elsa.Core/Caching/Signal.cs rename to src/core/Elsa.Core/Caching/CacheSignal.cs index b4e0ee652..f0446199f 100644 --- a/src/core/Elsa.Core/Caching/Signal.cs +++ b/src/core/Elsa.Core/Caching/CacheSignal.cs @@ -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 _changeTokens; - public Signal() + public CacheSignal() { _changeTokens = new ConcurrentDictionary(); } @@ -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) { diff --git a/src/core/Elsa.Core/Decorators/CachingWorkflowRegistry.cs b/src/core/Elsa.Core/Decorators/CachingWorkflowRegistry.cs index 32bb57a3d..12c1a13d2 100644 --- a/src/core/Elsa.Core/Decorators/CachingWorkflowRegistry.cs +++ b/src/core/Elsa.Core/Decorators/CachingWorkflowRegistry.cs @@ -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> 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.Handle(WorkflowDefinitionSaved notification, CancellationToken cancellationToken) { - _signal.Trigger(CacheKey); + _cacheSignal.TriggerTokenAsync(CacheKey); return Task.CompletedTask; } } diff --git a/src/core/Elsa.Core/ElsaOptions.cs b/src/core/Elsa.Core/ElsaOptions.cs index c21b076a8..275bd8555 100644 --- a/src/core/Elsa.Core/ElsaOptions.cs +++ b/src/core/Elsa.Core/ElsaOptions.cs @@ -30,7 +30,6 @@ namespace Elsa WorkflowInstanceDispatcherFactory = sp => ActivatorUtilities.CreateInstance(sp); CorrelatingWorkflowDispatcherFactory = sp => ActivatorUtilities.CreateInstance(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 ActivityFactory { get; } = new(); public ServiceFactory WorkflowFactory { get; } = new(); public IEnumerable ActivityTypes => ActivityFactory.Types; public IList WorkflowTypes { get; } = new List(); - public IList MessageTypes { get; } = new List(); + public IList CompetingMessageTypes { get; } = new List(); + public IList PubSubMessageTypes { get; } = new List(); public ServiceBusOptions ServiceBusOptions { get; } = new(); public DistributedLockingOptions DistributedLockingOptions { get; set; } @@ -62,7 +63,6 @@ namespace Elsa internal Func WorkflowInstanceStoreFactory { get; set; } internal Func WorkflowExecutionLogStoreFactory { get; set; } internal Func WorkflowTriggerStoreFactory { get; set; } - internal Func SignalFactory { get; set; } internal Func CreateJsonSerializer { get; set; } internal Action JsonSerializerConfigurer { get; set; } internal Func WorkflowDefinitionDispatcherFactory { get; set; } diff --git a/src/core/Elsa.Core/ElsaOptionsBuilder.cs b/src/core/Elsa.Core/ElsaOptionsBuilder.cs index 9c262c3c5..ef4087677 100644 --- a/src/core/Elsa.Core/ElsaOptionsBuilder.cs +++ b/src/core/Elsa.Core/ElsaOptionsBuilder.cs @@ -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(); services.AddSingleton(); services.AddMemoryCache(); + services.AddSingleton(); 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() 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() => AddMessageType(typeof(T)); + public ElsaOptionsBuilder AddCompetingMessageType() => AddCompetingMessageType(typeof(T)); + + public ElsaOptionsBuilder AddPubSubMessageType(Type messageType) + { + ElsaOptions.PubSubMessageTypes.Add(messageType); + return this; + } + + public ElsaOptionsBuilder AddPubSubMessageType() => AddPubSubMessageType(typeof(T)); public ElsaOptionsBuilder ConfigureDistributedLockProvider(Action configureOptions) { @@ -146,12 +162,6 @@ namespace Elsa return this; } - public ElsaOptionsBuilder UseSignal(Func factory) - { - ElsaOptions.SignalFactory = factory; - return this; - } - public ElsaOptionsBuilder UseStorage(Func factory) => UseStorage(_ => factory()); public ElsaOptionsBuilder UseStorage(Func factory) diff --git a/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs b/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs index 42275dfe7..a24dcb194 100644 --- a/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs +++ b/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs @@ -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>(); } - public static ElsaOptionsBuilder AddConsumer(this ElsaOptionsBuilder elsaOptions) where TConsumer : class, IHandleMessages + /// + /// 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 + /// + /// + /// + /// + /// + public static ElsaOptionsBuilder AddCompetingConsumer(this ElsaOptionsBuilder elsaOptions) where TConsumer : class, IHandleMessages { elsaOptions.Services.AddTransient, TConsumer>(); - elsaOptions.AddMessageType(); + elsaOptions.AddCompetingMessageType(); + return elsaOptions; + } + + public static ElsaOptionsBuilder AddPubSubConsumer(this ElsaOptionsBuilder elsaOptions) where TConsumer : class, IHandleMessages + { + elsaOptions.Services.AddTransient, TConsumer>(); + elsaOptions.AddPubSubMessageType(); return elsaOptions; } @@ -116,6 +130,7 @@ namespace Microsoft.Extensions.DependencyInjection services .AddLogging() .AddLocalization() + .AddSingleton() .AddSingleton() .AddScoped() .AddSingleton() @@ -202,11 +217,11 @@ namespace Microsoft.Extensions.DependencyInjection .AddSingleton(); options - .AddConsumer() - .AddConsumer() - .AddConsumer() - .AddConsumer() - .AddConsumer(); + .AddCompetingConsumer() + .AddCompetingConsumer() + .AddCompetingConsumer() + .AddCompetingConsumer() + .AddCompetingConsumer(); // AutoMapper. services diff --git a/src/core/Elsa.Core/Services/OptionsContainerNameAccessor.cs b/src/core/Elsa.Core/Services/OptionsContainerNameAccessor.cs new file mode 100644 index 000000000..707b288d8 --- /dev/null +++ b/src/core/Elsa.Core/Services/OptionsContainerNameAccessor.cs @@ -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; + } +} \ No newline at end of file diff --git a/src/core/Elsa.Core/Services/ServiceBusFactory.cs b/src/core/Elsa.Core/Services/ServiceBusFactory.cs index 3af285955..d8e3b64ce 100644 --- a/src/core/Elsa.Core/Services/ServiceBusFactory.cs +++ b/src/core/Elsa.Core/Services/ServiceBusFactory.cs @@ -30,19 +30,21 @@ namespace Elsa.Services _handlerActivator = new DependencyInjectionHandlerActivator(serviceProvider); } - public async Task GetServiceBusAsync(Type messageType, CancellationToken cancellationToken) + public Task GetServiceBusAsync(Type messageType, CancellationToken cancellationToken = default) => GetServiceBusAsync(messageType, messageType.Name, cancellationToken); + + public async Task 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 { [messageType] = queueName }; - var configureContext = new ServiceBusEndpointConfigurationContext(configurer, queueName, map, _serviceProvider); + var map = new Dictionary { [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; } diff --git a/src/core/Elsa.Core/StartupTasks/CreateSubscriptions.cs b/src/core/Elsa.Core/StartupTasks/CreateSubscriptions.cs index 0be2bd24e..fbfff8537 100644 --- a/src/core/Elsa.Core/StartupTasks/CreateSubscriptions.cs +++ b/src/core/Elsa.Core/StartupTasks/CreateSubscriptions.cs @@ -9,23 +9,35 @@ namespace Elsa.StartupTasks public class CreateSubscriptions : IStartupTask { private readonly IServiceBusFactory _serviceBusFactory; - private readonly IEnumerable _messageTypes; + private readonly IContainerNameAccessor _containerNameAccessor; + private readonly IEnumerable _competingMessageTypes; + private readonly IEnumerable _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); + } } } } \ No newline at end of file diff --git a/src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj b/src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj index 091ecd4f3..51e2e0734 100644 --- a/src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj +++ b/src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj @@ -13,6 +13,7 @@ + @@ -20,6 +21,7 @@ + diff --git a/src/samples/server/Elsa.Samples.Server.Host/Startup.cs b/src/samples/server/Elsa.Samples.Server.Host/Startup.cs index 47f985e7e..6cd667a78 100644 --- a/src/samples/server/Elsa.Samples.Server.Host/Startup.cs +++ b/src/samples/server/Elsa.Samples.Server.Host/Startup.cs @@ -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() .AddRuntimeSelectItemsProvider() .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) diff --git a/src/samples/server/Elsa.Samples.Server.Host/appsettings.json b/src/samples/server/Elsa.Samples.Server.Host/appsettings.json index c476d9df8..6e20eb179 100644 --- a/src/samples/server/Elsa.Samples.Server.Host/appsettings.json +++ b/src/samples/server/Elsa.Samples.Server.Host/appsettings.json @@ -8,6 +8,9 @@ } }, "AllowedHosts": "*", + "ConnectionStrings": { + "RabbitMq": "amqp://localhost:5672" + }, "Elsa": { "Http": { "BaseUrl": "https://localhost:11000", diff --git a/src/servicebus/Elsa.Rebus.RabbitMq/Extensions/ElsaOptionsExtensions.cs b/src/servicebus/Elsa.Rebus.RabbitMq/Extensions/ElsaOptionsExtensions.cs index 713bf45bd..4c4a98ff7 100644 --- a/src/servicebus/Elsa.Rebus.RabbitMq/Extensions/ElsaOptionsExtensions.cs +++ b/src/servicebus/Elsa.Rebus.RabbitMq/Extensions/ElsaOptionsExtensions.cs @@ -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)); } } \ No newline at end of file