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

Fix Fork activity

Browse source
This commit is contained in:
Sipke Schoorstra 2021-04-02 16:54:26 +02:00
parent 8b0b479b92
commit 461e38bb32
1 changed files with 2 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

91
src/core/Elsa.Core/Activities/ControlFlow/Fork/Fork.cs
View file

@ -16,100 +16,13 @@ using Newtonsoft.Json.Linq;
namespace Elsa.Activities.ControlFlow
{
[Activity(Category = "Control Flow", Description = "Fork workflow execution into multiple branches.")]
public class Fork : Activity, INotificationHandler<ScopeEvicted>
public class Fork : Activity
{
private readonly IMediator _mediator;
public Fork(IMediator mediator)
{
_mediator = mediator;
}
[ActivityProperty(
UIHint = ActivityPropertyUIHints.MultiText,
Hint = "Enter one or more branch names.")]
public ISet<string> Branches { get; set; } = new HashSet<string>();
public bool EnteredScope
{
get => GetState<bool>();
set => SetState(value);
}
protected override async ValueTask<IActivityExecutionResult> OnExecuteAsync(ActivityExecutionContext context)
{
if (!context.WorkflowInstance.Scopes.Contains(x => x.ActivityId == Id))
{
if (!EnteredScope)
{
context.CreateScope();
EnteredScope = true;
}
else
{
EnteredScope = false;
await UnwindAsync(context);
return Done();
}
}
return Combine(Done(), Outcomes(Branches));
}
private async Task UnwindAsync(ActivityExecutionContext context)
{
RemoveBlockingActivities(context);
await RemoveScopedActivitiesAsync(context);
}
private async Task RemoveScopedActivitiesAsync(ActivityExecutionContext context)
{
var workflowExecutionContext = context.WorkflowExecutionContext;
var scopes = workflowExecutionContext.WorkflowInstance.Scopes.ToList();
var descendants = workflowExecutionContext.GetOutboundActivityPath(Id).ToHashSet();
for (var i = 0; i < scopes.Count; i++)
{
var activityScope = scopes.ElementAt(i);
if(!descendants.Contains(activityScope.ActivityId))
continue;
var evictedScopes = scopes.Skip(i).ToList();
scopes = scopes.Take(i).ToList();
foreach (var evictedScope in evictedScopes)
{
var activity = workflowExecutionContext.WorkflowBlueprint.GetActivity(evictedScope.ActivityId)!;
await _mediator.Publish(new ScopeEvicted(workflowExecutionContext, activity));
}
workflowExecutionContext.WorkflowInstance.Scopes = new SimpleStack<ActivityScope>(scopes.AsEnumerable());
break;
}
}
private void RemoveBlockingActivities(ActivityExecutionContext context)
{
var workflowExecutionContext = context.WorkflowExecutionContext;
var workflowInstance = workflowExecutionContext.WorkflowInstance;
var blockingActivities = workflowInstance.BlockingActivities.ToList();
var outboundActivityIds = workflowExecutionContext.GetOutboundActivityPath(Id).ToHashSet();
var blockingActivityIdsInFork = blockingActivities.Where(x => outboundActivityIds.Contains(x.ActivityId)).Select(x => x.ActivityId).ToHashSet();
workflowInstance.BlockingActivities.RemoveWhere(x => blockingActivityIdsInFork.Contains(x.ActivityId));
}
Task INotificationHandler<ScopeEvicted>.Handle(ScopeEvicted notification, CancellationToken cancellationToken)
{
if (notification.EvictedScope.Type != nameof(Fork))
return Task.CompletedTask;
var data = notification.WorkflowExecutionContext.WorkflowInstance.ActivityData.GetItem(notification.EvictedScope.Id, () => new JObject());
data.SetState(nameof(EnteredScope), false);
data.SetState("Unwinding", false);
return Task.CompletedTask;
}
protected override IActivityExecutionResult OnExecute(ActivityExecutionContext context) => Outcomes(Branches);
}
}