SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Add activity completion functionality to multiple contexts

Browse source 
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.
This commit is contained in:
Sipke Schoorstra 2024-04-01 20:46:18 +02:00
parent 59a6fa719a
commit 37e220050d
8 changed files with 172 additions and 131 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
src/modules/Elsa.Workflows.Core/Contexts/ActivityCompletedContext.cs
View file

@ -1,3 +1,6 @@
using System.Diagnostics.CodeAnalysis;
using Elsa.Workflows.Activities.Flowchart.Models;
namespace Elsa.Workflows;
/// <summary>
@ -21,4 +24,19 @@ public record ActivityCompletedContext(ActivityExecutionContext TargetContext, A
/// A cancellation token to use when invoking asynchronous operations.
/// </summary>
public CancellationToken CancellationToken => WorkflowExecutionContext.CancellationTokens.ApplicationCancellationToken;
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
[RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")]
public async ValueTask CompleteActivityAsync(object? result = default)
{
await TargetContext.CompleteActivityAsync(result);
}
/// <summary>
/// Complete the current activity with the specified outcomes.
/// </summary>
[RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")]
public ValueTask CompleteActivityWithOutcomesAsync(params string[] outcomes) => CompleteActivityAsync(new Outcomes(outcomes));
}

8
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs
View file

@ -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();

109
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs Normal file
View file

@ -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
{
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
[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<ISafeSerializer>();
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<IVariablePersistenceManager>();
await variablePersistenceManager.DeleteVariablesAsync(this);
}
// Update the completed at timestamp.
CompletedAt = WorkflowExecutionContext.SystemClock.UtcNow;
}
/// <summary>
/// Complete the current activity with the specified outcomes.
/// </summary>
[RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")]
public ValueTask CompleteActivityWithOutcomesAsync(params string[] outcomes)
{
return CompleteActivityAsync(new Outcomes(outcomes));
}
/// <summary>
/// Complete the current composite activity with the specified outcome.
/// </summary>
public async ValueTask CompleteCompositeAsync(params string[] outcomes)
{
await this.SendSignalAsync(new CompleteCompositeSignal(new Outcomes(outcomes)));
}
}

28
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs
View file

@ -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();
}
/// <summary>
/// Gets or sets the exception that occurred during the activity execution, if any.
/// </summary>
@ -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();

109
src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs
View file

@ -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
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
public static async ValueTask CompleteActivityAsync(this ActivityCompletedContext context, object? result = default)
{
await context.TargetContext.CompleteActivityAsync(result);
}
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
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<ISafeSerializer>();
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<IVariablePersistenceManager>();
await variablePersistenceManager.DeleteVariablesAsync(context);
}
// Update the completed at timestamp.
context.CompletedAt = context.WorkflowExecutionContext.SystemClock.UtcNow;
}
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
[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));
}
/// <summary>
/// Complete the current activity with the specified outcome.
/// </summary>
public static ValueTask CompleteActivityWithOutcomesAsync(this ActivityCompletedContext context, params string[] outcomes) => context.CompleteActivityAsync(new Outcomes(outcomes));
/// <summary>
/// Complete the current activity with the specified outcome.
/// </summary>
public static ValueTask CompleteActivityWithOutcomesAsync(this ActivityExecutionContext context, params string[] outcomes) => context.CompleteActivityAsync(new Outcomes(outcomes));
/// <summary>
/// Complete the current composite activity with the specified outcome.
/// </summary>
public static async ValueTask CompleteCompositeAsync(this ActivityExecutionContext context, params string[] outcomes) => await context.SendSignalAsync(new CompleteCompositeSignal(new Outcomes(outcomes)));
/// <summary>
/// Cancel the activity. For blocking activities, it means their bookmarks will be removed. For job activities, the background work will be cancelled.
/// </summary>

6
src/modules/Elsa.Workflows.Core/Models/CreateBookmarkArgs.cs
View file

@ -16,14 +16,14 @@ public class CreateBookmarkArgs
/// <summary>An optional name to associate with the bookmark.</summary>
public string? BookmarkName { get; set; }
/// <summary>Whether or not the bookmark should be automatically burned when triggered.</summary>
/// <summary>Whether the bookmark should be automatically burned when triggered.</summary>
public bool AutoBurn { get; set; } = true;
/// <summary>Whether or not the activity instance ID should be included in the bookmark payload.</summary>
/// <summary>Whether the activity instance ID should be included in the bookmark payload.</summary>
public bool IncludeActivityInstanceId { get; set; }
/// <summary>
/// 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.
/// </summary>
public bool AutoComplete { get; set; } = true;

11
src/modules/Elsa.Workflows.Core/Services/BookmarkHasher.cs
View file

@ -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
}
/// <inheritdoc />
[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);
}
}

14
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs
View file

@ -276,10 +276,10 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
/// <inheritdoc />
public async Task<ICollection<WorkflowExecutionResult>> 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
});
}