Incremental work on activity signaling

src/core/Elsa.Core/Activities/Break.cs

@@ -1,22 +1,14 @@
 using Elsa.Attributes;
-using Elsa.Contracts;
 using Elsa.Models;
-using Microsoft.Extensions.Logging;
+using Elsa.Signals;
 
 namespace Elsa.Activities;
 
 [Activity("Elsa", "Control Flow", "Break out of a loop")]
 public class Break : Activity
 {
-    protected override void Execute(ActivityExecutionContext context)
+    protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
     {
-        // Find the first parent looping construct.
-        var loopingConstructContext = context.GetAncestorActivityExecutionContexts().FirstOrDefault(x => x.Activity is ILoopingConstruct);
-        var loopingConstructActivity = (ILoopingConstruct?)loopingConstructContext?.Activity;
-        
-        if (loopingConstructActivity != null)
-        {
-            
-        }
+        await context.SignalAsync(new BreakSignal());
     }
 }

src/core/Elsa.Core/Activities/Container.cs

@@ -2,6 +2,7 @@ using System.Collections.ObjectModel;
 using Elsa.Attributes;
 using Elsa.Contracts;
 using Elsa.Models;
+using Elsa.Signals;
 
 namespace Elsa.Activities;
 
@@ -12,27 +13,52 @@ public abstract class Container : Activity, IContainer
 {
     protected Container()
     {
+        OnSignalReceived<ActivityCompleted>(OnChildActivityCompletedAsync);
     }
 
-    protected Container(params IActivity[] activities) => Activities = activities;
-    
-    protected Container(ICollection<Variable> variables, params IActivity[] activities)
+    protected Container(params IActivity[] activities) : this()
+    {
+        Activities = activities;
+    }
+
+    protected Container(ICollection<Variable> variables, params IActivity[] activities) : this(activities)
     {
         Variables = variables;
-        Activities = activities;
     }
 
     [Outbound] public ICollection<IActivity> Activities { get; set; } = new List<IActivity>();
     public ICollection<Variable> Variables { get; set; } = new Collection<Variable>();
 
-    protected override void Execute(ActivityExecutionContext context)
+    protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
     {
         // Register variables.
         context.ExpressionExecutionContext.Register.Declare(Variables);
 
         // Schedule children.
-        ScheduleChildren(context);
+        await ScheduleChildrenAsync(context);
     }
 
-    protected abstract void ScheduleChildren(ActivityExecutionContext context);
+    protected virtual async ValueTask OnChildActivityCompletedAsync(ActivityCompleted signal, SignalContext context)
+    {
+        var activityExecutionContext = context.ActivityExecutionContext;
+        var ownerActivity = activityExecutionContext.Activity;
+        var childActivityExecutionContext = context.SourceActivityExecutionContext;
+        var childActivity = childActivityExecutionContext.Activity;
+        var callbackEntry = activityExecutionContext.WorkflowExecutionContext.CompletionCallbacks.FirstOrDefault(x => x.Owner.Activity == ownerActivity && x.Child == childActivity);
+
+        if (callbackEntry == null)
+            return;
+
+        await callbackEntry.CompletionCallback(activityExecutionContext, childActivityExecutionContext);
+    }
+
+    protected virtual ValueTask ScheduleChildrenAsync(ActivityExecutionContext context)
+    {
+        ScheduleChildren(context);
+        return ValueTask.CompletedTask;
+    }
+
+    protected virtual void ScheduleChildren(ActivityExecutionContext context)
+    {
+    }
 }

src/core/Elsa.Core/Activities/For.cs

@@ -1,6 +1,8 @@
+using System.Text.Json.Serialization;
 using Elsa.Attributes;
 using Elsa.Contracts;
 using Elsa.Models;
+using Elsa.Signals;
 
 namespace Elsa.Activities;
 
@@ -15,11 +17,13 @@ public enum ForOperator
 [Activity("Elsa", "Control Flow", "Iterate over a sequence of steps between a start and an end number.")]
 public class For : Activity
 {
+    [JsonConstructor]
     public For()
     {
+        OnSignalReceived<BreakSignal>(OnBreak);
     }
 
-    public For(int start, int end, ForOperator forOperator = ForOperator.LessThanOrEqual)
+    public For(int start, int end, ForOperator forOperator = ForOperator.LessThanOrEqual) : this()
     {
         Start = new Input<int>(start);
         End = new Input<int>(end);
@@ -79,4 +83,9 @@ public class For : Activity
         HandleIteration(ownerActivityExecutionContext);
         return ValueTask.CompletedTask;
     }
+    
+    private void OnBreak(BreakSignal signal, SignalContext context)
+    {
+        context.StopPropagation();
+    }
 }

