| | | 1 | | using System.Runtime.CompilerServices; |
| | | 2 | | using System.Text.Json; |
| | | 3 | | using System.Text.Json.Nodes; |
| | | 4 | | using Elsa.Expressions.Helpers; |
| | | 5 | | using Elsa.Expressions.Models; |
| | | 6 | | using Elsa.Extensions; |
| | | 7 | | using Elsa.Workflows.Attributes; |
| | | 8 | | using Elsa.Workflows.Memory; |
| | | 9 | | using Elsa.Workflows.Models; |
| | | 10 | | |
| | | 11 | | namespace Elsa.Workflows.Activities; |
| | | 12 | | |
| | | 13 | | /// <summary> |
| | | 14 | | /// Schedule an activity for each item in parallel. |
| | | 15 | | /// </summary> |
| | | 16 | | /// <typeparam name="T"></typeparam> |
| | | 17 | | [Activity("Elsa", "Looping", "Schedule an activity for each item in parallel.")] |
| | | 18 | | public class ParallelForEach<T> : Activity |
| | | 19 | | { |
| | | 20 | | private const string ScheduledTagsProperty = nameof(ScheduledTagsProperty); |
| | | 21 | | private const string CompletedTagsProperty = nameof(CompletedTagsProperty); |
| | | 22 | | |
| | | 23 | | /// <inheritdoc /> |
| | 0 | 24 | | public ParallelForEach(Func<ExpressionExecutionContext, ICollection<T>> @delegate, [CallerFilePath] string? source = |
| | | 25 | | { |
| | 0 | 26 | | } |
| | | 27 | | |
| | | 28 | | /// <inheritdoc /> |
| | 0 | 29 | | public ParallelForEach(Func<ICollection<T>> @delegate, [CallerFilePath] string? source = null, [CallerLineNumber] in |
| | | 30 | | { |
| | 0 | 31 | | } |
| | | 32 | | |
| | | 33 | | /// <inheritdoc /> |
| | 15 | 34 | | public ParallelForEach(ICollection<T> items, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = |
| | | 35 | | { |
| | 15 | 36 | | } |
| | | 37 | | |
| | | 38 | | /// <inheritdoc /> |
| | 15 | 39 | | public ParallelForEach(Input<object> items, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = n |
| | | 40 | | { |
| | 15 | 41 | | Items = items; |
| | 15 | 42 | | } |
| | | 43 | | |
| | | 44 | | /// <inheritdoc /> |
| | 17 | 45 | | public ParallelForEach([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, l |
| | | 46 | | { |
| | 17 | 47 | | } |
| | | 48 | | |
| | | 49 | | /// <summary> |
| | | 50 | | /// The items to iterate. |
| | | 51 | | /// </summary> |
| | | 52 | | [Input(Description = "The items to iterate through.")] |
| | 71 | 53 | | public Input<object> Items { get; set; } = new(Array.Empty<T>()); |
| | | 54 | | |
| | | 55 | | /// <summary> |
| | | 56 | | /// The <see cref="IActivity"/> to execute each iteration. |
| | | 57 | | /// </summary> |
| | | 58 | | [Port] |
| | 64 | 59 | | public IActivity Body { get; set; } = null!; |
| | | 60 | | |
| | | 61 | | /// <inheritdoc /> |
| | | 62 | | protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) |
| | | 63 | | { |
| | 16 | 64 | | var items = context.GetItemSource<T>(Items); |
| | 16 | 65 | | var tags = new List<Guid>(); |
| | 16 | 66 | | var currentIndex = 0; |
| | | 67 | | |
| | 100 | 68 | | await foreach (var item in items) |
| | | 69 | | { |
| | | 70 | | // Schedule a body of work for each item. |
| | 34 | 71 | | var tag = Guid.NewGuid(); |
| | 34 | 72 | | tags.Add(tag); |
| | | 73 | | |
| | | 74 | | // Give each iteration its own variable ids so nested loops cannot share a parent memory block. |
| | 34 | 75 | | var currentValueVariable = new Variable<T>("CurrentValue", item, $"{tag}:CurrentValue") |
| | 34 | 76 | | { |
| | 34 | 77 | | // TODO: This should be configurable, because this won't work for e.g. file streams and other non-serial |
| | 34 | 78 | | StorageDriverType = typeof(WorkflowInstanceStorageDriver) |
| | 34 | 79 | | }; |
| | | 80 | | |
| | 34 | 81 | | var currentIndexVariable = new Variable<int>("CurrentIndex", currentIndex++, $"{tag}:CurrentIndex") |
| | 34 | 82 | | { |
| | 34 | 83 | | StorageDriverType = typeof(WorkflowInstanceStorageDriver) |
| | 34 | 84 | | }; |
| | 34 | 85 | | var variables = new List<Variable> { currentValueVariable, currentIndexVariable }; |
| | | 86 | | |
| | 34 | 87 | | await context.ScheduleActivityAsync(Body, OnChildCompleted, tag, variables); |
| | | 88 | | } |
| | | 89 | | |
| | 16 | 90 | | SetTagList(context, ScheduledTagsProperty, tags); |
| | 16 | 91 | | SetTagList(context, CompletedTagsProperty, new List<Guid>()); |
| | | 92 | | |
| | | 93 | | // If there were no items, we're done. |
| | 16 | 94 | | if (tags.Count == 0) |
| | 3 | 95 | | await context.CompleteActivityAsync(); |
| | 16 | 96 | | } |
| | | 97 | | |
| | | 98 | | private async ValueTask OnChildCompleted(ActivityCompletedContext context) |
| | | 99 | | { |
| | 22 | 100 | | var targetContext = context.TargetContext; |
| | 22 | 101 | | var scheduledTags = GetTagList(targetContext, ScheduledTagsProperty); |
| | 22 | 102 | | var completedTag = targetContext.Tag.ConvertTo<Guid>(); |
| | 22 | 103 | | var completedTags = GetTagList(targetContext, CompletedTagsProperty); |
| | | 104 | | |
| | 22 | 105 | | completedTags.Add(completedTag); |
| | 22 | 106 | | SetTagList(targetContext, CompletedTagsProperty, completedTags); |
| | | 107 | | |
| | | 108 | | // If not all scheduled activities have completed yet, we're not done yet. |
| | 22 | 109 | | if (!scheduledTags.IsEqualTo(completedTags)) |
| | 15 | 110 | | return; |
| | | 111 | | |
| | | 112 | | // We're done, so complete the activity. |
| | 7 | 113 | | await targetContext.CompleteActivityAsync(); |
| | 22 | 114 | | } |
| | | 115 | | |
| | | 116 | | private ICollection<Guid> GetTagList(ActivityExecutionContext context, string propertyName) |
| | | 117 | | { |
| | | 118 | | // Read the list of tags from the context using the specified property name. The value is stored as JsonArray, s |
| | 44 | 119 | | var jsonArray = context.GetProperty<JsonArray>(propertyName)!; |
| | 134 | 120 | | return jsonArray.Select(x => x.ConvertTo<Guid>()).ToList(); |
| | | 121 | | } |
| | | 122 | | |
| | | 123 | | private void SetTagList(ActivityExecutionContext context, string propertyName, ICollection<Guid> tags) |
| | | 124 | | { |
| | | 125 | | // Serialize the list of tags to a JsonArray and store it in the context using the specified property name. |
| | 54 | 126 | | var jsonArray = JsonSerializer.SerializeToNode(tags) as JsonArray; |
| | 54 | 127 | | context.SetProperty(propertyName, jsonArray); |
| | 54 | 128 | | } |
| | | 129 | | } |