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

Post workflow suspension tasks improvements

Browse source
This commit is contained in:
Sipke Schoorstra 2021-03-16 11:41:22 +01:00
parent 994e55abe1
commit 17e072a74e
4 changed files with 29 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/activities/Elsa.Activities.Temporal.Common/ActivityResults/ScheduleWorkflowResult.cs
View file

@ -30,7 +30,7 @@ namespace Elsa.Activities.Temporal.Common.ActivityResults
await scheduler.ScheduleWorkflowAsync(null, workflowInstanceId, activityId, tenantId, executeAt, null, cancellationToken);
}
activityExecutionContext.WorkflowExecutionContext.RegisterTask(nameof(ScheduleWorkflowResult), ScheduleWorkflowAsync);
activityExecutionContext.WorkflowExecutionContext.RegisterTask(ScheduleWorkflowAsync);
}
}
}

16
src/activities/Elsa.Activities.Temporal.Common/Handlers/ScheduleWorkflows.cs
View file

@ -1,16 +0,0 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.Temporal.Common.ActivityResults;
using Elsa.Events;
using MediatR;
namespace Elsa.Activities.Temporal.Common.Handlers
{
public class ScheduleWorkflows : INotificationHandler<WorkflowSuspended>
{
public async Task Handle(WorkflowSuspended notification, CancellationToken cancellationToken)
{
await notification.WorkflowExecutionContext.ExecuteRegisteredTasksAsync(nameof(ScheduleWorkflowResult), cancellationToken);
}
}
}

39
src/core/Elsa.Abstractions/Services/Models/WorkflowExecutionContext.cs
View file

@ -76,9 +76,11 @@ namespace Elsa.Services.Models
public object? WorkflowContext { get; set; }
/// <summary>
/// A collection of tasks to execute post-workflow burst execution.
/// A collection of tasks to execute after the workflow is suspended.
/// This is useful for avoiding race conditions, such as sending a message and then waiting for a message to be received using some MessageReceived activity for example. If the workflow didn't get suspended while that message is received, the workflow would get stuck.
/// By designing activities such as `SendMessage` to only actually send the message post-suspension using the Tasks collection, you can be sure that the workflow gets suspended and blocked on the `MessageReceived` activity before a message reply gets received.
/// </summary>
public IDictionary<string, ICollection<Func<WorkflowExecutionContext, CancellationToken, ValueTask>>> Tasks { get; set; } = new Dictionary<string, ICollection<Func<WorkflowExecutionContext, CancellationToken, ValueTask>>>();
public ICollection<Func<WorkflowExecutionContext, CancellationToken, ValueTask>> Tasks { get; private set; } = new List<Func<WorkflowExecutionContext, CancellationToken, ValueTask>>();
public string? ContextId
{
@ -86,30 +88,15 @@ namespace Elsa.Services.Models
set => WorkflowInstance.ContextId = value;
}
public void RegisterTask(string groupName, Func<WorkflowExecutionContext, CancellationToken, ValueTask> task)
{
if (!Tasks.ContainsKey(groupName))
{
var list = new List<Func<WorkflowExecutionContext, CancellationToken, ValueTask>> { task };
Tasks[groupName] = list;
}
else
{
Tasks[groupName].Add(task);
}
}
public void RegisterTask(Func<WorkflowExecutionContext, CancellationToken, ValueTask> task) => Tasks.Add(task);
public IEnumerable<Func<WorkflowExecutionContext, CancellationToken, ValueTask>> GetRegisteredTasks(string groupName) =>
Tasks.ContainsKey(groupName) ? Tasks[groupName] : Enumerable.Empty<Func<WorkflowExecutionContext, CancellationToken, ValueTask>>();
public async ValueTask ExecuteRegisteredTasksAsync(string groupName, CancellationToken cancellationToken = default)
public async ValueTask ProcessRegisteredTasksAsync(CancellationToken cancellationToken = default)
{
var tasks = GetRegisteredTasks(groupName);
var tasks = Tasks.ToList();
Tasks = new List<Func<WorkflowExecutionContext, CancellationToken, ValueTask>>();
foreach (var task in tasks)
await task(this, cancellationToken);
Tasks.Remove(groupName);
}
public async Task RemoveBlockingActivityAsync(BlockingActivity blockingActivity)
@ -123,7 +110,7 @@ namespace Elsa.Services.Models
public void SetVariable(string name, object? value) => WorkflowInstance.Variables.Set(name, value);
public T? GetVariable<T>() => GetVariable<T>(typeof(T).Name);
public T? GetVariable<T>(string name) => WorkflowInstance.Variables.Get<T>(name);
/// <summary>
/// Gets a variable from across all scopes, starting with the current scope, going up each scope until the requested variable is found.
/// </summary>
@ -131,16 +118,16 @@ namespace Elsa.Services.Models
public object? GetVariable(string name)
{
var scopes = WorkflowInstance.Scopes.ToList();
var mergedVariables = scopes.Select(x => x.Variables).Aggregate(new Variables(), (current, next) =>
{
var combined = current.Data.MergedWith(next.Data);
return new Variables(combined);
});
return mergedVariables.Get(name);
}
/// <summary>
/// Gets a workflow variable.
/// </summary>
@ -155,7 +142,7 @@ namespace Elsa.Services.Models
public ActivityScope CurrentScope => WorkflowInstance.Scopes.Peek();
public ActivityScope GetScope(string activityId) => WorkflowInstance.Scopes.First(x => x.ActivityId == activityId);
public ActivityScope GetNamedScope(string activityName)
{
var activityBlueprint = GetActivityBlueprintByName(activityName)!;

15
src/core/Elsa.Core/Handlers/ExecuteWorkflowsPostSuspensionTasks.cs Normal file
View file

@ -0,0 +1,15 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Events;
using MediatR;
namespace Elsa.Handlers
{
public class ExecuteWorkflowsPostSuspensionTasks : INotificationHandler<WorkflowSuspended>
{
public async Task Handle(WorkflowSuspended notification, CancellationToken cancellationToken)
{
await notification.WorkflowExecutionContext.ProcessRegisteredTasksAsync(cancellationToken);
}
}
}