diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs index 6aaf3759b..452b2f3b2 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs @@ -1,110 +1,27 @@ -using System; -using System.Diagnostics; -using System.Threading; -using System.Threading.Tasks; -using Elsa.Activities.AzureServiceBus.Bookmarks; -using Elsa.Activities.AzureServiceBus.Models; +using Elsa.Activities.AzureServiceBus.Bookmarks; using Elsa.Activities.AzureServiceBus.Options; using Elsa.Bookmarks; using Elsa.Dispatch; -using Elsa.DistributedLock; -using Elsa.DistributedLocking; -using Elsa.Models; -using Elsa.Persistence; -using Elsa.Persistence.Specifications; -using Elsa.Services; -using Elsa.Services.Models; -using Elsa.Triggers; using Microsoft.Azure.ServiceBus; using Microsoft.Azure.ServiceBus.Core; -using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; -using Open.Linq.AsyncExtensions; namespace Elsa.Activities.AzureServiceBus.Services { - public class QueueWorker : IAsyncDisposable + public class QueueWorker : WorkerBase { - // TODO: Figure out how to start jobs across multiple tenants / how to get a list of all tenants. - private const string TenantId = default; - - private readonly IMessageReceiver _messageReceiver; - private readonly ICorrelatingWorkflowDispatcher _workflowDispatcher; - private readonly ILogger _logger; - public QueueWorker( - IMessageReceiver messageReceiver, + IReceiverClient messageReceiver, ICorrelatingWorkflowDispatcher workflowDispatcher, - IServiceScopeFactory serviceScopeFactory, IOptions options, - ILogger logger) + ILogger logger) : base(messageReceiver, workflowDispatcher, options, logger) { - _messageReceiver = messageReceiver; - _workflowDispatcher = workflowDispatcher; - _logger = logger; - - _messageReceiver.RegisterMessageHandler(OnMessageReceived, new MessageHandlerOptions(ExceptionReceivedHandler) - { - AutoComplete = false, - MaxConcurrentCalls = options.Value.MaxConcurrentCalls - }); } - public async ValueTask DisposeAsync() => await _messageReceiver.CloseAsync(); + protected override string ActivityType => nameof(AzureServiceBusQueueMessageReceived); - private async Task OnMessageReceived(Message message, CancellationToken cancellationToken) - { - _logger.LogDebug("Message received with ID {MessageId}", message.MessageId); - await TriggerWorkflowsAsync(message, cancellationToken); - await _messageReceiver.CompleteAsync(message.SystemProperties.LockToken); - } - - private async Task TriggerWorkflowsAsync(Message message, CancellationToken cancellationToken) - { - var queueName = _messageReceiver.Path; - var correlationId = message.CorrelationId; - - var model = new MessageModel - { - Body = message.Body, - CorrelationId = message.CorrelationId, - ContentType = message.ContentType, - Label = message.Label, - To = message.To, - MessageId = message.MessageId, - PartitionKey = message.PartitionKey, - ViaPartitionKey = message.ViaPartitionKey, - ReplyTo = message.ReplyTo, - SessionId = message.SessionId, - ExpiresAtUtc = message.ExpiresAtUtc, - TimeToLive = message.TimeToLive, - ReplyToSessionId = message.ReplyToSessionId, - ScheduledEnqueueTimeUtc = message.ScheduledEnqueueTimeUtc - }; - - var bookmark = new QueueMessageReceivedBookmark(queueName, correlationId); - var trigger = new QueueMessageReceivedBookmark(queueName); - var activityType = nameof(AzureServiceBusQueueMessageReceived); - await _workflowDispatcher.DispatchAsync(new ExecuteCorrelatedWorkflowRequest(correlationId, bookmark, trigger, activityType, model, TenantId: TenantId), cancellationToken); - } - - private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e) - { - switch (e.Exception) - { - case MessageLockLostException: - _logger.LogDebug(e.Exception, "Message lock lost"); - break; - case ServiceBusCommunicationException: - _logger.LogDebug(e.Exception, "Lost service bus communication"); - break; - default: - _logger.LogError(e.Exception, "Unhandled exception"); - break; - } - - return Task.CompletedTask; - } + protected override IBookmark CreateBookmark(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path, message.CorrelationId); + protected override IBookmark CreateTrigger(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path); } } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs index 58b433474..23044a8be 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs @@ -1,167 +1,43 @@ -using System; -using System.Diagnostics; -using System.Threading; -using System.Threading.Tasks; using Elsa.Activities.AzureServiceBus.Bookmarks; -using Elsa.Activities.AzureServiceBus.Models; using Elsa.Activities.AzureServiceBus.Options; using Elsa.Bookmarks; -using Elsa.DistributedLock; -using Elsa.DistributedLocking; -using Elsa.Models; -using Elsa.Persistence; -using Elsa.Persistence.Specifications; -using Elsa.Services; -using Elsa.Triggers; +using Elsa.Dispatch; using Microsoft.Azure.ServiceBus; using Microsoft.Azure.ServiceBus.Core; -using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; -using Open.Linq.AsyncExtensions; namespace Elsa.Activities.AzureServiceBus.Services { - public class TopicWorker : IAsyncDisposable + public class TopicWorker : WorkerBase { - // TODO: Figure out how to start jobs across multiple tenants / how to get a list of all tenants. - private const string TenantId = default; - - private readonly IReceiverClient _messageReceiver; - private readonly IServiceScopeFactory _serviceScopeFactory; - private readonly IDistributedLockProvider _distributedLockProvider; - private readonly ILogger _logger; - public TopicWorker( - IReceiverClient messageReceiver, - IServiceScopeFactory serviceScopeFactory, - IDistributedLockProvider distributedLockProvider, + IReceiverClient receiverClient, + ICorrelatingWorkflowDispatcher correlatingWorkflowDispatcher, IOptions options, - ILogger logger) + ILogger logger) : base(receiverClient, correlatingWorkflowDispatcher, options, logger) { - _messageReceiver = messageReceiver; - _serviceScopeFactory = serviceScopeFactory; - _distributedLockProvider = distributedLockProvider; - _logger = logger; - - _messageReceiver.RegisterMessageHandler(OnMessageReceived, new MessageHandlerOptions(ExceptionReceivedHandler) - { - AutoComplete = false, - MaxConcurrentCalls = options.Value.MaxConcurrentCalls - }); } - public async ValueTask DisposeAsync() => await _messageReceiver.CloseAsync(); + protected override string ActivityType => nameof(AzureServiceBusTopicMessageReceived); - private async Task OnMessageReceived(Message message, CancellationToken cancellationToken) + protected override IBookmark CreateBookmark(Message message) { - _logger.LogDebug("Message received with ID {MessageId}", message.MessageId); - await TriggerWorkflowsAsync(message, cancellationToken); - await _messageReceiver.CompleteAsync(message.SystemProperties.LockToken); + GetTopicAndSubscription(out var topicName, out var subscriptionName); + return new TopicMessageReceivedBookmark(topicName, subscriptionName, message.CorrelationId); } - private async Task TriggerWorkflowsAsync(Message message, CancellationToken cancellationToken) + protected override IBookmark CreateTrigger(Message message) { - using var scope = _serviceScopeFactory.CreateScope(); - var workflowQueue = scope.ServiceProvider.GetRequiredService(); - var topicName = _messageReceiver.Path.Split('/')[0]; - var subscriptionName = _messageReceiver.Path.Split('/')[2]; - var correlationId = message.CorrelationId; - var triggerFinder = scope.ServiceProvider.GetRequiredService(); - - var model = new MessageModel - { - Body = message.Body, - CorrelationId = message.CorrelationId, - ContentType = message.ContentType, - Label = message.Label, - To = message.To, - MessageId = message.MessageId, - PartitionKey = message.PartitionKey, - ViaPartitionKey = message.ViaPartitionKey, - ReplyTo = message.ReplyTo, - SessionId = message.SessionId, - ExpiresAtUtc = message.ExpiresAtUtc, - TimeToLive = message.TimeToLive, - ReplyToSessionId = message.ReplyToSessionId, - ScheduledEnqueueTimeUtc = message.ScheduledEnqueueTimeUtc - }; - - async Task TriggerNewWorkflowAsync() - { - var bookmark = new TopicMessageReceivedBookmark(topicName, subscriptionName); - var triggers = await triggerFinder.FindTriggersAsync(bookmark, TenantId, cancellationToken); - - foreach (var trigger in triggers) - { - var workflowBlueprint = trigger.WorkflowBlueprint; - await workflowQueue.EnqueueWorkflowDefinition(workflowBlueprint.Id, workflowBlueprint.TenantId, trigger.ActivityId, model, correlationId, null, cancellationToken); - } - } - - if (string.IsNullOrWhiteSpace(correlationId)) - { - await TriggerNewWorkflowAsync(); - return; - } - - var lockKey = $"azure-service-bus:{topicName}:{subscriptionName}:correlation-{correlationId}"; - var stopwatch = new Stopwatch(); - - _logger.LogDebug("Acquiring lock {LockKey}", lockKey); - stopwatch.Start(); - - if (!await _distributedLockProvider.AcquireLockAsync(lockKey, cancellationToken)) - { - _logger.LogDebug("Lock {LockKey} already taken", lockKey); - return; - } - - try - { - var bookmarkFinder = scope.ServiceProvider.GetRequiredService(); - var workflowInstanceStore = scope.ServiceProvider.GetRequiredService(); - var correlatedWorkflowInstanceCount = await workflowInstanceStore.CountAsync(new CorrelationIdSpecification(model.CorrelationId), cancellationToken); - - if (correlatedWorkflowInstanceCount > 0) - { - // Trigger existing workflows (if blocked on this message). - _logger.LogDebug("{WorkflowInstanceCount} existing workflows found with correlation ID '{CorrelationId}'. Resuming them", correlatedWorkflowInstanceCount, correlationId); - var bookmark = new TopicMessageReceivedBookmark(topicName, subscriptionName, correlationId); - var existingWorkflows = await bookmarkFinder.FindBookmarksAsync(bookmark, TenantId, cancellationToken).ToList(); - await workflowQueue.EnqueueWorkflowsAsync(existingWorkflows, model, model.CorrelationId, cancellationToken: cancellationToken); - } - else - { - // Trigger new workflow. - _logger.LogDebug("No existing workflows found with correlation ID '{CorrelationId}'. Starting new workflow", correlationId); - await TriggerNewWorkflowAsync(); - } - } - finally - { - await _distributedLockProvider.ReleaseLockAsync(lockKey, cancellationToken); - stopwatch.Stop(); - _logger.LogDebug("Lock held for {ElapseTime}", stopwatch.Elapsed); - } + GetTopicAndSubscription(out var topicName, out var subscriptionName); + return new TopicMessageReceivedBookmark(topicName, subscriptionName); } - private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e) + private void GetTopicAndSubscription(out string topicName, out string subscriptionName) { - switch (e.Exception) - { - case MessageLockLostException: - _logger.LogDebug(e.Exception, "Message lock lost"); - break; - case ServiceBusCommunicationException: - _logger.LogDebug(e.Exception, "Lost service bus communication"); - break; - default: - _logger.LogError(e.Exception, "Unhandled exception"); - break; - } - - return Task.CompletedTask; + var segments = ReceiverClient.Path.Split('/'); + topicName = segments[0]; + subscriptionName = segments[2]; } } } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/WorkerBase.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/WorkerBase.cs new file mode 100644 index 000000000..d2029837b --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/WorkerBase.cs @@ -0,0 +1,101 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Elsa.Activities.AzureServiceBus.Models; +using Elsa.Activities.AzureServiceBus.Options; +using Elsa.Bookmarks; +using Elsa.Dispatch; +using Microsoft.Azure.ServiceBus; +using Microsoft.Azure.ServiceBus.Core; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace Elsa.Activities.AzureServiceBus.Services +{ + public abstract class WorkerBase : IAsyncDisposable + { + // TODO: Design multi-tenancy. + private const string TenantId = default; + + private readonly ICorrelatingWorkflowDispatcher _workflowDispatcher; + private readonly ILogger _logger; + + protected WorkerBase( + IReceiverClient receiverClient, + ICorrelatingWorkflowDispatcher workflowDispatcher, + IOptions options, + ILogger logger) + { + ReceiverClient = receiverClient; + _workflowDispatcher = workflowDispatcher; + _logger = logger; + + ReceiverClient.RegisterMessageHandler(OnMessageReceived, new MessageHandlerOptions(ExceptionReceivedHandler) + { + AutoComplete = false, + MaxConcurrentCalls = options.Value.MaxConcurrentCalls + }); + } + + protected IReceiverClient ReceiverClient { get; } + protected abstract string ActivityType { get; } + + public async ValueTask DisposeAsync() => await ReceiverClient.CloseAsync(); + + protected abstract IBookmark CreateBookmark(Message message); + protected abstract IBookmark CreateTrigger(Message message); + + private async Task TriggerWorkflowsAsync(Message message, CancellationToken cancellationToken) + { + var correlationId = message.CorrelationId; + + var model = new MessageModel + { + Body = message.Body, + CorrelationId = message.CorrelationId, + ContentType = message.ContentType, + Label = message.Label, + To = message.To, + MessageId = message.MessageId, + PartitionKey = message.PartitionKey, + ViaPartitionKey = message.ViaPartitionKey, + ReplyTo = message.ReplyTo, + SessionId = message.SessionId, + ExpiresAtUtc = message.ExpiresAtUtc, + TimeToLive = message.TimeToLive, + ReplyToSessionId = message.ReplyToSessionId, + ScheduledEnqueueTimeUtc = message.ScheduledEnqueueTimeUtc + }; + + var queueName = ReceiverClient.Path; + var bookmark = CreateBookmark(message); + var trigger = CreateTrigger(message); + await _workflowDispatcher.DispatchAsync(new ExecuteCorrelatedWorkflowRequest(correlationId, bookmark, trigger, ActivityType, model, TenantId: TenantId), cancellationToken); + } + + private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e) + { + switch (e.Exception) + { + case MessageLockLostException: + _logger.LogDebug(e.Exception, "Message lock lost"); + break; + case ServiceBusCommunicationException: + _logger.LogDebug(e.Exception, "Lost service bus communication"); + break; + default: + _logger.LogError(e.Exception, "Unhandled exception"); + break; + } + + return Task.CompletedTask; + } + + private async Task OnMessageReceived(Message message, CancellationToken cancellationToken) + { + _logger.LogDebug("Message received with ID {MessageId}", message.MessageId); + await TriggerWorkflowsAsync(message, cancellationToken); + await ReceiverClient.CompleteAsync(message.SystemProperties.LockToken); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Http/Middleware/HttpRequestMiddleware.cs b/src/activities/Elsa.Activities.Http/Middleware/HttpRequestMiddleware.cs index 396fdf780..b08b9de0c 100644 --- a/src/activities/Elsa.Activities.Http/Middleware/HttpRequestMiddleware.cs +++ b/src/activities/Elsa.Activities.Http/Middleware/HttpRequestMiddleware.cs @@ -17,7 +17,7 @@ namespace Elsa.Activities.Http.Middleware { public class HttpEndpointMiddleware { - // TODO: Figure out how to start jobs across multiple tenants / how to get a list of all tenants. + // TODO: Design multi-tenancy. private const string TenantId = default; private readonly RequestDelegate _next; diff --git a/src/activities/Elsa.Activities.Temporal.Hangfire/Jobs/RunHangfireWorkflowJob.cs b/src/activities/Elsa.Activities.Temporal.Hangfire/Jobs/RunHangfireWorkflowJob.cs index 3855108d5..68f450e80 100644 --- a/src/activities/Elsa.Activities.Temporal.Hangfire/Jobs/RunHangfireWorkflowJob.cs +++ b/src/activities/Elsa.Activities.Temporal.Hangfire/Jobs/RunHangfireWorkflowJob.cs @@ -1,5 +1,6 @@ using System.Threading.Tasks; using Elsa.Activities.Temporal.Hangfire.Models; +using Elsa.Dispatch; using Elsa.Models; using Elsa.Persistence; using Elsa.Persistence.Specifications; @@ -10,41 +11,21 @@ namespace Elsa.Activities.Temporal.Hangfire.Jobs { public class RunHangfireWorkflowJob { - private readonly IWorkflowRunner _workflowRunner; - private readonly IWorkflowRegistry _workflowRegistry; - private readonly IWorkflowInstanceStore _workflowInstanceStore; - private readonly IWorkflowQueue _workflowQueue; + private readonly IWorkflowDefinitionDispatcher _workflowDefinitionDispatcher; + private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher; - public RunHangfireWorkflowJob(IWorkflowRunner workflowRunner, IWorkflowRegistry workflowRegistry, IWorkflowInstanceStore workflowInstanceStore, IWorkflowQueue workflowQueue) + public RunHangfireWorkflowJob(IWorkflowDefinitionDispatcher workflowDefinitionDispatcher, IWorkflowInstanceDispatcher workflowInstanceDispatcher) { - _workflowRunner = workflowRunner; - _workflowRegistry = workflowRegistry; - _workflowInstanceStore = workflowInstanceStore; - _workflowQueue = workflowQueue; + _workflowDefinitionDispatcher = workflowDefinitionDispatcher; + _workflowInstanceDispatcher = workflowInstanceDispatcher; } public async Task ExecuteAsync(RunHangfireWorkflowJobModel data) { - var workflowBlueprint = (await _workflowRegistry.GetAsync(data.WorkflowDefinitionId, data.TenantId, VersionOptions.Published)); - - if(workflowBlueprint == null) - return; - if (data.WorkflowInstanceId == null) - { - if (workflowBlueprint.IsSingleton == false || await GetWorkflowIsAlreadyExecutingAsync(data.TenantId, data.WorkflowDefinitionId) == false) - await _workflowRunner.RunWorkflowAsync(workflowBlueprint, data.ActivityId); - } + await _workflowDefinitionDispatcher.DispatchAsync(new ExecuteWorkflowDefinitionRequest(data.WorkflowDefinitionId!, data.ActivityId, TenantId: data.TenantId)); else - { - await _workflowQueue.EnqueueWorkflowInstance(data.WorkflowInstanceId, data.ActivityId, default); - } - } - - private async Task GetWorkflowIsAlreadyExecutingAsync(string? tenantId, string workflowDefinitionId) - { - var specification = new TenantSpecification(tenantId).WithWorkflowDefinition(workflowDefinitionId).And(new WorkflowIsAlreadyExecutingSpecification()); - return await _workflowInstanceStore.FindAsync(specification) != null; + await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(data.WorkflowInstanceId, data.ActivityId)); } } } diff --git a/src/core/Elsa.Abstractions/Services/IWorkflowQueue.cs b/src/core/Elsa.Abstractions/Services/IWorkflowQueue.cs deleted file mode 100644 index 4605bc5c9..000000000 --- a/src/core/Elsa.Abstractions/Services/IWorkflowQueue.cs +++ /dev/null @@ -1,48 +0,0 @@ -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using Elsa.Bookmarks; -using Elsa.Builders; - -namespace Elsa.Services -{ - public interface IWorkflowQueue - { - /// - /// Selects workflows and workflow instances based on the specified trigger predicate and enqueues the results for execution. - /// - Task EnqueueWorkflowsAsync( - string activityType, - IBookmark bookmark, - string? tenantId, - object? input = default, - string? correlationId = default, - string? contextId = default, - CancellationToken cancellationToken = default); - - /// - /// Enqueues the specified workflows for execution. - /// - Task EnqueueWorkflowsAsync( - IEnumerable results, - object? input = default, - string? correlationId = default, - string? contextId = default, - CancellationToken cancellationToken = default); - - /// - /// Enqueues the specified workflow instance and activity for execution. - /// - Task EnqueueWorkflowInstance(string workflowInstanceId, string activityId, object? input, CancellationToken cancellationToken = default); - - /// - /// Enqueues the specified workflow definition and activity for execution. - /// - Task EnqueueWorkflowDefinition(string? tenantId, string activityId, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken) where T : IWorkflow; - - /// - /// Enqueues the specified workflow definition and activity for execution. - /// - Task EnqueueWorkflowDefinition(string workflowDefinitionId, string? tenantId, string activityId, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken = default); - } -} \ No newline at end of file diff --git a/src/core/Elsa.Core/Consumers/RunWorkflowDefinitionConsumer.cs b/src/core/Elsa.Core/Consumers/RunWorkflowDefinitionConsumer.cs deleted file mode 100644 index 17ed39416..000000000 --- a/src/core/Elsa.Core/Consumers/RunWorkflowDefinitionConsumer.cs +++ /dev/null @@ -1,106 +0,0 @@ -using System; -using System.Threading.Tasks; -using Elsa.DistributedLock; -using Elsa.DistributedLocking; -using Elsa.Messages; -using Elsa.Models; -using Elsa.Persistence; -using Elsa.Persistence.Specifications; -using Elsa.Persistence.Specifications.WorkflowInstances; -using Elsa.Services; -using Elsa.Services.Models; -using Microsoft.Extensions.Logging; -using Rebus.Handlers; - -namespace Elsa.Consumers -{ - [Obsolete] - public class RunWorkflowDefinitionConsumer : IHandleMessages - { - private readonly IWorkflowRunner _workflowRunner; - private readonly IWorkflowRegistry _workflowRegistry; - private readonly IWorkflowInstanceStore _workflowInstanceStore; - private readonly IDistributedLockProvider _distributedLockProvider; - private readonly ILogger _logger; - private IWorkflowFactory _workflowFactory; - - public RunWorkflowDefinitionConsumer( - IWorkflowRunner workflowRunner, - IWorkflowRegistry workflowRegistry, - IWorkflowInstanceStore workflowInstanceStore, - IDistributedLockProvider distributedLockProvider, - IWorkflowFactory workflowFactory, - ILogger logger) - { - _workflowRunner = workflowRunner; - _workflowRegistry = workflowRegistry; - _workflowInstanceStore = workflowInstanceStore; - _distributedLockProvider = distributedLockProvider; - _logger = logger; - _workflowFactory = workflowFactory; - } - - public async Task Handle(RunWorkflowDefinition message) - { - var workflowDefinitionId = message.WorkflowDefinitionId; - var tenantId = message.TenantId; - var workflowBlueprint = await _workflowRegistry.GetAsync(workflowDefinitionId, tenantId, VersionOptions.Published); - - if (!ValidatePreconditions(workflowDefinitionId, workflowBlueprint)) - return; - - var correlationId = message.CorrelationId; - - if (!string.IsNullOrWhiteSpace(message.CorrelationId)) - { - var lockKey = $"{nameof(RunWorkflowDefinitionConsumer)}:workflow-definition-{message.WorkflowDefinitionId}:correlation-{message.CorrelationId}"; - - if (!await _distributedLockProvider.AcquireLockAsync(lockKey)) - { - _logger.LogDebug("Lock {LockKey} already taken", lockKey); - return; - } - - try - { - var correlatedWorkflowInstanceCount = await _workflowInstanceStore.CountAsync(new WorkflowDefinitionIdSpecification(workflowDefinitionId).And(new CorrelationIdSpecification(correlationId))); - - if (correlatedWorkflowInstanceCount > 0) - { - // Do not create a new workflow instance. - _logger.LogWarning("There's already a workflow with correlation ID '{CorrelationId}'", correlationId); - return; - } - - _logger.LogDebug("No existing workflows found with correlation ID '{CorrelationId}'. Starting new workflow", correlationId); - - // Persist workflow immediately before leaving the lock to prevent scenarios where the workflow takes a long time before the first time it gets persisted, while a new message comes in with the same workflow and correlation. - var workflowInstance = await _workflowFactory.InstantiateAsync(workflowBlueprint!, correlationId, message.ContextId); - await _workflowInstanceStore.SaveAsync(workflowInstance); - - // Run the workflow instance. - await _workflowRunner.RunWorkflowAsync(workflowBlueprint!, workflowInstance, message.ActivityId, message.Input); - - return; - } - finally - { - await _distributedLockProvider.ReleaseLockAsync(lockKey); - } - } - - await _workflowRunner.RunWorkflowAsync(workflowBlueprint!, message.ActivityId, message.Input, message.CorrelationId, message.ContextId); - } - - private bool ValidatePreconditions(string? workflowDefinitionId, IWorkflowBlueprint? workflowBlueprint) - { - if (workflowBlueprint == null) - { - _logger.LogError("Could not run workflow with ID {WorkflowDefinitionId} because it does not exist", workflowDefinitionId); - return false; - } - - return true; - } - } -} \ No newline at end of file diff --git a/src/core/Elsa.Core/Consumers/RunWorkflowInstanceConsumer.cs b/src/core/Elsa.Core/Consumers/RunWorkflowInstanceConsumer.cs deleted file mode 100644 index d2538ebd4..000000000 --- a/src/core/Elsa.Core/Consumers/RunWorkflowInstanceConsumer.cs +++ /dev/null @@ -1,102 +0,0 @@ -using System.Diagnostics; -using System.Linq; -using System.Threading.Tasks; -using Elsa.DistributedLock; -using Elsa.DistributedLocking; -using Elsa.Messages; -using Elsa.Models; -using Elsa.Persistence; -using Elsa.Services; -using Microsoft.Extensions.Logging; -using Rebus.Handlers; - -namespace Elsa.Consumers -{ - public class RunWorkflowInstanceConsumer : IHandleMessages - { - private readonly IWorkflowRunner _workflowRunner; - private readonly IWorkflowInstanceStore _workflowInstanceStore; - private readonly IDistributedLockProvider _distributedLockProvider; - private readonly ICommandSender _commandSender; - private readonly ILogger _logger; - private readonly Stopwatch _stopwatch = new(); - - public RunWorkflowInstanceConsumer( - IWorkflowRunner workflowRunner, - IWorkflowInstanceStore workflowInstanceStore, - IDistributedLockProvider distributedLockProvider, - ICommandSender commandSender, - ILogger logger) - { - _workflowRunner = workflowRunner; - _workflowInstanceStore = workflowInstanceStore; - _distributedLockProvider = distributedLockProvider; - _commandSender = commandSender; - _logger = logger; - } - - public async Task Handle(RunWorkflowInstance message) - { - var workflowInstanceId = message.WorkflowInstanceId; - var lockKey = workflowInstanceId; - - _logger.LogDebug("Acquiring lock on workflow instance {WorkflowInstanceId}", workflowInstanceId); - _stopwatch.Restart(); - - if (!await _distributedLockProvider.AcquireLockAsync(lockKey)) - { - _logger.LogDebug("Failed to acquire lock on workflow instance {WorkflowInstanceId}. Re-queueing message", workflowInstanceId); - await _commandSender.SendAsync(message); - return; - } - - try - { - var workflowInstance = await _workflowInstanceStore.FindByIdAsync(message.WorkflowInstanceId); - - if (!ValidatePreconditions(workflowInstanceId, workflowInstance, message.ActivityId)) - return; - - await _workflowRunner.RunWorkflowAsync( - workflowInstance!, - message.ActivityId, - message.Input); - } - finally - { - await _distributedLockProvider.ReleaseLockAsync(lockKey); - _stopwatch.Stop(); - _logger.LogDebug("Held lock on workflow instance {WorkflowInstanceId} for {ElapsedTime}", workflowInstanceId, _stopwatch.Elapsed); - } - } - - private bool ValidatePreconditions(string? workflowInstanceId, WorkflowInstance? workflowInstance, string? activityId) - { - if (workflowInstance == null) - { - _logger.LogWarning("Could not run workflow instance with ID {WorkflowInstanceId} because it does not exist", workflowInstanceId); - return false; - } - - if (workflowInstance.WorkflowStatus != WorkflowStatus.Suspended && workflowInstance.WorkflowStatus != WorkflowStatus.Running) - { - _logger.LogWarning("Could not run workflow instance with ID {WorkflowInstanceId} because it has a status other than Suspended or Running. Its actual status is {WorkflowStatus}", workflowInstanceId, workflowInstance.WorkflowStatus); - return false; - } - - if (activityId != null) - { - var activityIsBlocking = workflowInstance.BlockingActivities.Any(x => x.ActivityId == activityId); - var activityIsScheduled = workflowInstance.ScheduledActivities.Any(x => x.ActivityId == activityId) || workflowInstance.CurrentActivity?.ActivityId == activityId; - - if (!activityIsBlocking && !activityIsScheduled) - { - _logger.LogWarning("Did not run workflow {WorkflowInstanceId} for activity {ActivityId} because the workflow is not blocked on that activity nor is that activity scheduled for execution", workflowInstanceId, activityId); - return false; - } - } - - return true; - } - } -} \ No newline at end of file diff --git a/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowRequestConsumer.cs b/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowInstanceRequestConsumer.cs similarity index 94% rename from src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowRequestConsumer.cs rename to src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowInstanceRequestConsumer.cs index 7dfe83570..ca28967fc 100644 --- a/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowRequestConsumer.cs +++ b/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowInstanceRequestConsumer.cs @@ -10,7 +10,7 @@ using Rebus.Handlers; namespace Elsa.Dispatch.Consumers { - public class ExecuteWorkflowRequestConsumer : IHandleMessages + public class ExecuteWorkflowInstanceRequestConsumer : IHandleMessages { private readonly IWorkflowRunner _workflowRunner; private readonly IWorkflowInstanceStore _workflowInstanceStore; @@ -19,12 +19,12 @@ namespace Elsa.Dispatch.Consumers private readonly ILogger _logger; private readonly Stopwatch _stopwatch = new(); - public ExecuteWorkflowRequestConsumer( + public ExecuteWorkflowInstanceRequestConsumer( IWorkflowRunner workflowRunner, IWorkflowInstanceStore workflowInstanceStore, IDistributedLockProvider distributedLockProvider, ICommandSender commandSender, - ILogger logger) + ILogger logger) { _workflowRunner = workflowRunner; _workflowInstanceStore = workflowInstanceStore; diff --git a/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs b/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs index 6e9095b6f..7d81316ee 100644 --- a/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs +++ b/src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs @@ -7,13 +7,13 @@ using Elsa.ActivityProviders; using Elsa.ActivityTypeProviders; using Elsa.Bookmarks; using Elsa.Builders; -using Elsa.Consumers; using Elsa.Decorators; +using Elsa.Dispatch; +using Elsa.Dispatch.Consumers; using Elsa.Expressions; using Elsa.Handlers; using Elsa.HostedServices; using Elsa.Mapping; -using Elsa.Messages; using Elsa.Metadata; using Elsa.Persistence; using Elsa.Persistence.Decorators; @@ -59,8 +59,6 @@ namespace Microsoft.Extensions.DependencyInjection .AddCoreActivities(); options.AddAutoMapper(); - options.AddConsumer(); - options.AddConsumer(); services.Decorate(); services.Decorate(); @@ -160,15 +158,15 @@ namespace Microsoft.Extensions.DependencyInjection // Service Bus. services - .AddScoped() .AddSingleton() .AddSingleton() .AddSingleton() .AddSingleton(); options - .AddConsumer() - .AddConsumer(); + .AddConsumer() + .AddConsumer() + .AddConsumer(); // AutoMapper. services diff --git a/src/core/Elsa.Core/Services/WorkflowQueue.cs b/src/core/Elsa.Core/Services/WorkflowQueue.cs deleted file mode 100644 index 37cd569c2..000000000 --- a/src/core/Elsa.Core/Services/WorkflowQueue.cs +++ /dev/null @@ -1,57 +0,0 @@ -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using Elsa.Bookmarks; -using Elsa.Builders; -using Elsa.Messages; -using Open.Linq.AsyncExtensions; - -namespace Elsa.Services -{ - public class WorkflowQueue : IWorkflowQueue - { - private readonly IBookmarkFinder _bookmarkFinder; - private readonly ICommandSender _commandSender; - - public WorkflowQueue(IBookmarkFinder bookmarkFinder, ICommandSender commandSender) - { - _bookmarkFinder = bookmarkFinder; - _commandSender = commandSender; - } - - public async Task EnqueueWorkflowsAsync( - string activityType, - IBookmark bookmark, - string? tenantId, - object? input = default, - string? correlationId = default, - string? contextId = default, - CancellationToken cancellationToken = default) - { - var results = await _bookmarkFinder.FindBookmarksAsync(activityType, bookmark, tenantId, cancellationToken).ToList(); - await EnqueueWorkflowsAsync(results, input, correlationId, contextId, cancellationToken); - } - - public async Task EnqueueWorkflowsAsync(IEnumerable results, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken) - { - foreach (var result in results) - await EnqueueWorkflowInstance(result.WorkflowInstanceId, result.ActivityId, input, cancellationToken); - } - - public async Task EnqueueWorkflowInstance(string workflowInstanceId, string activityId, object? input, CancellationToken cancellationToken = default) - { - await _commandSender.SendAsync(new RunWorkflowInstance(workflowInstanceId, activityId, input)); - } - - public async Task EnqueueWorkflowDefinition(string? tenantId, string activityId, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken) where T : IWorkflow - { - var workflowDefinitionId = typeof(T).Name; - await EnqueueWorkflowDefinition(workflowDefinitionId, tenantId, activityId, input, correlationId, contextId, cancellationToken); - } - - public async Task EnqueueWorkflowDefinition(string workflowDefinitionId, string? tenantId, string activityId, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken) - { - await _commandSender.SendAsync(new RunWorkflowDefinition(workflowDefinitionId, tenantId, activityId, input, correlationId, contextId)); - } - } -} \ No newline at end of file diff --git a/src/core/Elsa.Core/Services/WorkflowReviver.cs b/src/core/Elsa.Core/Services/WorkflowReviver.cs index 01b2ee519..f368c9890 100644 --- a/src/core/Elsa.Core/Services/WorkflowReviver.cs +++ b/src/core/Elsa.Core/Services/WorkflowReviver.cs @@ -2,6 +2,7 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; +using Elsa.Dispatch; using Elsa.Exceptions; using Elsa.Models; using Elsa.Persistence; @@ -11,24 +12,25 @@ namespace Elsa.Services public class WorkflowReviver : IWorkflowReviver { private readonly IWorkflowRunner _workflowRunner; - private readonly IWorkflowQueue _workflowQueue; + private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher; private readonly IWorkflowRegistry _workflowRegistry; private readonly IWorkflowInstanceStore _workflowInstanceStore; - private readonly IGetsStartActivitiesForCompositeActivityBlueprint startingActivitiesProvider; + private readonly IGetsStartActivitiesForCompositeActivityBlueprint _startingActivitiesProvider; - public WorkflowReviver(IWorkflowRunner workflowRunner, - IWorkflowQueue workflowQueue, - IWorkflowRegistry workflowRegistry, - IWorkflowInstanceStore workflowInstanceStore, - IGetsStartActivitiesForCompositeActivityBlueprint startingActivitiesProvider) + public WorkflowReviver( + IWorkflowRunner workflowRunner, + IWorkflowInstanceDispatcher workflowInstanceDispatcher, + IWorkflowRegistry workflowRegistry, + IWorkflowInstanceStore workflowInstanceStore, + IGetsStartActivitiesForCompositeActivityBlueprint startingActivitiesProvider) { _workflowRunner = workflowRunner; - _workflowQueue = workflowQueue; + _workflowInstanceDispatcher = workflowInstanceDispatcher; _workflowRegistry = workflowRegistry; _workflowInstanceStore = workflowInstanceStore; - this.startingActivitiesProvider = startingActivitiesProvider ?? throw new ArgumentNullException(nameof(startingActivitiesProvider)); + _startingActivitiesProvider = startingActivitiesProvider ?? throw new ArgumentNullException(nameof(startingActivitiesProvider)); } - + public async Task ReviveAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken) { if (workflowInstance.WorkflowStatus != WorkflowStatus.Faulted) @@ -38,7 +40,7 @@ namespace Elsa.Services if (fault == null) throw new WorkflowException("Cannot revive a workflow with no fault"); - + var faultedActivityId = fault.FaultedActivityId; if (faultedActivityId == null) @@ -66,12 +68,12 @@ namespace Elsa.Services workflowInstance = await ReviveAsync(workflowInstance, cancellationToken); return await _workflowRunner.RunWorkflowAsync(workflowInstance, null, null, cancellationToken); } - + public async Task ReviveAndQueueAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken) { workflowInstance = await ReviveAsync(workflowInstance, cancellationToken); - var currentActivity = await GetActivityToScheduleAsync(workflowInstance, cancellationToken); - await _workflowQueue.EnqueueWorkflowInstance(workflowInstance.Id, currentActivity.ActivityId, currentActivity.Input, cancellationToken); + var currentActivity = await GetActivityToScheduleAsync(workflowInstance, cancellationToken); + await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(workflowInstance.Id, currentActivity.ActivityId, currentActivity.Input), cancellationToken); return workflowInstance; } @@ -86,8 +88,8 @@ namespace Elsa.Services if (workflowBlueprint == null) throw new WorkflowException($"Could not find associated workflow definition {workflowInstance.DefinitionId} with version {workflowInstance.Version}"); - - var startActivity = startingActivitiesProvider.GetStartActivities(workflowBlueprint).FirstOrDefault(); + + var startActivity = _startingActivitiesProvider.GetStartActivities(workflowBlueprint).FirstOrDefault(); if (startActivity == null) throw new WorkflowException($"Cannot revive workflow {workflowInstance.Id} because it has no start activities"); diff --git a/src/core/Elsa.Core/StartupTasks/ContinueRunningWorkflows.cs b/src/core/Elsa.Core/StartupTasks/ContinueRunningWorkflows.cs index b31439f37..139eb0823 100644 --- a/src/core/Elsa.Core/StartupTasks/ContinueRunningWorkflows.cs +++ b/src/core/Elsa.Core/StartupTasks/ContinueRunningWorkflows.cs @@ -1,6 +1,7 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; +using Elsa.Dispatch; using Elsa.DistributedLock; using Elsa.DistributedLocking; using Elsa.Models; @@ -19,18 +20,18 @@ namespace Elsa.StartupTasks public class ContinueRunningWorkflows : IStartupTask { private readonly IWorkflowInstanceStore _workflowInstanceStore; - private readonly IWorkflowQueue _workflowQueue; + private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher; private readonly IDistributedLockProvider _distributedLockProvider; private readonly ILogger _logger; public ContinueRunningWorkflows( IWorkflowInstanceStore workflowInstanceStore, - IWorkflowQueue workflowQueue, + IWorkflowInstanceDispatcher workflowInstanceDispatcher, IDistributedLockProvider distributedLockProvider, ILogger logger) { _workflowInstanceStore = workflowInstanceStore; - _workflowQueue = workflowQueue; + _workflowInstanceDispatcher = workflowInstanceDispatcher; _distributedLockProvider = distributedLockProvider; _logger = logger; } @@ -48,7 +49,7 @@ namespace Elsa.StartupTasks { var instances = await _workflowInstanceStore.FindManyAsync(new WorkflowStatusSpecification(WorkflowStatus.Running), cancellationToken: cancellationToken).ToList(); - if(instances.Any()) + if (instances.Any()) _logger.LogInformation("Found {WorkflowInstanceCount} workflows with status 'Running'. Resuming each one of them", instances.Count); else _logger.LogInformation("Found no workflows with status 'Running'. Nothing to resume"); @@ -62,7 +63,9 @@ namespace Elsa.StartupTasks { if (instance.BlockingActivities.Any()) { - _logger.LogWarning("Workflow '{WorkflowInstanceId}' was in the Running state, but has no scheduled activities not has a currently executing one. However, it does have blocking activities, so switching to Suspended status", instance.Id); + _logger.LogWarning( + "Workflow '{WorkflowInstanceId}' was in the Running state, but has no scheduled activities not has a currently executing one. However, it does have blocking activities, so switching to Suspended status", + instance.Id); instance.WorkflowStatus = WorkflowStatus.Suspended; await _workflowInstanceStore.SaveAsync(instance, cancellationToken); continue; @@ -74,11 +77,7 @@ namespace Elsa.StartupTasks var scheduledActivity = instance.CurrentActivity ?? instance.ScheduledActivities.Peek(); - await _workflowQueue.EnqueueWorkflowInstance( - instance.Id, - scheduledActivity.ActivityId, - scheduledActivity.Input, - cancellationToken); + await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(instance.Id, scheduledActivity.ActivityId, scheduledActivity.Input), cancellationToken); } } finally