diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Bookmarks/TopicMessageReceivedBookmark.cs b/src/activities/Elsa.Activities.AzureServiceBus/Bookmarks/TopicMessageReceivedBookmark.cs index 087a46127..deaa258ff 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Bookmarks/TopicMessageReceivedBookmark.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Bookmarks/TopicMessageReceivedBookmark.cs @@ -11,16 +11,14 @@ namespace Elsa.Activities.AzureServiceBus.Bookmarks { } - public TopicMessageReceivedBookmark(string topicName, string subscriptionName, string? correlationId = default) + public TopicMessageReceivedBookmark(string topicName, string subscriptionName) { TopicName = topicName; SubscriptionName = subscriptionName; - CorrelationId = correlationId; } - public string TopicName { get; set; } = default!; + public string TopicName { get; set; } = default!; public string SubscriptionName { get; set; } = default!; - public string? CorrelationId { get; set; } } public class TopicMessageReceivedBookmarkProvider : BookmarkProvider @@ -32,7 +30,6 @@ namespace Elsa.Activities.AzureServiceBus.Bookmarks { TopicName = (await context.ReadActivityPropertyAsync(x => x.TopicName, cancellationToken))!, SubscriptionName = (await context.ReadActivityPropertyAsync(x => x.SubscriptionName, cancellationToken))!, - CorrelationId = context.ActivityExecutionContext.WorkflowExecutionContext.CorrelationId }) }; } diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs index ff75f7c33..8dd0b7deb 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs @@ -6,6 +6,7 @@ using Elsa.Activities.AzureServiceBus.Services; using Elsa.Activities.AzureServiceBus.StartupTasks; using Elsa.Events; using Elsa.Runtime; +using Elsa.Services; using Microsoft.Azure.ServiceBus; using Microsoft.Azure.ServiceBus.Management; using Microsoft.Extensions.DependencyInjection; @@ -32,6 +33,7 @@ namespace Elsa.Activities.AzureServiceBus.Extensions .AddSingleton(sp => sp.GetRequiredService()) .AddSingleton() .AddSingleton() + .AddSingleton>() .AddStartupTask() .AddStartupTask() .AddBookmarkProvider() diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs index 686c5fc40..c0b704c26 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs @@ -4,6 +4,7 @@ using Elsa.Activities.AzureServiceBus.Bookmarks; using Elsa.Activities.AzureServiceBus.Options; using Elsa.Bookmarks; using Elsa.Dispatch; +using Elsa.Services; using Microsoft.Azure.ServiceBus; using Microsoft.Azure.ServiceBus.Core; using Microsoft.Extensions.Logging; @@ -15,10 +16,10 @@ namespace Elsa.Activities.AzureServiceBus.Services { public QueueWorker( IReceiverClient messageReceiver, - IWorkflowDispatcher workflowDispatcher, + Scoped workflowLaunchpad, IOptions options, Func disposeReceiverAction, - ILogger logger) : base(messageReceiver, workflowDispatcher, options, disposeReceiverAction, logger) + ILogger logger) : base(messageReceiver, workflowLaunchpad, options, disposeReceiverAction, logger) { } diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs index fdb890583..055fdfd21 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs @@ -4,6 +4,7 @@ using Elsa.Activities.AzureServiceBus.Bookmarks; using Elsa.Activities.AzureServiceBus.Options; using Elsa.Bookmarks; using Elsa.Dispatch; +using Elsa.Services; using Microsoft.Azure.ServiceBus; using Microsoft.Azure.ServiceBus.Core; using Microsoft.Extensions.Logging; @@ -15,10 +16,10 @@ namespace Elsa.Activities.AzureServiceBus.Services { public TopicWorker( IReceiverClient receiverClient, - IWorkflowDispatcher workflowDispatcher, + Scoped workflowLaunchpad, IOptions options, Func disposeReceiverAction, - ILogger logger) : base(receiverClient, workflowDispatcher, options, disposeReceiverAction, logger) + ILogger logger) : base(receiverClient, workflowLaunchpad, options, disposeReceiverAction, logger) { } @@ -27,7 +28,7 @@ namespace Elsa.Activities.AzureServiceBus.Services protected override IBookmark CreateBookmark(Message message) { GetTopicAndSubscription(out var topicName, out var subscriptionName); - return new TopicMessageReceivedBookmark(topicName, subscriptionName, message.CorrelationId); + return new TopicMessageReceivedBookmark(topicName, subscriptionName); } protected override IBookmark CreateTrigger(Message message) diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/WorkerBase.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/WorkerBase.cs index a2802bf82..4097b2eb3 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Services/WorkerBase.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/WorkerBase.cs @@ -6,6 +6,7 @@ using Elsa.Activities.AzureServiceBus.Models; using Elsa.Activities.AzureServiceBus.Options; using Elsa.Bookmarks; using Elsa.Dispatch; +using Elsa.Services; using Microsoft.Azure.ServiceBus; using Microsoft.Azure.ServiceBus.Core; using Microsoft.Extensions.Logging; @@ -16,21 +17,21 @@ namespace Elsa.Activities.AzureServiceBus.Services public abstract class WorkerBase : IAsyncDisposable { // TODO: Design multi-tenancy. - private const string TenantId = default; + private const string? TenantId = default; - private readonly IWorkflowDispatcher _workflowDispatcher; + private readonly Scoped _workflowLaunchpad; private readonly Func _disposeReceiverAction; private readonly ILogger _logger; protected WorkerBase( IReceiverClient receiverClient, - IWorkflowDispatcher workflowDispatcher, + Scoped workflowLaunchpad, IOptions options, Func disposeReceiverAction, ILogger logger) { ReceiverClient = receiverClient; - _workflowDispatcher = workflowDispatcher; + _workflowLaunchpad = workflowLaunchpad; _disposeReceiverAction = disposeReceiverAction; _logger = logger; @@ -73,7 +74,9 @@ namespace Elsa.Activities.AzureServiceBus.Services var bookmark = CreateBookmark(message); var trigger = CreateTrigger(message); - await _workflowDispatcher.DispatchAsync(new TriggerWorkflowsRequest(ActivityType, bookmark, trigger, model, correlationId, TenantId: TenantId), cancellationToken); + var launchContext = new CollectWorkflowsContext(ActivityType, bookmark, trigger, correlationId); + + await _workflowLaunchpad.UseServiceAsync(service => service.CollectAndDispatchWorkflowsAsync(launchContext, model, cancellationToken)); } private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e) diff --git a/src/core/Elsa.Core/Builders/InlineActivityBuilderExtensions.cs b/src/core/Elsa.Core/Builders/InlineActivityBuilderExtensions.cs index 8ddf6b055..8ec680bee 100644 --- a/src/core/Elsa.Core/Builders/InlineActivityBuilderExtensions.cs +++ b/src/core/Elsa.Core/Builders/InlineActivityBuilderExtensions.cs @@ -27,6 +27,9 @@ namespace Elsa.Builders public static IActivityBuilder Then(this IBuilder builder, Action activity) => builder.Then(inline => inline.Set(x => x.Function, RunInline(activity))); + + public static IActivityBuilder Then(this IBuilder builder, Func activity) => + builder.Then(inline => inline.Set(x => x.Function, RunInline(activity))); private static Func> RunInline( Func> activity) => @@ -60,7 +63,14 @@ namespace Elsa.Builders context => { activity(); - return new ValueTask(new OutcomeResult()); + return new ValueTask(new DoneResult()); + }; + + private static Func> RunInline(Func activity) => + async context => + { + await activity(); + return new DoneResult(); }; } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Services/WorkflowLaunchpad.cs b/src/core/Elsa.Core/Services/WorkflowLaunchpad.cs index 90a7bff60..e99d1e559 100644 --- a/src/core/Elsa.Core/Services/WorkflowLaunchpad.cs +++ b/src/core/Elsa.Core/Services/WorkflowLaunchpad.cs @@ -1,4 +1,5 @@ -using System.Collections.Generic; +using System; +using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -110,59 +111,41 @@ namespace Elsa.Services if (!ValidatePreconditions(workflowDefinitionId, workflowBlueprint)) return null; - var correlationLockHandle = default(IDistributedSynchronizationHandle?); + correlationId ??= Guid.NewGuid().ToString("N"); - // If we are creating a correlated workflow, make sure to acquire a lock on it to prevent duplicate workflow instances from being created. - if (!string.IsNullOrWhiteSpace(correlationId)) + // Acquire a lock on correlation ID to prevent duplicate workflow instances from being created. + await using var correlationLockHandle = await AcquireLockAsync(correlationId, cancellationToken); + + // Acquire a lock on the workflow definition so that we can ensure singleton-workflows never execute more than one instance. + var lockKey = $"execute-workflow-definition:tenant:{tenantId}:workflow-definition:{workflowDefinitionId}"; + await using var workflowDefinitionHandle = await AcquireLockAsync(lockKey, cancellationToken); + + if (workflowBlueprint.IsSingleton) { - _logger.LogDebug("Acquiring lock on correlation ID {CorrelationId}", correlationId); - correlationLockHandle = await _distributedLockProvider.AcquireLockAsync(correlationId, _elsaOptions.DistributedLockTimeout, cancellationToken); - - if (correlationLockHandle == null) - throw new LockAcquisitionException($"Failed to acquire a lock on correlation ID {correlationId}"); - } - - try - { - // Acquire a lock on the workflow definition so that we can ensure singleton-workflows never execute more than one instance. - var lockKey = $"execute-workflow-definition:tenant:{tenantId}:workflow-definition:{workflowDefinitionId}"; - await using var handle = await _distributedLockProvider.AcquireLockAsync(lockKey, _elsaOptions.DistributedLockTimeout, cancellationToken); - - if (handle == null) - throw new LockAcquisitionException($"Failed to acquire a lock on {lockKey}"); - - if (workflowBlueprint.IsSingleton) + if (await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId)) { - if (await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId)) - { - _logger.LogDebug("Workflow {WorkflowDefinitionId} is a singleton workflow and is already running"); - return null; - } - } - - var startActivities = _getsStartActivities.GetStartActivities(workflowBlueprint).Select(x => x.Id).ToHashSet(); - var startActivityId = activityId == null ? startActivities.FirstOrDefault() : startActivities.Contains(activityId) ? activityId : default; - - if (startActivityId == null) - { - _logger.LogWarning("Cannot start workflow {WorkflowDefinitionId} with version {WorkflowDefinitionVersion} because it has no starting activities", workflowBlueprint.Id, workflowBlueprint.Version); + _logger.LogDebug("Workflow {WorkflowDefinitionId} is a singleton workflow and is already running", workflowDefinitionId); return null; } - - var workflowInstance = await _workflowFactory.InstantiateAsync( - workflowBlueprint, - correlationId, - contextId, - cancellationToken); - - await _workflowInstanceStore.SaveAsync(workflowInstance, cancellationToken); - return new StartableWorkflow(workflowBlueprint, workflowInstance, startActivityId); } - finally + + var startActivities = _getsStartActivities.GetStartActivities(workflowBlueprint).Select(x => x.Id).ToHashSet(); + var startActivityId = activityId == null ? startActivities.FirstOrDefault() : startActivities.Contains(activityId) ? activityId : default; + + if (startActivityId == null) { - if (correlationLockHandle != null) - await correlationLockHandle.DisposeAsync(); + _logger.LogWarning("Cannot start workflow {WorkflowDefinitionId} with version {WorkflowDefinitionVersion} because it has no starting activities", workflowBlueprint.Id, workflowBlueprint.Version); + return null; } + + var workflowInstance = await _workflowFactory.InstantiateAsync( + workflowBlueprint, + correlationId, + contextId, + cancellationToken); + + await _workflowInstanceStore.SaveAsync(workflowInstance, cancellationToken); + return new StartableWorkflow(workflowBlueprint, workflowInstance, startActivityId); } public async Task CollectAndExecuteStartableWorkflowAsync(string workflowDefinitionId, string? activityId, string? correlationId = default, string? contextId = default, object? input = default, string? tenantId = default, CancellationToken cancellationToken = default) @@ -263,30 +246,36 @@ namespace Elsa.Services { var correlationId = context.CorrelationId!; var lockKey = correlationId; + + await using var handle = await AcquireLockAsync(lockKey, cancellationToken); - await using (var handle = await _distributedLockProvider.AcquireLockAsync(lockKey, _elsaOptions.DistributedLockTimeout, cancellationToken)) + var correlatedWorkflowInstanceCount = !string.IsNullOrWhiteSpace(correlationId) + ? await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification(correlationId).WithStatus(WorkflowStatus.Suspended), cancellationToken) + : 0; + + _logger.LogDebug("Found {CorrelatedWorkflowCount} workflows with correlation ID {CorrelationId}", correlatedWorkflowInstanceCount, correlationId); + + if (correlatedWorkflowInstanceCount > 0) { - if (handle == null) - throw new LockAcquisitionException($"Failed to acquire a lock on {lockKey}"); - - var correlatedWorkflowInstanceCount = !string.IsNullOrWhiteSpace(correlationId) - ? await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification(correlationId).WithStatus(WorkflowStatus.Suspended), cancellationToken) - : 0; - - _logger.LogDebug("Found {{CorrelatedWorkflowCount}} workflows with correlation ID {CorrelationId}", correlatedWorkflowInstanceCount, correlationId); - - if (correlatedWorkflowInstanceCount > 0) - { - var bookmarkResults = context.Bookmark != null ? await _bookmarkFinder.FindBookmarksAsync(context.ActivityType, context.Bookmark, correlationId, context.TenantId, cancellationToken).ToList() : new List(); - _logger.LogDebug("Found {BookmarkCount} bookmarks for activity type {ActivityType}", bookmarkResults.Count, context.ActivityType); - return bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList(); - } + var bookmarkResults = context.Bookmark != null ? await _bookmarkFinder.FindBookmarksAsync(context.ActivityType, context.Bookmark, correlationId, context.TenantId, cancellationToken).ToList() : new List(); + _logger.LogDebug("Found {BookmarkCount} bookmarks for activity type {ActivityType}", bookmarkResults.Count, context.ActivityType); + return bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList(); } var startableWorkflows = await CollectStartableWorkflowsAsync(context, cancellationToken); return startableWorkflows.Select(x => new PendingWorkflow(x.WorkflowInstance.Id, x.ActivityId)).ToList(); } + private async Task AcquireLockAsync(string resource, CancellationToken cancellationToken) + { + var handle = await _distributedLockProvider.AcquireLockAsync(resource, _elsaOptions.DistributedLockTimeout, cancellationToken); + + if (handle == null) + throw new LockAcquisitionException($"Failed to acquire a lock on {resource}"); + + return handle; + } + private bool ValidatePreconditions(string? workflowDefinitionId, IWorkflowBlueprint? workflowBlueprint) { if (workflowBlueprint != null)