From e4f9b7717b6f0035aef006894c3ea3baec1b60fc Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Sun, 30 Oct 2022 23:44:38 +0100 Subject: [PATCH] Initial support for activity cancellation --- .../Elsa.Workflows.Core/Activities/Break.cs | 2 +- .../Behaviors/BreakBehavior.cs | 11 ++++- .../ActivityExecutionContextExtensions.cs | 13 +++++- .../Models/ActivityExecutionContext.cs | 2 +- .../Models/WorkflowExecutionContext.cs | 15 ++++++- .../Signals/CancelSignal.cs | 3 ++ .../Activities/Fork/ForkTests.cs | 2 +- .../Scenarios/BlockingAndBreaking/Tests.cs | 2 +- .../Scenarios/Persistence/Tests.cs | 40 ------------------- .../Scenarios/Persistence/Workflows.cs | 21 ---------- 10 files changed, 41 insertions(+), 70 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Core/Signals/CancelSignal.cs delete mode 100644 test/Elsa.IntegrationTests/Scenarios/Persistence/Tests.cs delete mode 100644 test/Elsa.IntegrationTests/Scenarios/Persistence/Workflows.cs diff --git a/src/modules/Elsa.Workflows.Core/Activities/Break.cs b/src/modules/Elsa.Workflows.Core/Activities/Break.cs index 60bda0b53..50e249b7d 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Break.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Break.cs @@ -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()); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Behaviors/BreakBehavior.cs b/src/modules/Elsa.Workflows.Core/Behaviors/BreakBehavior.cs index a0b3946e3..2e825e972 100644 --- a/src/modules/Elsa.Workflows.Core/Behaviors/BreakBehavior.cs +++ b/src/modules/Elsa.Workflows.Core/Behaviors/BreakBehavior.cs @@ -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(); } diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs index 87a493f24..b42f58b66 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs @@ -128,7 +128,7 @@ public static class ActivityExecutionContextExtensions /// /// Send a signal up the current branch. /// - 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); } + + /// + /// Cancel the activity. For blocking activities, it means their bookmarks will be removed. For job activities, the background work will be cancelled. + /// + 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())); diff --git a/src/modules/Elsa.Workflows.Core/Models/ActivityExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Models/ActivityExecutionContext.cs index 47d4c64ea..b11c32923 100644 --- a/src/modules/Elsa.Workflows.Core/Models/ActivityExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Models/ActivityExecutionContext.cs @@ -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 diff --git a/src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs index f21f6b526..35c8200f7 100644 --- a/src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs @@ -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 GetMergedRegistersView() => new[] { MemoryRegister }.Concat(ActivityExecutionContexts.Select(x => x.ExpressionExecutionContext.Memory)).ToList(); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Signals/CancelSignal.cs b/src/modules/Elsa.Workflows.Core/Signals/CancelSignal.cs new file mode 100644 index 000000000..cd4fe1a32 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Signals/CancelSignal.cs @@ -0,0 +1,3 @@ +namespace Elsa.Workflows.Core.Signals; + +public record CancelSignal; \ No newline at end of file diff --git a/test/Elsa.IntegrationTests/Activities/Fork/ForkTests.cs b/test/Elsa.IntegrationTests/Activities/Fork/ForkTests.cs index ca4d7a646..4e8147a7d 100644 --- a/test/Elsa.IntegrationTests/Activities/Fork/ForkTests.cs +++ b/test/Elsa.IntegrationTests/Activities/Fork/ForkTests.cs @@ -28,7 +28,7 @@ public class ForkTests var workflow = await _workflowBuilderFactory.CreateBuilder().BuildWorkflowAsync(); 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")] diff --git a/test/Elsa.IntegrationTests/Scenarios/BlockingAndBreaking/Tests.cs b/test/Elsa.IntegrationTests/Scenarios/BlockingAndBreaking/Tests.cs index 89a0054bb..4f97ccfd5 100644 --- a/test/Elsa.IntegrationTests/Scenarios/BlockingAndBreaking/Tests.cs +++ b/test/Elsa.IntegrationTests/Scenarios/BlockingAndBreaking/Tests.cs @@ -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); } } \ No newline at end of file diff --git a/test/Elsa.IntegrationTests/Scenarios/Persistence/Tests.cs b/test/Elsa.IntegrationTests/Scenarios/Persistence/Tests.cs deleted file mode 100644 index 2519d91e6..000000000 --- a/test/Elsa.IntegrationTests/Scenarios/Persistence/Tests.cs +++ /dev/null @@ -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 _workflowInstanceStore; - - public WorkflowInstancePersistenceTests(ITestOutputHelper testOutputHelper) - { - var services = new TestApplicationBuilder(testOutputHelper).WithCapturingTextWriter(_capturingTextWriter).Build(); - _workflowRunner = services.GetRequiredService(); - _workflowInstanceStore = services.GetRequiredService>(); - - services.ConfigureDefaultWorkflowExecutionPipeline(pipeline => pipeline - .UsePersistentVariables() - .UseDefaultActivityScheduler()); - } - - [Fact(DisplayName = "Executing a workflow creates a workflow instance")] - public async Task Test1() - { - var result = await _workflowRunner.RunAsync(); - var workflowInstance = _workflowInstanceStore.Find(x => x.Id == result.WorkflowState.Id); - - Assert.NotNull(workflowInstance); - } -} \ No newline at end of file diff --git a/test/Elsa.IntegrationTests/Scenarios/Persistence/Workflows.cs b/test/Elsa.IntegrationTests/Scenarios/Persistence/Workflows.cs deleted file mode 100644 index 8f7ed005a..000000000 --- a/test/Elsa.IntegrationTests/Scenarios/Persistence/Workflows.cs +++ /dev/null @@ -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") - } - }); - } -} \ No newline at end of file