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

Initial support for activity cancellation

Browse source
This commit is contained in:
Sipke Schoorstra 2022-10-30 23:44:38 +01:00
parent 421324c9a7
commit e4f9b7717b
10 changed files with 41 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/modules/Elsa.Workflows.Core/Activities/Break.cs
View file

@ -10,6 +10,6 @@ public class Break : Activity
{
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
await context.SignalAsync(new BreakSignal());
await context.SendSignalAsync(new BreakSignal());
}
}

11
src/modules/Elsa.Workflows.Core/Behaviors/BreakBehavior.cs
View file

@ -18,13 +18,22 @@ public class BreakBehavior : Behavior
private async ValueTask OnBreakAsync(BreakSignal signal, SignalContext context)
{
var cancellationToken = context.CancellationToken;
// Prevent bubbling.
context.StopPropagation();
// Cancel each descendant to clear bookmarks and cancel jobs etc.
var descendants = context.ReceiverActivityExecutionContext.ActivityNode.Descendants().Select(x => x.Activity).ToList();
var workflowExecutionContext = context.SenderActivityExecutionContext.WorkflowExecutionContext;
foreach (var descendant in descendants)
await workflowExecutionContext.CancelActivityAsync(descendant.Id);
// Remove child activity execution contexts.
var childActivityExecutionContexts = context.ReceiverActivityExecutionContext.GetChildren().ToList();
context.ReceiverActivityExecutionContext.WorkflowExecutionContext.RemoveActivityExecutionContexts(childActivityExecutionContexts);
// Mark this activity as completed.
await context.ReceiverActivityExecutionContext.CompleteActivityAsync();
}

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

@ -128,7 +128,7 @@ public static class ActivityExecutionContextExtensions
/// <summary>
/// Send a signal up the current branch.
/// </summary>
public static async ValueTask SignalAsync(this ActivityExecutionContext context, object signal)
public static async ValueTask SendSignalAsync(this ActivityExecutionContext context, object signal)
{
var ancestorContexts = new[] { context }.Concat(context.GetAncestors());
@ -152,11 +152,20 @@ public static class ActivityExecutionContextExtensions
public static async ValueTask CompleteActivityAsync(this ActivityExecutionContext context, object? result = default)
{
// Send a signal.
await context.SignalAsync(new ActivityCompleted(result));
await context.SendSignalAsync(new ActivityCompleted(result));
// Remove the context.
context.WorkflowExecutionContext.ActivityExecutionContexts.Remove(context);
}
/// <summary>
/// Cancel the activity. For blocking activities, it means their bookmarks will be removed. For job activities, the background work will be cancelled.
/// </summary>
public static async Task CancelActivityAsync(this ActivityExecutionContext context)
{
context.ClearBookmarks();
await context.SendSignalAsync(new CancelSignal());
}
public static ILogger GetLogger(this ActivityExecutionContext context) => (ILogger)context.GetRequiredService(typeof(ILogger<>).MakeGenericType(context.Activity.GetType()));

2
src/modules/Elsa.Workflows.Core/Models/ActivityExecutionContext.cs
View file

@ -199,7 +199,7 @@ public class ActivityExecutionContext
var entriesToRemove = WorkflowExecutionContext.CompletionCallbacks.Where(x => x.Owner == this);
WorkflowExecutionContext.RemoveCompletionCallbacks(entriesToRemove);
}
private MemoryBlock? GetBlock(MemoryBlockReference locationBlockReference) =>
ExpressionExecutionContext.Memory.TryGetBlock(locationBlockReference.Id, out var location)
? location

15
src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs
View file

@ -4,6 +4,7 @@ using System.Collections.ObjectModel;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Common.Extensions;
using Elsa.Expressions.Models;
using Elsa.Workflows.Core.Services;
using Microsoft.Extensions.DependencyInjection;
@ -129,7 +130,7 @@ public class WorkflowExecutionContext
return value;
}
public void TransitionTo(WorkflowSubStatus subStatus)
{
var targetStatus = GetMainStatus(subStatus);
@ -150,6 +151,16 @@ public class WorkflowExecutionContext
return activityExecutionContext;
}
public async Task CancelActivityAsync(string activityId)
{
var activityExecutionContext = ActivityExecutionContexts.FirstOrDefault(x => x.Id == activityId);
if (activityExecutionContext != null)
await activityExecutionContext.CancelActivityAsync();
Bookmarks.RemoveWhere(x => x.ActivityId == activityId);
}
private WorkflowStatus GetMainStatus(WorkflowSubStatus subStatus) =>
subStatus switch
{
@ -166,6 +177,6 @@ public class WorkflowExecutionContext
var currentMainStatus = GetMainStatus(currentSubStatus);
return currentMainStatus != WorkflowStatus.Finished;
}
private IEnumerable<MemoryRegister> GetMergedRegistersView() => new[] { MemoryRegister }.Concat(ActivityExecutionContexts.Select(x => x.ExpressionExecutionContext.Memory)).ToList();
}

