From 37e220050d9e68c6d2ff6962fa15db0336bfb871 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 1 Apr 2024 20:46:18 +0200 Subject: [PATCH] Add activity completion functionality to multiple contexts This commit introduces multiple methods to handle activity completion across various contexts, including ActivityExecutionContext and ActivityCompletedContext. It also includes updates to bookmark serialization and the WorkflowRuntime. The resulting changes should improve handling of activity outcomes and status updates in the application flow. --- .../Contexts/ActivityCompletedContext.cs | 18 +++ .../ActivityExecutionContext.Cancel.cs | 8 ++ .../ActivityExecutionContext.Complete.cs | 109 ++++++++++++++++++ .../Contexts/ActivityExecutionContext.cs | 28 ++--- .../ActivityExecutionContextExtensions.cs | 109 +----------------- .../Models/CreateBookmarkArgs.cs | 6 +- .../Services/BookmarkHasher.cs | 11 +- .../Services/DefaultWorkflowRuntime.cs | 14 +-- 8 files changed, 172 insertions(+), 131 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityCompletedContext.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityCompletedContext.cs index 1c40937ea..078be3bea 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/ActivityCompletedContext.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityCompletedContext.cs @@ -1,3 +1,6 @@ +using System.Diagnostics.CodeAnalysis; +using Elsa.Workflows.Activities.Flowchart.Models; + namespace Elsa.Workflows; /// @@ -21,4 +24,19 @@ public record ActivityCompletedContext(ActivityExecutionContext TargetContext, A /// A cancellation token to use when invoking asynchronous operations. /// public CancellationToken CancellationToken => WorkflowExecutionContext.CancellationTokens.ApplicationCancellationToken; + + /// + /// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion. + /// + [RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")] + public async ValueTask CompleteActivityAsync(object? result = default) + { + await TargetContext.CompleteActivityAsync(result); + } + + /// + /// Complete the current activity with the specified outcomes. + /// + [RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")] + public ValueTask CompleteActivityWithOutcomesAsync(params string[] outcomes) => CompleteActivityAsync(new Outcomes(outcomes)); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs index 082935672..808577679 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs @@ -19,9 +19,17 @@ public partial class ActivityExecutionContext _ = Task.Run(async () => await CancelActivityAsync()); } + + private bool CanCancelActivity() + { + return Status is not ActivityStatus.Canceled and not ActivityStatus.Completed; + } private async Task CancelActivityAsync() { + if(!CanCancelActivity()) + return; + // Select all child contexts. var childContexts = WorkflowExecutionContext.ActivityExecutionContexts.Where(x => x.ParentActivityExecutionContext == this).ToList(); diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs new file mode 100644 index 000000000..f3963ed91 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs @@ -0,0 +1,109 @@ +using System.Diagnostics.CodeAnalysis; +using Elsa.Extensions; +using Elsa.Workflows.Activities.Flowchart.Models; +using Elsa.Workflows.Contracts; +using Elsa.Workflows.Signals; + +namespace Elsa.Workflows; + +public partial class ActivityExecutionContext +{ + /// + /// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion. + /// + [RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")] + public async ValueTask CompleteActivityAsync(object? result = default) + { + var outcomes = result as Outcomes; + + // If the activity is executing in the background, simply capture the result and return. + if (this.GetIsBackgroundExecution()) + { + if (outcomes != null) + this.SetBackgroundOutcomes(outcomes.Names); + else + this.SetBackgroundCompletion(); + return; + } + + // If the activity is not running, do nothing. + if (Status != ActivityStatus.Running) + return; + + // Cancel any non-completed child activities. + var childContexts = WorkflowExecutionContext.ActivityExecutionContexts.Where(x => x.ParentActivityExecutionContext == this && x.CanCancelActivity()).ToList(); + + foreach (var childContext in childContexts) + await childContext.CancelActivityAsync(); + + // Mark the activity as complete. + TransitionTo(ActivityStatus.Completed); + + // Record the outcomes, if any. + if (outcomes != null) + JournalData["Outcomes"] = outcomes.Names; + + // Record the output, if any. + var activity = Activity; + var expressionExecutionContext = ExpressionExecutionContext; + var activityDescriptor = ActivityDescriptor; + var outputDescriptors = activityDescriptor.Outputs; + var outputs = outputDescriptors.ToDictionary(x => x.Name, x => activity.GetOutput(expressionExecutionContext, x.Name)!); + var serializer = GetRequiredService(); + + foreach (var outputDescriptor in outputDescriptors) + { + if (outputDescriptor.IsSerializable == false) + continue; + + var outputName = outputDescriptor.Name; + var outputValue = outputs[outputName]; + + if (outputValue == null!) + continue; + + var serializedOutputValue = await serializer.SerializeAsync(outputValue, CancellationToken); + JournalData[outputName] = serializedOutputValue; + } + + // Add an execution log entry. + AddExecutionLogEntry("Completed", payload: JournalData); + + // Send a signal. + await this.SendSignalAsync(new ActivityCompleted(result)); + + // Clear bookmarks. + ClearBookmarks(); + WorkflowExecutionContext.Bookmarks.RemoveWhere(x => x.ActivityInstanceId == Id); + + // Remove completion callbacks. + ClearCompletionCallbacks(); + + // Remove all associated variables, unless this is the root context - in which case we want to keep the variables since we're not deleting that one. + if (ParentActivityExecutionContext != null) + { + var variablePersistenceManager = GetRequiredService(); + await variablePersistenceManager.DeleteVariablesAsync(this); + } + + // Update the completed at timestamp. + CompletedAt = WorkflowExecutionContext.SystemClock.UtcNow; + } + + /// + /// Complete the current activity with the specified outcomes. + /// + [RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")] + public ValueTask CompleteActivityWithOutcomesAsync(params string[] outcomes) + { + return CompleteActivityAsync(new Outcomes(outcomes)); + } + + /// + /// Complete the current composite activity with the specified outcome. + /// + public async ValueTask CompleteCompositeAsync(params string[] outcomes) + { + await this.SendSignalAsync(new CompleteCompositeSignal(new Outcomes(outcomes))); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs index 2fc1d4d94..2be9c0b3d 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs @@ -136,13 +136,11 @@ public partial class ActivityExecutionContext : IExecutionContext public void TransitionTo(ActivityStatus status) { Status = status; - - if (Status is ActivityStatus.Completed - or ActivityStatus.Canceled - or ActivityStatus.Faulted) + + if (Status is ActivityStatus.Completed or ActivityStatus.Canceled or ActivityStatus.Faulted) _cancellationRegistration.Dispose(); } - + /// /// Gets or sets the exception that occurred during the activity execution, if any. /// @@ -254,15 +252,17 @@ public partial class ActivityExecutionContext : IExecutionContext { ActivityNodeId = activityNode?.NodeId, OwnerActivityInstanceId = owner?.Id, - Options = options != null ? new ScheduledActivityOptions - { - CompletionCallback = options?.CompletionCallback?.Method.Name, - Tag = options?.Tag, - ExistingActivityInstanceId = options?.ExistingActivityExecutionContext?.Id, - PreventDuplicateScheduling = options?.PreventDuplicateScheduling ?? false, - Variables = options?.Variables?.ToList(), - Input = options?.Input - } : default + Options = options != null + ? new ScheduledActivityOptions + { + CompletionCallback = options?.CompletionCallback?.Method.Name, + Tag = options?.Tag, + ExistingActivityInstanceId = options?.ExistingActivityExecutionContext?.Id, + PreventDuplicateScheduling = options?.PreventDuplicateScheduling ?? false, + Variables = options?.Variables?.ToList(), + Input = options?.Input + } + : default }; var scheduledActivities = this.GetBackgroundScheduledActivities().ToList(); diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs index 4d934abcf..5bcc6f3c1 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs @@ -1,3 +1,4 @@ +using System.Diagnostics.CodeAnalysis; using System.Linq.Expressions; using System.Reflection; using System.Text.Json; @@ -6,7 +7,6 @@ using Elsa.Expressions.Helpers; using Elsa.Expressions.Models; using Elsa.Mediator.Contracts; using Elsa.Workflows; -using Elsa.Workflows.Activities.Flowchart.Models; using Elsa.Workflows.Attributes; using Elsa.Workflows.Contracts; using Elsa.Workflows.Memory; @@ -406,95 +406,7 @@ public static class ActivityExecutionContextExtensions /// /// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion. /// - public static async ValueTask CompleteActivityAsync(this ActivityCompletedContext context, object? result = default) - { - await context.TargetContext.CompleteActivityAsync(result); - } - - /// - /// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion. - /// - public static async ValueTask CompleteActivityAsync(this ActivityExecutionContext context, object? result = default) - { - var outcomes = result as Outcomes; - - // If the activity is executing in the background, simply capture the result and return. - if (context.GetIsBackgroundExecution()) - { - if (outcomes != null) - context.SetBackgroundOutcomes(outcomes.Names); - else - context.SetBackgroundCompletion(); - return; - } - - // If the activity is not running, do nothing. - if (context.Status != ActivityStatus.Running) - return; - - // Update all child contexts. - var childContexts = context.WorkflowExecutionContext.ActivityExecutionContexts.Where(x => x.ParentActivityExecutionContext == context).ToList(); - - foreach (var childContext in childContexts) - await childContext.CancelActivityAsync(); - - // Mark the activity as complete. - context.TransitionTo(ActivityStatus.Completed); - - // Record the outcomes, if any. - if (outcomes != null) - context.JournalData["Outcomes"] = outcomes.Names; - - // Record the output, if any. - var activity = context.Activity; - var expressionExecutionContext = context.ExpressionExecutionContext; - var activityDescriptor = context.ActivityDescriptor; - var outputDescriptors = activityDescriptor.Outputs; - var outputs = outputDescriptors.ToDictionary(x => x.Name, x => activity.GetOutput(expressionExecutionContext, x.Name)!); - var serializer = context.GetRequiredService(); - - foreach (var outputDescriptor in outputDescriptors) - { - if (outputDescriptor.IsSerializable == false) - continue; - - var outputName = outputDescriptor.Name; - var outputValue = outputs[outputName]; - - if (outputValue == null!) - continue; - - var serializedOutputValue = await serializer.SerializeAsync(outputValue); - context.JournalData[outputName] = serializedOutputValue; - } - - // Add an execution log entry. - context.AddExecutionLogEntry("Completed", payload: context.JournalData); - - // Send a signal. - await context.SendSignalAsync(new ActivityCompleted(result)); - - // Clear bookmarks. - context.ClearBookmarks(); - context.WorkflowExecutionContext.Bookmarks.RemoveWhere(x => x.ActivityInstanceId == context.Id); - - // Remove completion callbacks. - context.ClearCompletionCallbacks(); - - // Remove all associated variables, unless this is the root context - in which case we want to keep the variables since we're not deleting that one. - if (context.ParentActivityExecutionContext != null) - { - var variablePersistenceManager = context.GetRequiredService(); - await variablePersistenceManager.DeleteVariablesAsync(context); - } - - // Update the completed at timestamp. - context.CompletedAt = context.WorkflowExecutionContext.SystemClock.UtcNow; - } - - /// - /// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion. - /// + [RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")] public static async ValueTask ScheduleOutcomesAsync(this ActivityExecutionContext context, params string[] outcomes) { var cancellationToken = context.CancellationToken; @@ -525,22 +437,7 @@ public static class ActivityExecutionContextExtensions // Send a signal. await context.SendSignalAsync(new ScheduleActivityOutcomes(outcomes)); } - - /// - /// Complete the current activity with the specified outcome. - /// - public static ValueTask CompleteActivityWithOutcomesAsync(this ActivityCompletedContext context, params string[] outcomes) => context.CompleteActivityAsync(new Outcomes(outcomes)); - - /// - /// Complete the current activity with the specified outcome. - /// - public static ValueTask CompleteActivityWithOutcomesAsync(this ActivityExecutionContext context, params string[] outcomes) => context.CompleteActivityAsync(new Outcomes(outcomes)); - - /// - /// Complete the current composite activity with the specified outcome. - /// - public static async ValueTask CompleteCompositeAsync(this ActivityExecutionContext context, params string[] outcomes) => await context.SendSignalAsync(new CompleteCompositeSignal(new Outcomes(outcomes))); - + /// /// Cancel the activity. For blocking activities, it means their bookmarks will be removed. For job activities, the background work will be cancelled. /// diff --git a/src/modules/Elsa.Workflows.Core/Models/CreateBookmarkArgs.cs b/src/modules/Elsa.Workflows.Core/Models/CreateBookmarkArgs.cs index aef490975..422e4f2c6 100644 --- a/src/modules/Elsa.Workflows.Core/Models/CreateBookmarkArgs.cs +++ b/src/modules/Elsa.Workflows.Core/Models/CreateBookmarkArgs.cs @@ -16,14 +16,14 @@ public class CreateBookmarkArgs /// An optional name to associate with the bookmark. public string? BookmarkName { get; set; } - /// Whether or not the bookmark should be automatically burned when triggered. + /// Whether the bookmark should be automatically burned when triggered. public bool AutoBurn { get; set; } = true; - /// Whether or not the activity instance ID should be included in the bookmark payload. + /// Whether the activity instance ID should be included in the bookmark payload. public bool IncludeActivityInstanceId { get; set; } /// - /// Whether or not the activity being resumed should be automatically completed if CallBack is not specified. + /// Whether the activity being resumed should be automatically completed if CallBack is not specified. /// public bool AutoComplete { get; set; } = true; diff --git a/src/modules/Elsa.Workflows.Core/Services/BookmarkHasher.cs b/src/modules/Elsa.Workflows.Core/Services/BookmarkHasher.cs index 997c70296..7c14eb238 100644 --- a/src/modules/Elsa.Workflows.Core/Services/BookmarkHasher.cs +++ b/src/modules/Elsa.Workflows.Core/Services/BookmarkHasher.cs @@ -1,3 +1,4 @@ +using System.Diagnostics.CodeAnalysis; using System.Text.Json; using Elsa.Expressions.Contracts; using Elsa.Workflows.Contracts; @@ -30,6 +31,7 @@ public class BookmarkHasher : IBookmarkHasher } /// + [RequiresUnreferencedCode("Calls System.Text.Json.JsonSerializer.Serialize(Object, Type, JsonSerializerOptions)")] public string Hash(string activityTypeName, object? payload, string? activityInstanceId = default) { var json = payload != null ? Serialize(payload) : null; @@ -47,5 +49,12 @@ public class BookmarkHasher : IBookmarkHasher return hash; } - private string Serialize(object payload) => JsonSerializer.Serialize(payload, payload.GetType(), _settings); + [RequiresUnreferencedCode("Calls System.Text.Json.JsonSerializer.Serialize(Object, Type, JsonSerializerOptions)")] + private string Serialize(object payload) + { + if(payload is string s) + return s; + + return JsonSerializer.Serialize(payload, payload.GetType(), _settings); + } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs index 93a9795d7..05d9754c8 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs @@ -276,10 +276,10 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime /// public async Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options) { - var hash = _hasher.Hash(activityTypeName, bookmarkPayload, options.ActivityInstanceId); - var correlationId = options.CorrelationId; - var workflowInstanceId = options.WorkflowInstanceId; - var activityInstanceId = options.ActivityInstanceId; + var hash = _hasher.Hash(activityTypeName, bookmarkPayload, options?.ActivityInstanceId); + var correlationId = options?.CorrelationId; + var workflowInstanceId = options?.WorkflowInstanceId; + var activityInstanceId = options?.ActivityInstanceId; var filter = new BookmarkFilter { Hash = hash, @@ -287,15 +287,15 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime WorkflowInstanceId = workflowInstanceId, ActivityInstanceId = activityInstanceId }; - var bookmarks = await _bookmarkStore.FindManyAsync(filter, options.CancellationTokens.SystemCancellationToken); + var bookmarks = await _bookmarkStore.FindManyAsync(filter, options?.CancellationTokens.SystemCancellationToken ?? default); return await ResumeWorkflowsAsync( bookmarks, new ResumeWorkflowRuntimeParams { CorrelationId = correlationId, - Input = options.Input, - CancellationTokens = options.CancellationTokens + Input = options?.Input, + CancellationTokens = options?.CancellationTokens ?? default }); }