src/core/Elsa.Core/Activities/ForEach.cs

@@ -2,6 +2,7 @@ using System.Text.Json.Serialization;
 using Elsa.Attributes;
 using Elsa.Contracts;
 using Elsa.Models;
+using Elsa.Signals;
 
 namespace Elsa.Activities;
 
@@ -9,7 +10,12 @@ namespace Elsa.Activities;
 public class ForEach : Activity
 {
     private const string CurrentIndexProperty = "CurrentIndex";
-    
+
+    public ForEach()
+    {
+        OnSignalReceived<BreakSignal>(OnBreak);
+    }
+
     /// <summary>
     /// The set of values to iterate.
     /// </summary>
@@ -57,6 +63,11 @@ public class ForEach : Activity
         HandleIteration(context);
         return ValueTask.CompletedTask;
     }
+    
+    private void OnBreak(BreakSignal signal, SignalContext context)
+    {
+        context.StopPropagation();
+    }
 }
 
 public class ForEach<T> : ForEach
@@ -66,7 +77,7 @@ public class ForEach<T> : ForEach
     {
     }
 
-    public ForEach(Input<ICollection<T>> items)
+    public ForEach(Input<ICollection<T>> items) : this()
     {
         Items = items;
     }

src/core/Elsa.Core/Activities/Sequence.cs

@@ -2,6 +2,7 @@ using System.ComponentModel;
 using Elsa.Attributes;
 using Elsa.Contracts;
 using Elsa.Models;
+using Elsa.Signals;
 
 namespace Elsa.Activities;
 
@@ -23,27 +24,29 @@ public class Sequence : Container
     {
     }
 
-    protected override void ScheduleChildren(ActivityExecutionContext context)
+    protected override async ValueTask ScheduleChildrenAsync(ActivityExecutionContext context)
     {
-        HandleItem(context);
+        await HandleItemAsync(context);
     }
 
-    private void HandleItem(ActivityExecutionContext context)
+    private async ValueTask HandleItemAsync(ActivityExecutionContext context)
     {
         var currentIndex = context.GetProperty<int>(CurrentIndexProperty);
         var childActivities = Activities.ToList();
             
         if (currentIndex >= childActivities.Count)
+        {
+            await context.SignalAsync(new ActivityCompleted());
             return;
+        }
             
         var nextActivity = childActivities.ElementAt(currentIndex);
         context.PostActivity(nextActivity, OnChildCompleted);
         context.UpdateProperty<int>(CurrentIndexProperty, x => x + 1);
     }
 
-    private ValueTask OnChildCompleted(ActivityExecutionContext context, ActivityExecutionContext childContext)
+    private async ValueTask OnChildCompleted(ActivityExecutionContext context, ActivityExecutionContext childContext)
     {
-        HandleItem(context);
-        return ValueTask.CompletedTask;
+        await HandleItemAsync(context);
     }
 }

src/core/Elsa.Core/Activities/While.cs

@@ -2,16 +2,24 @@ using System.Text.Json.Serialization;
 using Elsa.Attributes;
 using Elsa.Contracts;
 using Elsa.Models;
+using Elsa.Signals;
 
 namespace Elsa.Activities;
 
 [Activity("Elsa", "Primitives", "Execute an activity while a given condition evaluates to true.")]
 public class While : Activity
 {
+    public static While True(IActivity body) => new(body)
+    {
+        Condition = new Input<bool>(true)
+    };
+    
     [JsonConstructor]
     public While(IActivity? body = default)
     {
         Body = body!;
+        
+        OnSignalReceived<BreakSignal>(OnBreak);
     }
 
     public While(Input<bool> condition, IActivity? body = default) : this(body)
@@ -36,7 +44,7 @@ public class While : Activity
     }
 
     [Input] public Input<bool> Condition { get; set; } = new(false);