3
src/modules/Elsa.Workflows.Core/Signals/CancelSignal.cs Normal file
View file

@ -0,0 +1,3 @@
namespace Elsa.Workflows.Core.Signals;
public record CancelSignal;

2
test/Elsa.IntegrationTests/Activities/Fork/ForkTests.cs
View file

@ -28,7 +28,7 @@ public class ForkTests
var workflow = await _workflowBuilderFactory.CreateBuilder().BuildWorkflowAsync<BasicForkWorkflow>();
await _workflowRunner.RunAsync(workflow);
var lines = _capturingTextWriter.Lines.ToList();
Assert.Equal(new[]{ "Branch 1", "Branch 2", "Branch 3" }, lines);
Assert.Equal(new[]{ "Branch 3", "Branch 2", "Branch 1" }, lines);
}
[Fact(DisplayName = "Wait Any causes workflow to continue")]

2
test/Elsa.IntegrationTests/Scenarios/BlockingAndBreaking/Tests.cs
View file

@ -39,6 +39,6 @@ public class Tests
// Verify expected output.
var lines = _capturingTextWriter.Lines.ToList();
Assert.Equal(new[] { "Branch 1", "Branch 2", "Branch 1 - Resumed" }, lines);
Assert.Equal(new[] { "Branch 2", "Branch 1", "Branch 1 - Resumed" }, lines);
}
}

40
test/Elsa.IntegrationTests/Scenarios/Persistence/Tests.cs
View file

@ -1,40 +0,0 @@
using System.Threading.Tasks;
using Elsa.Common.Implementations;
using Elsa.Extensions;
using Elsa.Testing.Shared;
using Elsa.Workflows.Core.Middleware.Workflows;
using Elsa.Workflows.Core.Services;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Runtime.Extensions;
using Microsoft.Extensions.DependencyInjection;
using Xunit;
using Xunit.Abstractions;
namespace Elsa.IntegrationTests.Scenarios.Persistence;
public class WorkflowInstancePersistenceTests
{
private readonly IWorkflowRunner _workflowRunner;
private readonly CapturingTextWriter _capturingTextWriter = new();
private readonly MemoryStore<WorkflowInstance> _workflowInstanceStore;
public WorkflowInstancePersistenceTests(ITestOutputHelper testOutputHelper)
{
var services = new TestApplicationBuilder(testOutputHelper).WithCapturingTextWriter(_capturingTextWriter).Build();
_workflowRunner = services.GetRequiredService<IWorkflowRunner>();
_workflowInstanceStore = services.GetRequiredService<MemoryStore<WorkflowInstance>>();
services.ConfigureDefaultWorkflowExecutionPipeline(pipeline => pipeline
.UsePersistentVariables()
.UseDefaultActivityScheduler());
}
[Fact(DisplayName = "Executing a workflow creates a workflow instance")]
public async Task Test1()
{
var result = await _workflowRunner.RunAsync<SequentialWorkflow>();
var workflowInstance = _workflowInstanceStore.Find(x => x.Id == result.WorkflowState.Id);
Assert.NotNull(workflowInstance);
}
}

21
test/Elsa.IntegrationTests/Scenarios/Persistence/Workflows.cs
View file

@ -1,21 +0,0 @@
using Elsa.Workflows.Core.Activities;
using Elsa.Workflows.Core.Services;
namespace Elsa.IntegrationTests.Scenarios.Persistence;
class SequentialWorkflow : WorkflowBase
{
protected override void Build(IWorkflowBuilder workflow)
{
workflow.WithRoot(new Sequence
{
Activities =
{
new WriteLine("Line 1"),
new Event("Resume"){ Id = "Resume"},
new WriteLine("Line 2"),
new WriteLine("Line 3")
}
});
}
}