diff --git a/src/modules/Elsa.Workflows.Core/Activities/ParallelForEachT.cs b/src/modules/Elsa.Workflows.Core/Activities/ParallelForEachT.cs index 9aa4c311c..cef5648b5 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/ParallelForEachT.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/ParallelForEachT.cs @@ -2,6 +2,7 @@ using System.Runtime.CompilerServices; using System.Text.Json; using System.Text.Json.Nodes; using Elsa.Expressions.Helpers; +using Elsa.Expressions.Models; using Elsa.Extensions; using Elsa.Workflows.Attributes; using Elsa.Workflows.Memory; @@ -19,6 +20,27 @@ public class ParallelForEach : Activity private const string ScheduledTagsProperty = nameof(ScheduledTagsProperty); private const string CompletedTagsProperty = nameof(CompletedTagsProperty); + /// + public ParallelForEach(Func> @delegate, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : this(new Input(@delegate), source, line) + { + } + + /// + public ParallelForEach(Func> @delegate, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : this(new Input(@delegate), source, line) + { + } + + /// + public ParallelForEach(ICollection items, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : this(new Input(items), source, line) + { + } + + /// + public ParallelForEach(Input items, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : this(source, line) + { + Items = items; + } + /// public ParallelForEach([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line) { @@ -42,7 +64,7 @@ public class ParallelForEach : Activity var items = context.GetItemSource(Items); var tags = new List(); var currentIndex = 0; - + await foreach (var item in items) { // For each item, declare a new variable for the work to be scheduled. diff --git a/test/integration/Elsa.Activities.IntegrationTests/ParallelForEachTests.cs b/test/integration/Elsa.Activities.IntegrationTests/ParallelForEachTests.cs new file mode 100644 index 000000000..c7e6f972d --- /dev/null +++ b/test/integration/Elsa.Activities.IntegrationTests/ParallelForEachTests.cs @@ -0,0 +1,135 @@ +using Elsa.Extensions; +using Elsa.Testing.Shared; +using Elsa.Workflows; +using Elsa.Workflows.Activities; +using Elsa.Workflows.Models; +using Xunit.Abstractions; + +namespace Elsa.Activities.IntegrationTests; + +public class ParallelForEachTests(ITestOutputHelper testOutputHelper) +{ + private readonly WorkflowTestFixture _fixture = new(testOutputHelper); + private const string CurrentValueVar = "CurrentValue"; + private static readonly string[] ThreeItems = ["a", "b", "c"]; + + private CapturingTextWriter CapturingTextWriter => _fixture.CapturingTextWriter; + + [Theory(DisplayName = "ParallelForEach executes body for all items")] + [MemberData(nameof(ItemTestCases))] + public async Task ParallelForEach_ExecutesBody_ForAllItems(string[] items) + { + await ExecuteAndAssertAllItems(items); + } + + [Fact(DisplayName = "ParallelForEach completes when collection is empty")] + public async Task ParallelForEach_Completes_WhenCollectionEmpty() + { + await ExecuteAndAssertStatus([], ActivityStatus.Completed); + Assert.Empty(CapturingTextWriter.Lines); + } + + [Fact(DisplayName = "ParallelForEach completes when collection is null")] + public async Task ParallelForEach_Completes_WhenCollectionNull() + { + await ExecuteAndAssertStatus(null, ActivityStatus.Completed); + } + + [Fact(DisplayName = "ParallelForEach executes all items when one faults")] + public async Task ParallelForEach_ExecutesAllItems_WhenOneFaults() + { + var body = new Sequence + { + Activities = + [ + new If(context => context.GetVariable(CurrentValueVar) == "b") + { + Then = new Fault { Message = new("Faulted") }, + }, + WriteCurrentValue() + ] + }; + + await RunActivityAsync(ThreeItems, body); + + Assert.Contains("a", CapturingTextWriter.Lines); + Assert.DoesNotContain("b", CapturingTextWriter.Lines); + Assert.Contains("c", CapturingTextWriter.Lines); + } + + [Fact(DisplayName = "ParallelForEach executes body for different item types")] + public async Task ParallelForEach_ExecutesBody_ForDifferentItemTypes() + { + var items = new object?[] + { + "a", 2, null, new Foo() + }; + var parallelForEach = new ParallelForEach(items) + { + Body = new WriteLine(context => context.GetVariable(CurrentValueVar)?.ToString() ?? "") + }; + + await _fixture.RunActivityAsync(parallelForEach); + + Assert.Equal(items.Length, CapturingTextWriter.Lines.Count); + Assert.Contains("a", CapturingTextWriter.Lines); + Assert.Contains("2", CapturingTextWriter.Lines); + Assert.Contains("", CapturingTextWriter.Lines); + Assert.Contains("Baz", CapturingTextWriter.Lines); + } + + [Fact(DisplayName = "ParallelForEach provides CurrentIndex variable")] + public async Task ParallelForEach_ProvidesCurrentIndex_ForEachIteration() + { + var body = new WriteLine(context => context.GetVariable("CurrentIndex").ToString()); + + await RunActivityAsync(ThreeItems, body); + + Assert.Equal(ThreeItems.Length, CapturingTextWriter.Lines.Count); + Assert.Contains("0", CapturingTextWriter.Lines); + Assert.Contains("1", CapturingTextWriter.Lines); + Assert.Contains("2", CapturingTextWriter.Lines); + } + + [Fact(DisplayName = "ParallelForEach completes when all bodies complete")] + public async Task ParallelForEach_Completes_WhenAllBodiesComplete() + { + await ExecuteAndAssertStatus(ThreeItems, ActivityStatus.Completed); + } + + public static TheoryData ItemTestCases => + [ + ThreeItems, + ["single"] + ]; + + private async Task ExecuteAndAssertAllItems(string[] items) + { + await RunActivityAsync(items, WriteCurrentValue()); + + Assert.Equal(items.Length, CapturingTextWriter.Lines.Count); + Assert.All(items, item => Assert.Contains(item, CapturingTextWriter.Lines)); + } + + private async Task ExecuteAndAssertStatus(string[]? items, ActivityStatus expectedStatus) + { + var result = await RunActivityAsync(items, WriteCurrentValue()); + var context = GetParallelForEachContext(result); + Assert.Equal(expectedStatus, context.Status); + } + + private async Task RunActivityAsync(string[]? items, IActivity body) + { + var parallelForEach = new ParallelForEach(items!) { Body = body }; + return await _fixture.RunActivityAsync(parallelForEach); + } + + private static WriteLine WriteCurrentValue() => new(context => context.GetVariable(CurrentValueVar)); + + private static ActivityExecutionContext GetParallelForEachContext(RunWorkflowResult result) + { + var context = result.Journal.ActivityExecutionContexts.FirstOrDefault(x => x.Activity is ParallelForEach); + Assert.NotNull(context); + return context; + } +} \ No newline at end of file diff --git a/test/integration/Elsa.Activities.IntegrationTests/ParallelTests.cs b/test/integration/Elsa.Activities.IntegrationTests/ParallelTests.cs index bfcec4cac..01e0d5d2d 100644 --- a/test/integration/Elsa.Activities.IntegrationTests/ParallelTests.cs +++ b/test/integration/Elsa.Activities.IntegrationTests/ParallelTests.cs @@ -1,7 +1,6 @@ using Elsa.Testing.Shared; using Elsa.Workflows; using Elsa.Workflows.Activities; -using Elsa.Workflows.Models; using Xunit.Abstractions; using Parallel = Elsa.Workflows.Activities.Parallel; diff --git a/test/unit/Elsa.Activities.UnitTests/Looping/ParallelForEachTests.cs b/test/unit/Elsa.Activities.UnitTests/Looping/ParallelForEachTests.cs new file mode 100644 index 000000000..05a94ed1e --- /dev/null +++ b/test/unit/Elsa.Activities.UnitTests/Looping/ParallelForEachTests.cs @@ -0,0 +1,94 @@ +using Elsa.Testing.Shared; +using Elsa.Workflows; + +namespace Elsa.Activities.UnitTests.Looping; + +public class ParallelForEachTests +{ + [Theory] + [InlineData(1)] + [InlineData(2)] + [InlineData(3)] + public async Task Should_Schedule_Body_For_Each_Item(int itemCount) + { + var items = Enumerable.Range(0, itemCount).Select(i => $"item{i}").ToArray(); + var body = new MockBodyActivity(); + var parallelForEach = new ParallelForEach(items) { Body = body }; + + var context = await ExecuteAsync(parallelForEach); + + AssertScheduledCount(context, itemCount); + Assert.True(context.HasScheduledActivity(body)); + } + + [Fact] + public async Task Should_Complete_When_Items_Empty() + { + var body = new MockBodyActivity(); + var parallelForEach = new ParallelForEach(Array.Empty()) { Body = body }; + + var context = await ExecuteAsync(parallelForEach); + + Assert.Equal(ActivityStatus.Completed, context.Status); + Assert.False(context.HasScheduledActivity(body)); + } + + [Fact] + public async Task Should_Not_Schedule_When_Body_Null() + { + var items = new[] { "a", "b" }; + var parallelForEach = new ParallelForEach(items) { Body = null! }; + + var context = await ExecuteAsync(parallelForEach); + + AssertScheduledCount(context, 0); + } + + [Fact] + public async Task Should_Handle_Integer_Items() + { + var items = new[] { 1, 2, 3 }; + var body = new MockBodyActivity(); + var parallelForEach = new ParallelForEach(items) { Body = body }; + + var context = await ExecuteAsync(parallelForEach); + + AssertScheduledCount(context, items.Length); + } + + [Fact] + public void Verify_Activity_Attributes() + { + var parallelForEach = new ParallelForEach(); + var fixture = new ActivityTestFixture(parallelForEach); + + fixture.AssertActivityAttributes( + expectedNamespace: "Elsa", + expectedKind: ActivityKind.Action, + expectedCategory: "Looping", + expectedDescription: "Schedule an activity for each item in parallel." + ); + } + + [Fact] + public void Default_Property_Values() + { + var parallelForEach = new ParallelForEach(); + + Assert.NotNull(parallelForEach.Items); + } + + private static Task ExecuteAsync(IActivity activity) => + new ActivityTestFixture(activity).ExecuteAsync(); + + private static void AssertScheduledCount(ActivityExecutionContext context, int expectedCount) + { + var scheduledActivities = context.WorkflowExecutionContext.Scheduler.List().ToList(); + Assert.Equal(expectedCount, scheduledActivities.Count); + } + + private class MockBodyActivity : Activity + { + protected override ValueTask ExecuteAsync(ActivityExecutionContext context) => ValueTask.CompletedTask; + } +}