-    [Outbound] public IActivity Body { get; set; } = default!;
+    [Outbound] public IActivity Body { get; set; }
 
     protected override void Execute(ActivityExecutionContext context)
     {
@@ -53,9 +61,9 @@ public class While : Activity
         if (loop)
             context.PostActivity(Body, OnBodyCompleted);
     }
-
-    public static While True(IActivity body) => new(body)
+    
+    private void OnBreak(BreakSignal signal, SignalContext context)
     {
-        Condition = new Input<bool>(true)
-    };
+        context.StopPropagation();
+    }
 }

src/core/Elsa.Core/Contracts/ILoopingConstruct.cs

@@ -1,6 +0,0 @@
-namespace Elsa.Contracts;
-
-public interface ILoopingConstruct : IActivity
-{
-    
-}

src/core/Elsa.Core/Contracts/ISignalHandler.cs

@@ -0,0 +1,28 @@
+using Elsa.Models;
+
+namespace Elsa.Contracts;
+
+public interface ISignalHandler : IActivity
+{
+    ValueTask HandleSignalAsync(object signal, SignalContext context);
+}
+
+public class SignalContext
+{
+    public SignalContext(ActivityExecutionContext activityExecutionContext, ActivityExecutionContext sourceActivityExecutionContext, CancellationToken cancellationToken)
+    {
+        ActivityExecutionContext = activityExecutionContext;
+        SourceActivityExecutionContext = sourceActivityExecutionContext;
+        CancellationToken = cancellationToken;
+    }
+
+    public ActivityExecutionContext ActivityExecutionContext { get; init; }
+    public ActivityExecutionContext SourceActivityExecutionContext { get; init; }
+    public CancellationToken CancellationToken { get; init; }
+    internal bool StopPropagationRequested { get; private set; }
+
+    /// <summary>
+    /// Stops the signal from propagating further up the activity execution context hierarchy.
+    /// </summary>
+    public void StopPropagation() => StopPropagationRequested = true;
+}

src/core/Elsa.Core/Models/Activity.cs

@@ -1,11 +1,13 @@
 using System.Linq.Expressions;
 using Elsa.Contracts;
 using Elsa.Helpers;
+using Elsa.Signals;
 
 namespace Elsa.Models;
 
-public abstract class Activity : IActivity
+public abstract class Activity : ISignalHandler
 {
+    private readonly ICollection<SignalHandlerRegistration> _signalHandlers = new List<SignalHandlerRegistration>();
     protected Activity() => TypeName = TypeNameHelper.GenerateTypeName(GetType());
     protected Activity(string activityType) => TypeName = activityType;
 
@@ -15,17 +17,58 @@ public abstract class Activity : IActivity
     public IDictionary<string, object> ApplicationProperties { get; set; } = new Dictionary<string, object>();
     public IDictionary<string, object> Metadata { get; set; } = new Dictionary<string, object>();
 
-    protected virtual ValueTask ExecuteAsync(ActivityExecutionContext context)
+    protected virtual async ValueTask ExecuteAsync(ActivityExecutionContext context)
     {
         Execute(context);
+        await OnExecutedAsync(context);
+    }
+
+    protected virtual async ValueTask OnExecutedAsync(ActivityExecutionContext context)
+    {
+        // By default, signal that the activity is completed.
+        await context.SignalAsync(new ActivityCompleted());
+    }
+
+    protected virtual ValueTask OnSignalReceivedAsync(object signal, SignalContext context)
+    {
+        OnSignalReceived(signal, context);
         return ValueTask.CompletedTask;
     }
 
+    protected virtual void OnSignalReceived(object signal, SignalContext context)
+    {
+    }
+
     protected virtual void Execute(ActivityExecutionContext context)
     {
     }
 
+    protected void OnSignalReceived(Type signalType, Func<object, SignalContext, ValueTask> handler) => _signalHandlers.Add(new SignalHandlerRegistration(signalType, handler));
+    protected void OnSignalReceived<T>(Func<T, SignalContext, ValueTask> handler) => OnSignalReceived(typeof(T), (signal, context) => handler((T)signal, context));
+    
+    protected void OnSignalReceived<T>(Action<T, SignalContext> handler)
+    {
+        OnSignalReceived<T>((signal, context) =>
+        {
+            handler(signal, context);
+            return ValueTask.CompletedTask;
+        });
+    }
+
     ValueTask IActivity.ExecuteAsync(ActivityExecutionContext context) => ExecuteAsync(context);
+    
+    async ValueTask ISignalHandler.HandleSignalAsync(object signal, SignalContext context)
+    {
+        // Give derived activity a chance to do something with the signal.
+        await OnSignalReceivedAsync(signal, context);
+
+        // Invoke registered signal delegates for this particular type of signal.
+        var signalType = signal.GetType();
+        var handlers = _signalHandlers.Where(x => x.SignalType == signalType);
+
+        foreach (var registration in handlers)
+            await registration.Handler(signal, context);
+    }
 }
 
 public abstract class ActivityWithResult : Activity
