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 412e23c38..6c72282e7 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs @@ -30,7 +30,7 @@ public partial class Flowchart : Container /// /// The activity to execute when the flowchart starts. /// - [Port] [Browsable(false)] public IActivity? Start { get; set; } + [Port][Browsable(false)] public IActivity? Start { get; set; } /// /// A list of connections between activities. @@ -52,343 +52,6 @@ public partial class Flowchart : Container await context.ScheduleActivityAsync(startActivity, OnChildCompletedAsync); } - private IActivity? GetStartActivity(ActivityExecutionContext context) - { - // If there's a trigger that triggered this workflow, use that. - var triggerActivityId = context.WorkflowExecutionContext.TriggerActivityId; - var triggerActivity = triggerActivityId != null ? Activities.FirstOrDefault(x => x.Id == triggerActivityId) : null; - - if (triggerActivity != null) - return triggerActivity; - - // If an explicit Start activity was provided, use that. - if (Start != null) - return Start; - - // If there is a Start activity on the flowchart, use that. - var startActivity = Activities.FirstOrDefault(x => x is Start); - - if (startActivity != null) - return startActivity; - - // If there's an activity marked as "Can Start Workflow", use that. - var canStartWorkflowActivity = Activities.FirstOrDefault(x => x.GetCanStartWorkflow()); - - if (canStartWorkflowActivity != null) - return canStartWorkflowActivity; - - // If there is a single activity that has no inbound connections, use that. - var root = GetRootActivity(); - - if (root != null) - return root; - - // If no start activity found, return the first activity. - return Activities.FirstOrDefault(); - } - - /// - /// Checks if there is any pending work for the flowchart. - /// - private bool HasPendingWork(ActivityExecutionContext context) - { - var workflowExecutionContext = context.WorkflowExecutionContext; - var activityIds = Activities.Select(x => x.Id).ToList(); - var children = context.Children; - var hasRunningActivityInstances = children.Where(x => activityIds.Contains(x.Activity.Id)).Any(x => x.Status == ActivityStatus.Running); - - var hasPendingWork = workflowExecutionContext.Scheduler.List().Any(workItem => - { - var ownerInstanceId = workItem.Owner?.Id; - - if (ownerInstanceId == null) - return false; - - if (ownerInstanceId == context.Id) - return true; - - var ownerContext = context.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.Id == ownerInstanceId); - var ancestors = ownerContext.GetAncestors().ToList(); - - return ancestors.Any(x => x == context); - }); - - return hasRunningActivityInstances || hasPendingWork; - } - - private IActivity? GetRootActivity() - { - // Get the first activity that has no inbound connections. - var query = - from activity in Activities - let inboundConnections = Connections.Any(x => x.Target.Activity == activity) - where !inboundConnections - select activity; - - var rootActivity = query.FirstOrDefault(); - return rootActivity; - } - - private FlowGraph GetFlowGraph(ActivityExecutionContext context) - { - // Store in TransientProperties so FlowChart is not persisted in WorkflowState - return context.TransientProperties.GetOrAdd(GraphTransientProperty, () => new FlowGraph(Connections, GetStartActivity(context))); - } - - private FlowScope GetFlowScope(ActivityExecutionContext context) - { - return context.GetProperty(ScopeProperty, () => new FlowScope()); - } - - private async ValueTask OnChildCompletedAsync(ActivityCompletedContext context) - { - var flowchartContext = context.TargetContext; - var completedActivityContext = context.ChildContext; - var completedActivity = completedActivityContext.Activity; - var result = context.Result; - - if (flowchartContext.Activity != this) - { - throw new("Target context activity must be this flowchart"); - } - - // If the completed activity's status is anything but "Completed", do not schedule its outbound activities. - if (completedActivityContext.Status != ActivityStatus.Completed) - { - return; - } - - // If the complete activity is a terminal node, complete the flowchart immediately. - if (completedActivity is ITerminalNode) - { - await flowchartContext.CompleteActivityAsync(); - return; - } - - // Determine the outcomes from the completed activity. - var outcomes = result is Outcomes o ? o : Outcomes.Default; - - // Schedule the outbound activities - var flowGraph = GetFlowGraph(flowchartContext); - var flowScope = GetFlowScope(flowchartContext); - var completedActivityExcecutedByBackwardConnection = completedActivityContext.ActivityInput.GetValueOrDefault(BackwardConnectionActivityInput); - var hasScheduledActivity = await ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, completedActivity, outcomes, completedActivityExcecutedByBackwardConnection); - - // If there are not any outbound connections, complete the flowchart activity if there is no other pending work. - if (!hasScheduledActivity) - { - await CompleteIfNoPendingWorkAsync(flowchartContext); - } - } - - /// - /// Schedules outbound activities based on the flowchart's structure and execution state. - /// This method determines whether an activity should be scheduled based on visited connections, - /// forward traversal rules, and backward connections. - /// - /// The flowchart containing the activities. - /// The graph representation of the flowchart. - /// Tracks activity and connection visits. - /// The execution context of the flowchart. - /// The current activity being processed. - /// The outcomes that determine which connections were followed. - /// Indicates if the completed activity was executed due to a backward connection. - /// True if at least one activity was scheduled; otherwise, false. - private async ValueTask ScheduleOutboundActivitiesAsync(FlowGraph flowGraph, FlowScope flowScope, ActivityExecutionContext flowchartContext, IActivity activity, Outcomes outcomes, bool completedActivityExecutedByBackwardConnection = false) - { - var hasScheduledActivity = false; - - // Check if the activity is dangling (i.e., it is not reachable from the flowchart graph) - if (flowGraph.IsDanglingActivity(activity)) - { - throw new($"Activity {activity.Id} is not reachable from the flowchart graph. Unable to schedule it's outbound activities."); - } - - // Register the activity as visited unless it was executed due to a backward connection - if (!completedActivityExecutedByBackwardConnection) - { - flowScope.RegisterActivityVisit(activity); - } - - // Process each outbound connection from the current activity - foreach (var outboundConnection in flowGraph.GetOutboundConnections(activity)) - { - var connectionFollowed = outcomes.Names.Contains(outboundConnection.Source.Port); - flowScope.RegisterConnectionVisit(outboundConnection, connectionFollowed); - - if(!connectionFollowed) - continue; // Skip if connection was not followed. - - var outboundActivity = outboundConnection.Target.Activity; - - // Determine the scheduling strategy based on connection-type. - if (flowGraph.IsBackwardConnection(outboundConnection, out var backwardConnectionIsValid)) - { - hasScheduledActivity |= await ScheduleBackwardConnectionActivityAsync(flowGraph, flowchartContext, outboundConnection, outboundActivity, connectionFollowed, backwardConnectionIsValid); - } - else if (outboundActivity is not IJoinNode) - { - hasScheduledActivity |= await ScheduleNonJoinActivityAsync(flowGraph, flowScope, flowchartContext, outboundActivity); - } - else - { - hasScheduledActivity |= await ScheduleJoinActivityAsync(flowGraph, flowScope, flowchartContext, outboundConnection, outboundActivity); - } - } - return hasScheduledActivity; - } - - /// - /// Schedules an outbound activity that originates from a backward connection. - /// - private async ValueTask ScheduleBackwardConnectionActivityAsync(FlowGraph flowGraph, ActivityExecutionContext flowchartContext, Connection outboundConnection, IActivity outboundActivity, bool connectionFollowed, bool backwardConnectionIsValid) - { - if (!connectionFollowed) - { - return false; - } - - if (!backwardConnectionIsValid) - { - throw new($"Invalid backward connection: Every path from the source ('{outboundConnection.Source.Activity.Id}') must go through the target ('{outboundConnection.Target.Activity.Id}') when tracing back to the start."); - } - - var scheduleWorkOptions = new ScheduleWorkOptions - { - CompletionCallback = OnChildCompletedAsync, - Input = new Dictionary() { { BackwardConnectionActivityInput, true } } - }; - - await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions); - return true; - } - - /// - /// Schedules a non-join activity if all its forward inbound connections have been visited. - /// - private async ValueTask ScheduleNonJoinActivityAsync(FlowGraph flowGraph, FlowScope flowScope, ActivityExecutionContext flowchartContext, IActivity outboundActivity) - { - if (!flowScope.AllInboundConnectionsVisited(flowGraph, outboundActivity)) - { - return false; - } - - if (flowScope.HasFollowedInboundConnection(flowGraph, outboundActivity)) - { - await flowchartContext.ScheduleActivityAsync(outboundActivity, OnChildCompletedAsync); - return true; - } - else - { - // Propagate skipped connections by scheduling with Outcomes.Empty - return await ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, outboundActivity, Outcomes.Empty); - } - } - - /// - /// Schedules a join activity based on inbound connection statuses. - /// - private async ValueTask ScheduleJoinActivityAsync(FlowGraph flowGraph, FlowScope flowScope, ActivityExecutionContext flowchartContext, Connection outboundConnection, IActivity outboundActivity) - { - // Ignore the connection if the join activity has already completed (JoinAny scenario) - if (flowScope.ShouldIgnoreConnection(outboundConnection, outboundActivity)) - { - return false; - } - - // Schedule the join activity only if at least one inbound connection was followed - if (!flowScope.HasFollowedInboundConnection(flowGraph, outboundActivity)) - { - if (flowScope.AllInboundConnectionsVisited(flowGraph, outboundActivity)) - { - // Propagate skipped connections by scheduling with Outcomes.Empty - return await ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, outboundActivity, Outcomes.Empty); - } - return false; - } - - // Check for an existing execution context for the join activity - var joinContext = flowchartContext.WorkflowExecutionContext.ActivityExecutionContexts.LastOrDefault(x => - x.ParentActivityExecutionContext == flowchartContext && - x.Activity == outboundActivity && - x.Status is ActivityStatus.Pending or ActivityStatus.Running); - - // If the join activity was already scheduled, do not schedule it again - if (joinContext == null) - { - var activityScheduled = flowchartContext.WorkflowExecutionContext.Scheduler.List().Any(workItem => workItem.Owner == flowchartContext && workItem.Activity == outboundActivity); - if (activityScheduled) - { - return true; - } - } - - var scheduleWorkOptions = new ScheduleWorkOptions - { - CompletionCallback = OnChildCompletedAsync, - ExistingActivityExecutionContext = joinContext - }; - await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions); - return true; - } - - public static bool CanWaitAllProceed(ActivityExecutionContext context) - { - var flowchartContext = context.ParentActivityExecutionContext!; - var flowchart = (Flowchart)flowchartContext.Activity; - var flowGraph = flowchart.GetFlowGraph(flowchartContext); - var flowScope = flowchart.GetFlowScope(flowchartContext); - var activity = context.Activity; - - return flowScope.AllInboundConnectionsVisited(flowGraph, activity); - } - - public static async void CancelAncestorActivatesAsync(ActivityExecutionContext context) - { - var flowchartContext = context.ParentActivityExecutionContext!; - var flowchart = (Flowchart)flowchartContext.Activity; - var flowGraph = flowchart.GetFlowGraph(flowchartContext); - var ancestorActivities = flowGraph.GetAncestorActivities(context.Activity); - var inboundActivityExecutionContexts = context.WorkflowExecutionContext.ActivityExecutionContexts.Where(x => ancestorActivities.Contains(x.Activity) && x.ParentActivityExecutionContext == flowchartContext).ToList(); - - // Cancel each ancestor activity. - foreach (var activityExecutionContext in inboundActivityExecutionContexts) - { - await activityExecutionContext.CancelActivityAsync(); - } - } - - private async Task CompleteIfNoPendingWorkAsync(ActivityExecutionContext context) - { - var hasPendingWork = HasPendingWork(context); - - if (!hasPendingWork) - { - var hasFaultedActivities = context.Children.Any(x => x.Status == ActivityStatus.Faulted); - - if (!hasFaultedActivities) - { - await context.CompleteActivityAsync(); - } - } - } - - 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); - } - } - private async ValueTask OnScheduleChildActivityAsync(ScheduleChildActivity signal, SignalContext context) { var flowchartContext = context.ReceiverActivityExecutionContext;