Reuse certain queues for certain event types (#1317)

src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs

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

src/activities/Elsa.Activities.Conductor/Extensions/ElsaOptionsBuilderExtensions.cs

@@ -43,8 +43,8 @@ namespace Elsa.Activities.Conductor.Extensions
 
             elsa
                 .AddActivitiesFrom<SendCommand>()
-                .AddCompetingConsumer<SendCommandConsumer, SendCommandModel>()
-                .AddCompetingConsumer<RunTaskConsumer, RunTaskModel>();
+                .AddCompetingConsumer<SendCommandConsumer, SendCommandModel>("ConductorCommand")
+                .AddCompetingConsumer<RunTaskConsumer, RunTaskModel>("ConductorCommand");
 
             return elsa;
         }

src/core/Elsa.Core/ElsaOptions.cs

@@ -22,19 +22,20 @@ using Storage.Net.Blobs;
 
 namespace Elsa
 {
-    public record CompetingMessageType(Type MessageType, string? Queue = default);
-    
+    public record MessageTypeConfig(Type MessageType, string? QueueName = default);
+
     public class ElsaOptions
     {
         public static string FormatChannelQueueName<TMessage>(string channel) => FormatChannelQueueName(typeof(TMessage), channel);
-        
-        public static string FormatChannelQueueName(Type messageType, string channel)
+        public static string FormatChannelQueueName(Type messageType, string channel) => FormatChannelQueueName(messageType.Name, channel);
+
+        public static string FormatChannelQueueName(string queueName, string channel)
         {
-            var queue = !string.IsNullOrWhiteSpace(channel) ? $"{messageType.Name}{channel}" : messageType.Name;
+            var queue = !string.IsNullOrWhiteSpace(channel) ? $"{queueName}{channel}" : queueName;
             return FormatQueueName(queue);
         }
         
-        public static string FormatQueueName(string queue) => queue.Dehumanize().Underscore().Dasherize();
+        public static string FormatQueueName(string queue) => queue.Humanize().Dehumanize().Underscore().Dasherize();
 
         internal ElsaOptions()
         {
@@ -65,8 +66,8 @@ namespace Elsa
         public IEnumerable<Type> ActivityTypes => ActivityFactory.Types;
 
         public IList<Type> WorkflowTypes { get; } = new List<Type>();
-        public IList<CompetingMessageType> CompetingMessageTypes { get; } = new List<CompetingMessageType>();
-        public IList<Type> PubSubMessageTypes { get; } = new List<Type>();
+        public IList<MessageTypeConfig> CompetingMessageTypes { get; } = new List<MessageTypeConfig>();
+        public IList<MessageTypeConfig> PubSubMessageTypes { get; } = new List<MessageTypeConfig>();
         public ServiceBusOptions ServiceBusOptions { get; } = new();
         public DistributedLockingOptions DistributedLockingOptions { get; set; }
 

src/core/Elsa.Core/ElsaOptionsBuilder.cs

@@ -150,21 +150,21 @@ namespace Elsa
             return this;
         }
 
-        public ElsaOptionsBuilder AddCompetingMessageType(Type messageType, string? queue = default)
+        public ElsaOptionsBuilder AddCompetingMessageType(Type messageType, string? queueName = default)
         {
-            ElsaOptions.CompetingMessageTypes.Add(new CompetingMessageType(messageType, queue));
+            ElsaOptions.CompetingMessageTypes.Add(new MessageTypeConfig(messageType, queueName));
             return this;
         }
 
-        public ElsaOptionsBuilder AddCompetingMessageType<T>(string? queue = default) => AddCompetingMessageType(typeof(T), queue);
+        public ElsaOptionsBuilder AddCompetingMessageType<T>(string? queueName = default) => AddCompetingMessageType(typeof(T), queueName);
 
-        public ElsaOptionsBuilder AddPubSubMessageType(Type messageType)
+        public ElsaOptionsBuilder AddPubSubMessageType(Type messageType, string? queueName = default)
         {
-            ElsaOptions.PubSubMessageTypes.Add(messageType);
+            ElsaOptions.PubSubMessageTypes.Add(new MessageTypeConfig(messageType, queueName));
             return this;
         }
 
-        public ElsaOptionsBuilder AddPubSubMessageType<T>() => AddPubSubMessageType(typeof(T));
+        public ElsaOptionsBuilder AddPubSubMessageType<T>(string? queueName = default) => AddPubSubMessageType(typeof(T), queueName);
 
         public ElsaOptionsBuilder ConfigureDistributedLockProvider(Action<DistributedLockingOptionsBuilder> configureOptions)
         {

src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs

@@ -109,14 +109,10 @@ namespace Microsoft.Extensions.DependencyInjection
         /// 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>
+        public static ElsaOptionsBuilder AddCompetingConsumer<TConsumer, TMessage>(this ElsaOptionsBuilder elsaOptions, string? queueName = default) where TConsumer : class, IHandleMessages<TMessage>
         {
             elsaOptions.AddCompetingConsumerService<TConsumer, TMessage>();
-            elsaOptions.AddCompetingMessageType<TMessage>();
+            elsaOptions.AddCompetingMessageType<TMessage>(queueName);
             return elsaOptions;
         }
 
@@ -126,10 +122,10 @@ namespace Microsoft.Extensions.DependencyInjection
             return elsaOptions;
         }
         
-        public static ElsaOptionsBuilder AddPubSubConsumer<TConsumer, TMessage>(this ElsaOptionsBuilder elsaOptions) where TConsumer : class, IHandleMessages<TMessage>
+        public static ElsaOptionsBuilder AddPubSubConsumer<TConsumer, TMessage>(this ElsaOptionsBuilder elsaOptions, string? queueName = default) where TConsumer : class, IHandleMessages<TMessage>
         {
             elsaOptions.Services.AddTransient<IHandleMessages<TMessage>, TConsumer>();
-            elsaOptions.AddPubSubMessageType<TMessage>();
+            elsaOptions.AddPubSubMessageType<TMessage>(queueName);
             return elsaOptions;
         }
 
@@ -249,12 +245,12 @@ namespace Microsoft.Extensions.DependencyInjection
                 .AddSingleton<IEventPublisher, EventPublisher>();
 
             options
-                .AddCompetingConsumer<TriggerWorkflowsRequestConsumer, TriggerWorkflowsRequest>()
-                .AddCompetingConsumerService<ExecuteWorkflowDefinitionRequestConsumer, ExecuteWorkflowDefinitionRequest>()
-                .AddCompetingConsumerService<ExecuteWorkflowInstanceRequestConsumer, ExecuteWorkflowInstanceRequest>()
-                .AddPubSubConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionPublished>()
-                .AddPubSubConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionRetracted>()
-                .AddPubSubConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionDeleted>();
+                .AddCompetingConsumer<TriggerWorkflowsRequestConsumer, TriggerWorkflowsRequest>("ExecuteWorkflow")
+                .AddCompetingConsumer<ExecuteWorkflowDefinitionRequestConsumer, ExecuteWorkflowDefinitionRequest>("ExecuteWorkflow")
+                .AddCompetingConsumer<ExecuteWorkflowInstanceRequestConsumer, ExecuteWorkflowInstanceRequest>("ExecuteWorkflow")
+                .AddPubSubConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionPublished>("WorkflowDefinitionEvents")
+                .AddPubSubConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionRetracted>("WorkflowDefinitionEvents")
+                .AddPubSubConsumer<UpdateWorkflowTriggersIndexConsumer, WorkflowDefinitionDeleted>("WorkflowDefinitionEvents");
 
             // AutoMapper.
             services

src/core/Elsa.Core/Services/Messaging/ServiceBusFactory.cs

@@ -32,8 +32,9 @@ namespace Elsa.Services.Messaging
 
         public async Task<IBus> GetServiceBusAsync(Type messageType, string? queueName = default, CancellationToken cancellationToken = default)
         {
-            if (string.IsNullOrWhiteSpace(queueName))
-                queueName = ElsaOptions.FormatChannelQueueName(messageType, _elsaOptions.WorkflowChannelOptions.Default);
+            queueName = string.IsNullOrWhiteSpace(queueName) 
+                ? ElsaOptions.FormatChannelQueueName(messageType, _elsaOptions.WorkflowChannelOptions.Default) 
+                : ElsaOptions.FormatQueueName(queueName);
             
             var prefixedQueueName = PrefixQueueName(queueName);
             await _semaphore.WaitAsync(cancellationToken);

src/core/Elsa.Core/StartupTasks/CreateSubscriptions.cs

@@ -10,14 +10,18 @@ namespace Elsa.StartupTasks
     public class CreateSubscriptions : IStartupTask
     {
         private readonly IServiceBusFactory _serviceBusFactory;
+        private readonly ElsaOptions _elsaOptions;
         private readonly IContainerNameAccessor _containerNameAccessor;
-        private readonly IList<CompetingMessageType> _competingMessageTypes;
-        private readonly IEnumerable<Type> _pubSubMessageTypes;
+        private readonly IDistributedLockProvider _distributedLockProvider;
+        private readonly IList<MessageTypeConfig> _competingMessageTypes;
+        private readonly IEnumerable<MessageTypeConfig> _pubSubMessageTypes;
 
-        public CreateSubscriptions(IServiceBusFactory serviceBusFactory, ElsaOptions elsaOptions, IContainerNameAccessor containerNameAccessor)
+        public CreateSubscriptions(IServiceBusFactory serviceBusFactory, ElsaOptions elsaOptions, IContainerNameAccessor containerNameAccessor, IDistributedLockProvider distributedLockProvider)
         {
             _serviceBusFactory = serviceBusFactory;
+            _elsaOptions = elsaOptions;
             _containerNameAccessor = containerNameAccessor;
+            _distributedLockProvider = distributedLockProvider;
             _competingMessageTypes = elsaOptions.CompetingMessageTypes.ToList();
             _pubSubMessageTypes = elsaOptions.PubSubMessageTypes;
 
@@ -27,8 +31,8 @@ namespace Elsa.StartupTasks
             // For each workflow channel, register a competing message type for workflow definition and workflow instance consumers.
             foreach (var workflowChannel in workflowChannels)
             {
-                _competingMessageTypes.Add(new CompetingMessageType(typeof(ExecuteWorkflowDefinitionRequest), ElsaOptions.FormatChannelQueueName<ExecuteWorkflowDefinitionRequest>(workflowChannel)));
-                _competingMessageTypes.Add(new CompetingMessageType(typeof(ExecuteWorkflowInstanceRequest), ElsaOptions.FormatChannelQueueName<ExecuteWorkflowInstanceRequest>(workflowChannel)));
+                _competingMessageTypes.Add(new MessageTypeConfig(typeof(ExecuteWorkflowDefinitionRequest), ElsaOptions.FormatChannelQueueName("ExecuteWorkflow", workflowChannel)));
+                _competingMessageTypes.Add(new MessageTypeConfig(typeof(ExecuteWorkflowInstanceRequest), ElsaOptions.FormatChannelQueueName("ExecuteWorkflow", workflowChannel)));
             }
         }
 
@@ -36,18 +40,23 @@ namespace Elsa.StartupTasks
 
         public async Task ExecuteAsync(CancellationToken cancellationToken)
         {
+            await using var handle = await _distributedLockProvider.AcquireLockAsync(nameof(CreateSubscriptions), _elsaOptions.DistributedLockTimeout, cancellationToken);
+
+            if (handle == null)
+                throw new Exception("Could not acquire a lock within the maximum amount of time configured");
+            
             foreach (var messageType in _competingMessageTypes)
             {
-                var bus = await _serviceBusFactory.GetServiceBusAsync(messageType.MessageType, messageType.Queue, cancellationToken);
+                var bus = await _serviceBusFactory.GetServiceBusAsync(messageType.MessageType, messageType.QueueName, cancellationToken);
                 await bus.Subscribe(messageType.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);
+                var queueName = $"{containerName}:{messageType.QueueName ?? messageType.MessageType.Name}";
+                var bus = await _serviceBusFactory.GetServiceBusAsync(messageType.MessageType, queueName, cancellationToken);
+                await bus.Subscribe(messageType.MessageType);
             }
         }
     }