diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs index df335a24d..ca1ccd2ae 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs @@ -104,19 +104,11 @@ public class BulkDispatchWorkflows : Activity protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { var waitForCompletion = WaitForCompletion.GetOrDefault(context); - var items = context.GetItemSource(Items); - var dispatchedInstancesCount = 0; - - await foreach (var item in items) - { - await DispatchChildWorkflowAsync(context, item); - dispatchedInstancesCount++; - } - - context.SetProperty(DispatchedInstancesCountKey, dispatchedInstancesCount); + var items = await context.GetItemSource(Items).ToListAsync(context.CancellationToken); + var count = items.Count; // If we need to wait for the child workflows to complete (if any), create a bookmark. - if (waitForCompletion && dispatchedInstancesCount > 0) + if (waitForCompletion && count > 0) { var workflowInstanceId = context.WorkflowExecutionContext.Id; var bookmarkOptions = new CreateBookmarkArgs @@ -125,23 +117,45 @@ public class BulkDispatchWorkflows : Activity Stimulus = new BulkDispatchWorkflowsStimulus(workflowInstanceId) { ParentInstanceId = context.WorkflowExecutionContext.Id, - ScheduledInstanceIdsCount = dispatchedInstancesCount + ScheduledInstanceIdsCount = count }, IncludeActivityInstanceId = false, AutoBurn = false, }; + + // Create bookmarks first. context.CreateBookmark(bookmarkOptions); + + // Dispatch workflows afterwards. + await DispatchWorkflowsAsync(); } else { // Otherwise, we can complete immediately. + await DispatchWorkflowsAsync(); await context.CompleteActivityWithOutcomesAsync("Done"); } + + return; + + async Task DispatchWorkflowsAsync() + { + foreach (var item in items) + await DispatchChildWorkflowAsync(context, item, waitForCompletion); + + context.SetProperty(DispatchedInstancesCountKey, count); + } } - private async ValueTask DispatchChildWorkflowAsync(ActivityExecutionContext context, object item) + private async ValueTask DispatchChildWorkflowAsync(ActivityExecutionContext context, object item, bool waitForCompletion) { var workflowDefinitionId = WorkflowDefinitionId.Get(context); + var workflowDefinitionService = context.GetRequiredService(); + var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published); + + if (workflowGraph == null) + throw new($"No published version of workflow definition with ID {workflowDefinitionId} found."); + var parentInstanceId = context.WorkflowExecutionContext.Id; var input = Input.GetOrDefault(context) ?? new Dictionary(); var channelName = ChannelName.GetOrDefault(context); @@ -150,6 +164,9 @@ public class BulkDispatchWorkflows : Activity { ["ParentInstanceId"] = parentInstanceId }; + + if(waitForCompletion) + properties["WaitForCompletion"] = true; var itemDictionary = new Dictionary { @@ -168,12 +185,6 @@ public class BulkDispatchWorkflows : Activity var workflowDispatcher = context.GetRequiredService(); var identityGenerator = context.GetRequiredService(); var evaluator = context.GetRequiredService(); - var workflowDefinitionService = context.GetRequiredService(); - var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published); - - if (workflowGraph == null) - throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found."); - var correlationId = CorrelationIdFunction != null ? await evaluator.EvaluateAsync(CorrelationIdFunction!, context.ExpressionExecutionContext, evaluatorOptions) : null; var instanceId = identityGenerator.GenerateId(); var request = new DispatchWorkflowDefinitionRequest(workflowGraph.Workflow.Identity.Id) diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs index fa424c91d..4e579d60c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs @@ -72,7 +72,7 @@ public class DispatchWorkflow : Activity var waitForCompletion = WaitForCompletion.GetOrDefault(context); // Dispatch the child workflow. - var instanceId = await DispatchChildWorkflowAsync(context); + var instanceId = await DispatchChildWorkflowAsync(context, waitForCompletion); // If we need to wait for the child workflow to complete, create a bookmark. if (waitForCompletion) @@ -92,28 +92,39 @@ public class DispatchWorkflow : Activity } } - private async ValueTask DispatchChildWorkflowAsync(ActivityExecutionContext context) + private async ValueTask DispatchChildWorkflowAsync(ActivityExecutionContext context, bool waitForCompletion) { var workflowDefinitionId = WorkflowDefinitionId.Get(context); - var input = Input.GetOrDefault(context) ?? new Dictionary(); - var channelName = ChannelName.GetOrDefault(context); - - input["ParentInstanceId"] = context.WorkflowExecutionContext.Id; - - var correlationId = CorrelationId.GetOrDefault(context); - var workflowDispatcher = context.GetRequiredService(); - var identityGenerator = context.GetRequiredService(); var workflowDefinitionService = context.GetRequiredService(); var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken); if (workflowGraph == null) - throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found."); + throw new($"No published version of workflow definition with ID {workflowDefinitionId} found."); + + var input = Input.GetOrDefault(context) ?? new Dictionary(); + var channelName = ChannelName.GetOrDefault(context); + var parentInstanceId = context.WorkflowExecutionContext.Id; + var properties = new Dictionary + { + ["ParentInstanceId"] = parentInstanceId + }; + + if(waitForCompletion) + properties["WaitForCompletion"] = true; + + input["ParentInstanceId"] = parentInstanceId; + + var correlationId = CorrelationId.GetOrDefault(context); + var workflowDispatcher = context.GetRequiredService(); + var identityGenerator = context.GetRequiredService(); + var instanceId = identityGenerator.GenerateId(); var request = new DispatchWorkflowDefinitionRequest(workflowGraph.Workflow.Identity.Id) { - ParentWorkflowInstanceId = context.WorkflowExecutionContext.Id, + ParentWorkflowInstanceId = parentInstanceId, Input = input, + Properties = properties, CorrelationId = correlationId, InstanceId = instanceId, }; diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs index 7aaf05a5b..cba602166 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs @@ -19,7 +19,7 @@ namespace Elsa.Workflows.Runtime.Activities; public class ExecuteWorkflow : Activity { /// - public ExecuteWorkflow([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) + public ExecuteWorkflow([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line) { } @@ -31,7 +31,7 @@ public class ExecuteWorkflow : Activity Description = "The definition ID of the workflow to execute.", UIHint = InputUIHints.WorkflowDefinitionPicker )] - public Input WorkflowDefinitionId { get; set; } = default!; + public Input WorkflowDefinitionId { get; set; } = null!; /// /// The correlation ID to associate the workflow with. @@ -40,25 +40,25 @@ public class ExecuteWorkflow : Activity DisplayName = "Correlation ID", Description = "The correlation ID to associate the workflow with." )] - public Input CorrelationId { get; set; } = default!; + public Input CorrelationId { get; set; } = null!; /// /// The input to send to the workflow. /// [Input(Description = "The input to send to the workflow.")] - public Input?> Input { get; set; } = default!; + public Input?> Input { get; set; } = null!; /// /// True to wait for the child workflow to complete before completing this activity. If not set, the child workflow will be executed until it either completes or goes idle before this activity completes. /// [Input(Description = "Wait for the child workflow to complete before completing this activity.")] - public Input WaitForCompletion { get; set; } = default!; + public Input WaitForCompletion { get; set; } = null!; /// protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { - var result = await ExecuteWorkflowAsync(context); var waitForCompletion = WaitForCompletion.Get(context); + var result = await ExecuteWorkflowAsync(context, waitForCompletion); if(!waitForCompletion || result.Status == WorkflowStatus.Finished) { @@ -77,23 +77,35 @@ public class ExecuteWorkflow : Activity context.CreateBookmark(bookmarkOptions); } - private async ValueTask ExecuteWorkflowAsync(ActivityExecutionContext context) + private async ValueTask ExecuteWorkflowAsync(ActivityExecutionContext context, bool waitForCompletion) { var workflowDefinitionId = WorkflowDefinitionId.Get(context); - var input = Input.GetOrDefault(context) ?? new Dictionary(); - var correlationId = CorrelationId.GetOrDefault(context); - var workflowInvoker = context.GetRequiredService(); - var identityGenerator = context.GetRequiredService(); var workflowDefinitionService = context.GetRequiredService(); var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken); if (workflowGraph == null) - throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found."); - + throw new($"No published version of workflow definition with ID {workflowDefinitionId} found."); + + var parentInstanceId = context.WorkflowExecutionContext.Id; + var input = Input.GetOrDefault(context) ?? new Dictionary(); + var correlationId = CorrelationId.GetOrDefault(context); + var workflowInvoker = context.GetRequiredService(); + var identityGenerator = context.GetRequiredService(); + var properties = new Dictionary + { + ["ParentInstanceId"] = parentInstanceId + }; + + if(waitForCompletion) + properties["WaitForCompletion"] = true; + + input["ParentInstanceId"] = parentInstanceId; + var options = new RunWorkflowOptions { - ParentWorkflowInstanceId = context.WorkflowExecutionContext.Id, + ParentWorkflowInstanceId = parentInstanceId, Input = input, + Properties = properties, CorrelationId = correlationId, WorkflowInstanceId = identityGenerator.GenerateId() }; diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeBulkDispatchWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeBulkDispatchWorkflowActivity.cs index 6b683305b..7532d376e 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeBulkDispatchWorkflowActivity.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeBulkDispatchWorkflowActivity.cs @@ -21,11 +21,12 @@ internal class ResumeBulkDispatchWorkflowActivity(IBookmarkQueue bookmarkQueue, if (workflowState.Status != WorkflowStatus.Finished) return; - var parentInstanceId = workflowState.Properties.TryGetValue("ParentInstanceId", out var parentInstanceIdValue) ? parentInstanceIdValue.ToString() : default; - - if (string.IsNullOrWhiteSpace(parentInstanceId)) + var waitForCompletion = workflowState.Properties.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue; + + if (!waitForCompletion) return; - + + var parentInstanceId = (string)workflowState.Properties["ParentInstanceId"]; var activityTypeName = ActivityTypeNameHelper.GenerateTypeName(); var stimulus = new BulkDispatchWorkflowsStimulus(parentInstanceId); var stimulusHash = stimulusHasher.Hash(activityTypeName, stimulus); diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs index 607edf1f6..f1b52b181 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs @@ -2,7 +2,6 @@ using Elsa.Mediator.Contracts; using Elsa.Workflows.Helpers; using Elsa.Workflows.Notifications; using Elsa.Workflows.Runtime.Activities; -using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Stimuli; using JetBrains.Annotations; using Microsoft.Extensions.Logging; @@ -22,21 +21,32 @@ internal class ResumeDispatchWorkflowActivity(IBookmarkQueue bookmarkQueue, ISti var workflowState = notification.WorkflowState; logger.LogDebug("Handling workflow executed notification for workflow {WorkflowInstanceId}", notification.WorkflowState.Id); - + if (workflowState.Status != WorkflowStatus.Finished) { logger.LogDebug("Workflow {WorkflowInstanceId} is not in a finished state. Skipping resumption of any blocking DispatchWorkflow activities", notification.WorkflowState.Id); return; } + var props = workflowState.Properties; + var waitForCompletion = props.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue; + + if (!waitForCompletion) + { + logger.LogDebug("Workflow {WorkflowInstanceId} does not have a WaitForCompletion property set to true. Skipping resumption of any blocking DispatchWorkflow activities", notification.WorkflowState.Id); + return; + } + + var parentInstanceId = (string) props["ParentInstanceId"]; var stimulus = new DispatchWorkflowStimulus(notification.WorkflowState.Id); var input = workflowState.Output; - + var bookmarkQueueItem = new NewBookmarkQueueItem { + WorkflowInstanceId = parentInstanceId, ActivityTypeName = ActivityTypeName, StimulusHash = stimulusHasher.Hash(ActivityTypeName, stimulus), - Options = new ResumeBookmarkOptions + Options = new() { Input = input } diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs index 256621e0b..d4bb203a6 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs @@ -2,7 +2,6 @@ using Elsa.Mediator.Contracts; using Elsa.Workflows.Helpers; using Elsa.Workflows.Notifications; using Elsa.Workflows.Runtime.Activities; -using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Stimuli; using JetBrains.Annotations; @@ -11,7 +10,7 @@ namespace Elsa.Workflows.Runtime.Handlers; /// /// Resumes any blocking activities when its child workflow completes. /// -[PublicAPI] +[UsedImplicitly] internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStimulusHasher stimulusHasher) : INotificationHandler { private static readonly string ActivityTypeName = ActivityTypeNameHelper.GenerateTypeName(); @@ -23,14 +22,23 @@ internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStim if (workflowState.Status != WorkflowStatus.Finished) return; + var props = workflowState.Properties; + + var waitForCompletion = props.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue; + + if (!waitForCompletion) + return; + + var parentInstanceId = (string)props["ParentInstanceId"]; var stimulus = new ExecuteWorkflowStimulus(notification.WorkflowState.Id); var input = workflowState.Output; var bookmarkQueueItem = new NewBookmarkQueueItem { + WorkflowInstanceId = parentInstanceId, ActivityTypeName = ActivityTypeName, StimulusHash = stimulusHasher.Hash(ActivityTypeName, stimulus), - Options = new ResumeBookmarkOptions + Options = new() { Input = input }