SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Add ability to schedule outcome names.

Browse source
This commit is contained in:
Sipke Schoorstra 2023-09-10 15:34:26 +02:00
parent 73326bb228
commit 5d7188a4e8
5 changed files with 135 additions and 91 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

163
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs
View file

@ -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";
/// <inheritdoc />
public Flowchart([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
{
OnSignalReceived<ScheduleActivityOutcomes>(OnScheduleOutcomesAsync);
}
/// <summary>
@ -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<Connection, bool> 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<Connection>() : 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);
}
/// <summary>
/// Checks if there is any pending work for the flowchart.
/// </summary>
@ -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<Connection, bool> 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<Connection>() : 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);
}
}
}

4
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs
View file

@ -183,7 +183,7 @@ public class ActivityExecutionContext : IExecutionContext
/// <param name="variables">An optional list of variables to declare with the activity execution.</param>
public async ValueTask ScheduleActivityAsync(IActivity? activity, ActivityCompletionCallback? completionCallback, object? tag = default, IEnumerable<Variable>? 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
/// <param name="variables">An optional list of variables to declare with the activity execution.</param>
public async ValueTask ScheduleActivities(IEnumerable<IActivity?> activities, ActivityCompletionCallback? completionCallback, object? tag = default, IEnumerable<Variable>? variables = default)
{
var options = new ScheduleWorkOptions(completionCallback, tag, variables);
var options = new ScheduleWorkOptions(completionCallback, tag, variables?.ToList());
await ScheduleActivities(activities, options);
}

50
src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs
View file

@ -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;
}
/// <summary>
/// Schedules the specified activity.
/// </summary>
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);
}
/// <summary>
/// Returns the outcome name for the specified port property name.
/// </summary>
@ -417,6 +403,38 @@ public static class ActivityExecutionContextExtensions
// Update the completed at timestamp.
context.CompletedAt = context.WorkflowExecutionContext.SystemClock.UtcNow;
}
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
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<ISafeSerializer>();
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));
}
/// <summary>
/// Complete the current activity with the specified outcome.

2
src/modules/Elsa.Workflows.Core/Options/ScheduleWorkOptions.cs
View file

@ -9,4 +9,4 @@ namespace Elsa.Workflows.Core.Options;
/// <param name="CompletionCallback">The callback to invoke when the work item has completed.</param>
/// <param name="Tag">A tag that can be used to identify the work item.</param>
/// <param name="Variables">A collection of variables to declare in the activity execution context that will be created for this work item.</param>
public record ScheduleWorkOptions(ActivityCompletionCallback? CompletionCallback = default, object? Tag = default, IEnumerable<Variable>? Variables = default);
public record ScheduleWorkOptions(ActivityCompletionCallback? CompletionCallback = default, object? Tag = default, ICollection<Variable>? Variables = default);

7
src/modules/Elsa.Workflows.Core/Signals/ScheduleActivityOutcomes.cs Normal file
View file

@ -0,0 +1,7 @@
namespace Elsa.Workflows.Core.Signals;
/// <summary>
/// Signaled when an activity requests the scheduling of the specified set of outcomes.
/// </summary>
/// <param name="Outcomes">The outcomes to schedule.</param>
public record ScheduleActivityOutcomes(params string[] Outcomes);