@@ -63,7 +106,7 @@ public abstract class Activity<T> : ActivityWithResult
 
 public static class ActivityWithResultExtensions
 {
-    public static T CaptureOutput<T>(this T activity, Expression<Func<T, Output>> propertyExpression, RegisterLocationReference locationReference) where T:IActivity
+    public static T CaptureOutput<T>(this T activity, Expression<Func<T, Output>> propertyExpression, RegisterLocationReference locationReference) where T : IActivity
     {
         var output = activity.GetPropertyValue(propertyExpression)!;
         output.Targets.Add(locationReference);
@@ -71,4 +114,6 @@ public static class ActivityWithResultExtensions
     }
 
     public static T CaptureOutput<T>(this T activity, RegisterLocationReference locationReference) where T : ActivityWithResult => activity.CaptureOutput(x => x.Result, locationReference);
-}
+}
+
+internal record SignalHandlerRegistration(Type SignalType, Func<object, SignalContext, ValueTask> Handler);

src/core/Elsa.Core/Models/ActivityExecutionContext.cs

@@ -1,4 +1,6 @@
 using System.Collections.ObjectModel;
+using System.Reflection;
+using Elsa.Activities;
 using Elsa.Contracts;
 
 namespace Elsa.Models;
@@ -167,6 +169,27 @@ public class ActivityExecutionContext
     /// Stops further execution of the workflow.
     /// </summary>
     public void PreventContinuation() => Continue = false;
+    
+    /// <summary>
+    /// Send a signal up the current branch.
+    /// </summary>
+    public async ValueTask SignalAsync(object signal)
+    {
+        var ancestorContexts = GetAncestorActivityExecutionContexts();
+        
+        foreach (var ancestorContext in ancestorContexts)
+        {
+            var signalContext = new SignalContext(ancestorContext, this, CancellationToken);
+
+            if (ancestorContext.Activity is not ISignalHandler handler) 
+                continue;
+            
+            await handler.HandleSignalAsync(signal, signalContext);
+
+            if (signalContext.StopPropagationRequested)
+                return;
+        }
+    }
 
     /// <summary>
     /// Returns a flattened list of the current context's ancestors.
@@ -182,7 +205,7 @@ public class ActivityExecutionContext
             current = current.ParentActivityExecutionContext;
         }
     }
-    
+
     private RegisterLocation? GetLocation(RegisterLocationReference locationReference) =>
         ExpressionExecutionContext.Register.TryGetLocation(locationReference.Id, out var location)
             ? location

src/core/Elsa.Core/Pipelines/ActivityExecution/Components/ActivityInvokerMiddleware.cs

@@ -1,10 +1,7 @@
-using System.Text.Json;
 using Elsa.Contracts;
 using Elsa.Models;
 using Microsoft.Extensions.Logging;
 using Delegate = System.Delegate;
-using JsonNode = System.Text.Json.Nodes.JsonNode;
-using JsonObject = System.Text.Json.Nodes.JsonObject;
 
 namespace Elsa.Pipelines.ActivityExecution.Components;
 
@@ -62,9 +59,6 @@ public class ActivityInvokerMiddleware : IActivityExecutionMiddleware
             // Block current path of execution.
             return;
         }
-
-        // Complete parent chain.
-        await CompleteParentsAsync(context);
     }
 
     private void LogExecutionRecord(ActivityExecutionContext context, string eventName, string? message = default, string? source = default, object? payload = default) => context.AddExecutionLogEntry(eventName, message, source, payload);
@@ -84,42 +78,4 @@ public class ActivityInvokerMiddleware : IActivityExecutionMiddleware
             locationReference.Set(context, value);
         }
     }
