diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs index 13ed23122..48c9b7795 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs @@ -6,6 +6,7 @@ using Elsa.Workflows.Core.Activities.Flowchart.Extensions; using Elsa.Workflows.Core.Activities.Flowchart.Models; using Elsa.Workflows.Core.Attributes; using Elsa.Workflows.Core.Contracts; +using Elsa.Workflows.Core.Signals; namespace Elsa.Workflows.Core.Activities.Flowchart.Activities; @@ -17,10 +18,12 @@ namespace Elsa.Workflows.Core.Activities.Flowchart.Activities; public class Flowchart : Container { internal const string ScopeProperty = "Scope"; + internal const string BranchMonitorsProperty = "BranchMonitoring"; /// public Flowchart([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) { + OnSignalReceived(OnScheduleOutcomesAsync); } /// @@ -51,78 +54,6 @@ public class Flowchart : Container await context.ScheduleActivityAsync(startActivity, OnChildCompletedAsync); } - private async ValueTask OnChildCompletedAsync(ActivityCompletedContext context) - { - var targetContext = context.TargetContext; - var childContext = context.ChildContext; - var completedActivity = childContext.Activity; - var result = context.Result; - var alreadyCompleted = result is AlreadyCompleted; - - // If specific outcomes were provided by the completed activity, use them to find the connection to the next activity. - Func outboundConnectionsQuery = result is Outcomes outcomes - ? connection => connection.Source.Activity == completedActivity && outcomes.Names.Contains(connection.Source.Port) - : connection => connection.Source.Activity == completedActivity; - - // Only query the outbound connections if the completed activity wasn't already completed. - var outboundConnections = alreadyCompleted ? new List() : Connections.Where(outboundConnectionsQuery).ToList(); - var children = outboundConnections.Select(x => x.Target.Activity).ToList(); - var scope = targetContext.GetProperty(ScopeProperty, () => new FlowScope()); - - scope.RegisterActivityExecution(completedActivity); - - // If the completed activity is an End or Break activity, complete the flowchart immediately. - if (completedActivity is End or Break) - { - await targetContext.CompleteActivityAsync(); - } - else - { - if (children.Any()) - { - scope.AddActivities(children); - - // Schedule each child, but only if all of its left inbound activities have already executed. - foreach (var activity in children) - { - var inboundActivities = Connections.LeftInboundActivities(activity).ToList(); - - // If the completed activity is not part of the left inbound path, always allow its children to be scheduled. - if (!inboundActivities.Contains(completedActivity)) - { - await targetContext.ScheduleActivityAsync(activity, OnChildCompletedAsync); - continue; - } - - // If the activity is anything but a join activity, only schedule it if all of its left-inbound activities have executed, effectively implementing a "wait all" join. - if (activity is not IJoinNode) - { - var executionCount = scope.GetExecutionCount(activity); - var haveInboundActivitiesExecuted = inboundActivities.All(x => scope.GetExecutionCount(x) > executionCount); - - if (haveInboundActivitiesExecuted) - await targetContext.ScheduleActivityAsync(activity, OnChildCompletedAsync); - } - else - { - await targetContext.ScheduleActivityAsync(activity, OnChildCompletedAsync); - } - } - } - - if (!children.Any()) - { - // If there is no pending work, complete the flowchart activity. - var hasPendingWork = HasPendingWork(targetContext); - - if (!hasPendingWork) - await targetContext.CompleteActivityAsync(); - } - } - - targetContext.SetProperty(ScopeProperty, scope); - } - /// /// Checks if there is any pending work for the flowchart. /// @@ -216,4 +147,92 @@ public class Flowchart : Container var rootActivity = query.FirstOrDefault(); return rootActivity; } + + private async ValueTask OnChildCompletedAsync(ActivityCompletedContext context) + { + var flowchartContext = context.TargetContext; + var completedActivityContext = context.ChildContext; + var completedActivity = completedActivityContext.Activity; + var result = context.Result; + var alreadyCompleted = result is AlreadyCompleted; + + // If specific outcomes were provided by the completed activity, use them to find the connection to the next activity. + Func outboundConnectionsQuery = result is Outcomes outcomes + ? connection => connection.Source.Activity == completedActivity && outcomes.Names.Contains(connection.Source.Port) + : connection => connection.Source.Activity == completedActivity; + + // Only query the outbound connections if the completed activity wasn't already completed. + var outboundConnections = alreadyCompleted ? new List() : Connections.Where(outboundConnectionsQuery).ToList(); + var children = outboundConnections.Select(x => x.Target.Activity).ToList(); + var scope = flowchartContext.GetProperty(ScopeProperty, () => new FlowScope()); + + scope.RegisterActivityExecution(completedActivity); + + // If the completed activity is an End or Break activity, complete the flowchart immediately. + if (completedActivity is End or Break) + { + await flowchartContext.CompleteActivityAsync(); + } + else + { + if (children.Any()) + { + scope.AddActivities(children); + + // Schedule each child, but only if all of its left inbound activities have already executed. + foreach (var activity in children) + { + var inboundActivities = Connections.LeftInboundActivities(activity).ToList(); + + // If the completed activity is not part of the left inbound path, always allow its children to be scheduled. + if (!inboundActivities.Contains(completedActivity)) + { + await flowchartContext.ScheduleActivityAsync(activity, OnChildCompletedAsync); + continue; + } + + // If the activity is anything but a join activity, only schedule it if all of its left-inbound activities have executed, effectively implementing a "wait all" join. + if (activity is not IJoinNode) + { + var executionCount = scope.GetExecutionCount(activity); + var haveInboundActivitiesExecuted = inboundActivities.All(x => scope.GetExecutionCount(x) > executionCount); + + if (haveInboundActivitiesExecuted) + await flowchartContext.ScheduleActivityAsync(activity, OnChildCompletedAsync); + } + else + { + await flowchartContext.ScheduleActivityAsync(activity, OnChildCompletedAsync); + } + } + } + + if (!children.Any()) + { + // If there is no pending work, complete the flowchart activity. + var hasPendingWork = HasPendingWork(flowchartContext); + + if (!hasPendingWork) + await flowchartContext.CompleteActivityAsync(); + } + } + + flowchartContext.SetProperty(ScopeProperty, scope); + } + + private async ValueTask OnScheduleOutcomesAsync(ScheduleActivityOutcomes signal, SignalContext context) + { + var flowchartContext = context.ReceiverActivityExecutionContext; + var schedulingActivityContext = context.SenderActivityExecutionContext; + var schedulingActivity = schedulingActivityContext.Activity; + var outcomes = signal.Outcomes; + var outboundConnections = Connections.Where(connection => connection.Source.Activity == schedulingActivity && outcomes.Contains(connection.Source.Port!)).ToList(); + var outboundActivities = outboundConnections.Select(x => x.Target.Activity).ToList(); + + if (outboundActivities.Any()) + { + // Schedule each child. + foreach (var activity in outboundActivities) await flowchartContext.ScheduleActivityAsync(activity, OnChildCompletedAsync); + } + } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs index 8e8e5a406..59c2ec0dd 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs @@ -183,7 +183,7 @@ public class ActivityExecutionContext : IExecutionContext /// An optional list of variables to declare with the activity execution. public async ValueTask ScheduleActivityAsync(IActivity? activity, ActivityCompletionCallback? completionCallback, object? tag = default, IEnumerable? variables = default) { - var options = new ScheduleWorkOptions(completionCallback, tag, variables); + var options = new ScheduleWorkOptions(completionCallback, tag, variables?.ToList()); await ScheduleActivityAsync(activity, options); } @@ -252,7 +252,7 @@ public class ActivityExecutionContext : IExecutionContext /// An optional list of variables to declare with the activity execution. public async ValueTask ScheduleActivities(IEnumerable activities, ActivityCompletionCallback? completionCallback, object? tag = default, IEnumerable? variables = default) { - var options = new ScheduleWorkOptions(completionCallback, tag, variables); + var options = new ScheduleWorkOptions(completionCallback, tag, variables?.ToList()); await ScheduleActivities(activities, options); } diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs index b52dea927..8cc7ba456 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs @@ -14,6 +14,7 @@ using Elsa.Workflows.Core.Contracts; using Elsa.Workflows.Core.Memory; using Elsa.Workflows.Core.Models; using Elsa.Workflows.Core.Notifications; +using Elsa.Workflows.Core.Services; using Elsa.Workflows.Core.Signals; using JetBrains.Annotations; @@ -237,22 +238,7 @@ public static class ActivityExecutionContextExtensions context.ActivityState[inputDescriptor.Name] = value!; return value; } - - /// - /// Schedules the specified activity. - /// - public static async Task ScheduleOutcomeAsync(this ActivityExecutionContext context, IActivity? activity, [CallerArgumentExpression("activity")] string portPropertyName = default!) - { - if (activity == null) - { - var outcome = context.GetOutcomeName(portPropertyName); - await context.CompleteActivityWithOutcomesAsync(outcome); - return; - } - - await context.ScheduleActivityAsync(activity, context); - } - + /// /// Returns the outcome name for the specified port property name. /// @@ -417,6 +403,38 @@ public static class ActivityExecutionContextExtensions // Update the completed at timestamp. context.CompletedAt = context.WorkflowExecutionContext.SystemClock.UtcNow; } + + /// + /// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion. + /// + public static async ValueTask ScheduleOutcomesAsync(this ActivityExecutionContext context, params string[] outcomes) + { + // Record the outcomes, if any. + context.JournalData["Outcomes"] = outcomes; + + // Record the output, if any. + var activity = context.Activity; + var expressionExecutionContext = context.ExpressionExecutionContext; + var activityDescriptor = context.ActivityDescriptor; + var outputDescriptors = activityDescriptor.Outputs; + var outputs = outputDescriptors.ToDictionary(x => x.Name, x => activity.GetOutput(expressionExecutionContext, x.Name)!); + var serializer = context.GetRequiredService(); + + foreach (var output in outputs) + { + var outputName = output.Key; + var outputValue = output.Value; + + if (outputValue == null!) + continue; + + var serializedOutputValue = serializer.Serialize(outputValue); + context.JournalData[outputName] = serializedOutputValue; + } + + // Send a signal. + await context.SendSignalAsync(new ScheduleActivityOutcomes(outcomes)); + } /// /// Complete the current activity with the specified outcome. diff --git a/src/modules/Elsa.Workflows.Core/Options/ScheduleWorkOptions.cs b/src/modules/Elsa.Workflows.Core/Options/ScheduleWorkOptions.cs index 00ecec97b..b21795cda 100644 --- a/src/modules/Elsa.Workflows.Core/Options/ScheduleWorkOptions.cs +++ b/src/modules/Elsa.Workflows.Core/Options/ScheduleWorkOptions.cs @@ -9,4 +9,4 @@ namespace Elsa.Workflows.Core.Options; /// The callback to invoke when the work item has completed. /// A tag that can be used to identify the work item. /// A collection of variables to declare in the activity execution context that will be created for this work item. -public record ScheduleWorkOptions(ActivityCompletionCallback? CompletionCallback = default, object? Tag = default, IEnumerable? Variables = default); \ No newline at end of file +public record ScheduleWorkOptions(ActivityCompletionCallback? CompletionCallback = default, object? Tag = default, ICollection? Variables = default); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Signals/ScheduleActivityOutcomes.cs b/src/modules/Elsa.Workflows.Core/Signals/ScheduleActivityOutcomes.cs new file mode 100644 index 000000000..3dd1ba212 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Signals/ScheduleActivityOutcomes.cs @@ -0,0 +1,7 @@ +namespace Elsa.Workflows.Core.Signals; + +/// +/// Signaled when an activity requests the scheduling of the specified set of outcomes. +/// +/// The outcomes to schedule. +public record ScheduleActivityOutcomes(params string[] Outcomes); \ No newline at end of file