-
-    private async Task CompleteParentsAsync(ActivityExecutionContext context)
-    {
-        var workflowExecutionContext = context.WorkflowExecutionContext;
-        var currentContext = context;
-        var currentParentContext = context.ParentActivityExecutionContext;
-
-        while (currentParentContext != null)
-        {
-            var scheduledNodes = workflowExecutionContext.Scheduler.List().Select(x => x.ActivityId).ToList();
-            var descendantNodes = currentParentContext.ActivityNode.Descendants().Select(x => x.Activity.Id).Distinct().ToList();
-            var hasScheduledChildren = scheduledNodes.Intersect(descendantNodes).Any();
-            var @continue = currentContext.Continue;
-
-            // Do not continue if the activity instructed not to.
-            if (!@continue)
-                return;
-
-            if (!hasScheduledChildren)
-            {
-                // Invoke completion callbacks.
-                var completionCallback = workflowExecutionContext.PopCompletionCallback(currentParentContext, currentContext.Activity);
-
-                if (completionCallback != null)
-                    await completionCallback.Invoke(currentParentContext, currentContext);
-
-                // Remove current activity context.
-                workflowExecutionContext.ActivityExecutionContexts.Remove(currentContext);
-            }
-
-            // Do not continue completion callbacks of parents while there are scheduled nodes.
-            if (workflowExecutionContext.Scheduler.HasAny)
-                return;
-
-            currentContext = currentParentContext;
-            currentParentContext = currentContext.ParentActivityExecutionContext;
-        }
-    }
 }

src/core/Elsa.Core/Signals/ActivityCompleted.cs

@@ -0,0 +1,3 @@
+namespace Elsa.Signals;
+
+public record ActivityCompleted;

src/core/Elsa.Core/Signals/BreakSignal.cs

@@ -0,0 +1,3 @@
+namespace Elsa.Signals;
+
+public record BreakSignal;

src/samples/console/Elsa.Samples.Console1/Program.cs

@@ -50,7 +50,7 @@ class Program
         var workflow14 = new Func<IActivity>(FlowchartWorkflow.Create);
         var workflow15 = new Func<IActivity>(BreakForWorkflow.Create);
 
-        var workflowFactory = workflow15;
+        var workflowFactory = workflow2;
         var workflowGraph = workflowFactory();
         var workflow = Workflow.FromActivity(workflowGraph);
 

test/Elsa.IntegrationTests/Workflows/SequentialWorkflowTests.cs

@@ -0,0 +1,59 @@
+using System.Linq;
+using System.Threading.Tasks;
+using Elsa.Activities;
+using Elsa.Builders;
+using Elsa.Contracts;
+using Elsa.Models;
+using Elsa.Modules.Activities.Activities.Console;
+using Elsa.Testing.Shared;
+using Microsoft.Extensions.DependencyInjection;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Elsa.IntegrationTests.Workflows;
+
+public class SequentialWorkflowTests
+{
+    private readonly IWorkflowRunner _workflowRunner;
+    private readonly CapturingTextWriter _capturingTextWriter = new();
+    private readonly Workflow _workflow;
+
+    public SequentialWorkflowTests(ITestOutputHelper testOutputHelper)
+    {
+        var services = new TestApplicationBuilder(testOutputHelper).WithCapturingTextWriter(_capturingTextWriter).Build();
+        _workflowRunner = services.GetRequiredService<IWorkflowRunner>();
+        _workflow = new WorkflowDefinitionBuilder().BuildWorkflow(new SequentialWorkflow());
+    }
+
+    [Fact(DisplayName = "Sequence completes only after its child activities complete")]
+    public async Task Test1()
+    {
+        await _workflowRunner.RunAsync(_workflow);
+        var lines = _capturingTextWriter.Lines.ToList();
+        Assert.Equal(new[] { "Start", "Line 1", "Line 2", "Line 3", "End" }, lines);
+    }
+
+    private class SequentialWorkflow : IWorkflow
+    {
+        public void Build(IWorkflowDefinitionBuilder workflow)
+        {
+            workflow.WithRoot(new Sequence
+            {
+                Activities =
+                {
+                    new WriteLine("Start"),
+                    new Sequence
+                    {
+                        Activities =
+                        {
+                            new WriteLine("Line 1"),
+                            new WriteLine("Line 2"),
+                            new WriteLine("Line 3")
+                        }
+                    },
+                    new WriteLine("End"),
+                }
+            });
+        }
+    }
+}