Implement delay activity resumption of bookmarks

This commit is contained in:
Sipke Schoorstra 2022-01-18 20:28:48 +01:00
parent 59f44b4927
commit bc31319f2a
30 changed files with 276 additions and 132 deletions

View file

@ -12,8 +12,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.Console1", "sr
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Core", "src\core\Elsa.Core\Elsa.Core.csproj", "{6FA9ED7D-2D32-46BC-B47F-E503B2A4C0A9}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Extensions.DependencyInjection", "src\core\Elsa.Extensions.DependencyInjection\Elsa.Extensions.DependencyInjection.csproj", "{042022EA-62BC-44BD-A93F-11ABDB39C791}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "solution", "solution", "{7D21EB5E-D6B4-48A5-A9BC-A26387A65F0B}"
ProjectSection(SolutionItems) = preProject
README.md = README.md
@ -50,16 +48,12 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Runtime", "src\runtime
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.Abstractions", "src\persistence\Elsa.Persistence.Abstractions\Elsa.Persistence.Abstractions.csproj", "{C0226CDD-DF79-4767-A4A5-3D9037535929}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Runtime.HostedServices", "src\runtime\Elsa.Runtime.HostedServices\Elsa.Runtime.HostedServices.csproj", "{4A874C29-2915-4F5C-A01A-80F56B929541}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "dsl", "dsl", "{CE121DFD-72D4-4DB7-968F-C0DB320D08FB}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Dsl", "src\dsl\Elsa.Dsl\Elsa.Dsl.csproj", "{61F5C50C-C572-46AB-8F37-54D1F84CC31D}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.Console2", "src\samples\console\Elsa.Samples.Console2\Elsa.Samples.Console2.csproj", "{48960E35-903C-4EF0-A8F4-3871B000B12D}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Timers", "src\activities\Elsa.Activities.Timers\Elsa.Activities.Timers.csproj", "{9FE7C169-F639-4F6D-86EB-D97FE3E9D206}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "scripting", "scripting", "{2633B8B9-4AEC-4A54-8832-1C940B54E041}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Scripting.JavaScript", "src\scripting\Elsa.Scripting.JavaScript\Elsa.Scripting.JavaScript.csproj", "{5D349665-3B5D-4DA4-A7B0-C6F11611B0C9}"
@ -80,10 +74,22 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Proto.Remote", "..\..\Proto
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Proto.Remote.GrpcCore", "..\..\ProtoActor\protoactor-dotnet\src\Proto.Remote.GrpcCore\Proto.Remote.GrpcCore.csproj", "{466E5047-4184-42C6-9C4D-14A5707E2F49}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Api.Core", "src\api\Elsa.Api.Core\Elsa.Api.Core.csproj", "{A335B51D-8219-4361-BAAF-5F5D1B74EF89}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa", "src\core\Elsa\Elsa.csproj", "{521365AE-A681-4E4B-87E6-B4D982CDD520}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Scheduling", "src\activities\Elsa.Activities.Scheduling\Elsa.Activities.Scheduling.csproj", "{33D18969-2522-4D82-86DF-74668CBA2AB7}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Management", "src\core\Elsa.Management\Elsa.Management.csproj", "{DE817AE7-F45E-415B-91AD-42DD73F462C9}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Mediator", "src\core\Elsa.Mediator\Elsa.Mediator.csproj", "{B4FB10B4-08FF-4D38-A721-4F786C00E446}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Scheduling", "src\core\Elsa.Scheduling\Elsa.Scheduling.csproj", "{877F4392-6F0A-49CA-88A6-9A390DAE8E59}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Serialization", "src\core\Elsa.Serialization\Elsa.Serialization.csproj", "{ECF6407B-BCEE-4367-B6B3-C058C45E1D5C}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.EntityFrameworkCore", "src\persistence\Elsa.Persistence.EntityFrameworkCore\Elsa.Persistence.EntityFrameworkCore.csproj", "{6BE75B1F-9CB5-4D07-8129-4FC9E30CB597}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.EntityFrameworkCore.Sqlite", "src\persistence\Elsa.Persistence.EntityFrameworkCore.Sqlite\Elsa.Persistence.EntityFrameworkCore.Sqlite.csproj", "{81DBB365-F608-4C19-961B-6D9D3A141A24}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -98,10 +104,6 @@ Global
{6FA9ED7D-2D32-46BC-B47F-E503B2A4C0A9}.Debug|Any CPU.Build.0 = Debug|Any CPU
{6FA9ED7D-2D32-46BC-B47F-E503B2A4C0A9}.Release|Any CPU.ActiveCfg = Release|Any CPU
{6FA9ED7D-2D32-46BC-B47F-E503B2A4C0A9}.Release|Any CPU.Build.0 = Release|Any CPU
{042022EA-62BC-44BD-A93F-11ABDB39C791}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{042022EA-62BC-44BD-A93F-11ABDB39C791}.Debug|Any CPU.Build.0 = Debug|Any CPU
{042022EA-62BC-44BD-A93F-11ABDB39C791}.Release|Any CPU.ActiveCfg = Release|Any CPU
{042022EA-62BC-44BD-A93F-11ABDB39C791}.Release|Any CPU.Build.0 = Release|Any CPU
{A2E6E9A4-4C21-4022-A53E-BDF5ED0D0678}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{A2E6E9A4-4C21-4022-A53E-BDF5ED0D0678}.Debug|Any CPU.Build.0 = Debug|Any CPU
{A2E6E9A4-4C21-4022-A53E-BDF5ED0D0678}.Release|Any CPU.ActiveCfg = Release|Any CPU
@ -126,10 +128,6 @@ Global
{C0226CDD-DF79-4767-A4A5-3D9037535929}.Debug|Any CPU.Build.0 = Debug|Any CPU
{C0226CDD-DF79-4767-A4A5-3D9037535929}.Release|Any CPU.ActiveCfg = Release|Any CPU
{C0226CDD-DF79-4767-A4A5-3D9037535929}.Release|Any CPU.Build.0 = Release|Any CPU
{4A874C29-2915-4F5C-A01A-80F56B929541}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{4A874C29-2915-4F5C-A01A-80F56B929541}.Debug|Any CPU.Build.0 = Debug|Any CPU
{4A874C29-2915-4F5C-A01A-80F56B929541}.Release|Any CPU.ActiveCfg = Release|Any CPU
{4A874C29-2915-4F5C-A01A-80F56B929541}.Release|Any CPU.Build.0 = Release|Any CPU
{61F5C50C-C572-46AB-8F37-54D1F84CC31D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{61F5C50C-C572-46AB-8F37-54D1F84CC31D}.Debug|Any CPU.Build.0 = Debug|Any CPU
{61F5C50C-C572-46AB-8F37-54D1F84CC31D}.Release|Any CPU.ActiveCfg = Release|Any CPU
@ -138,10 +136,6 @@ Global
{48960E35-903C-4EF0-A8F4-3871B000B12D}.Debug|Any CPU.Build.0 = Debug|Any CPU
{48960E35-903C-4EF0-A8F4-3871B000B12D}.Release|Any CPU.ActiveCfg = Release|Any CPU
{48960E35-903C-4EF0-A8F4-3871B000B12D}.Release|Any CPU.Build.0 = Release|Any CPU
{9FE7C169-F639-4F6D-86EB-D97FE3E9D206}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{9FE7C169-F639-4F6D-86EB-D97FE3E9D206}.Debug|Any CPU.Build.0 = Debug|Any CPU
{9FE7C169-F639-4F6D-86EB-D97FE3E9D206}.Release|Any CPU.ActiveCfg = Release|Any CPU
{9FE7C169-F639-4F6D-86EB-D97FE3E9D206}.Release|Any CPU.Build.0 = Release|Any CPU
{5D349665-3B5D-4DA4-A7B0-C6F11611B0C9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{5D349665-3B5D-4DA4-A7B0-C6F11611B0C9}.Debug|Any CPU.Build.0 = Debug|Any CPU
{5D349665-3B5D-4DA4-A7B0-C6F11611B0C9}.Release|Any CPU.ActiveCfg = Release|Any CPU
@ -174,21 +168,44 @@ Global
{466E5047-4184-42C6-9C4D-14A5707E2F49}.Debug|Any CPU.Build.0 = Debug|Any CPU
{466E5047-4184-42C6-9C4D-14A5707E2F49}.Release|Any CPU.ActiveCfg = Release|Any CPU
{466E5047-4184-42C6-9C4D-14A5707E2F49}.Release|Any CPU.Build.0 = Release|Any CPU
{A335B51D-8219-4361-BAAF-5F5D1B74EF89}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{A335B51D-8219-4361-BAAF-5F5D1B74EF89}.Debug|Any CPU.Build.0 = Debug|Any CPU
{A335B51D-8219-4361-BAAF-5F5D1B74EF89}.Release|Any CPU.ActiveCfg = Release|Any CPU
{A335B51D-8219-4361-BAAF-5F5D1B74EF89}.Release|Any CPU.Build.0 = Release|Any CPU
{521365AE-A681-4E4B-87E6-B4D982CDD520}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{521365AE-A681-4E4B-87E6-B4D982CDD520}.Debug|Any CPU.Build.0 = Debug|Any CPU
{521365AE-A681-4E4B-87E6-B4D982CDD520}.Release|Any CPU.ActiveCfg = Release|Any CPU
{521365AE-A681-4E4B-87E6-B4D982CDD520}.Release|Any CPU.Build.0 = Release|Any CPU
{33D18969-2522-4D82-86DF-74668CBA2AB7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{33D18969-2522-4D82-86DF-74668CBA2AB7}.Debug|Any CPU.Build.0 = Debug|Any CPU
{33D18969-2522-4D82-86DF-74668CBA2AB7}.Release|Any CPU.ActiveCfg = Release|Any CPU
{33D18969-2522-4D82-86DF-74668CBA2AB7}.Release|Any CPU.Build.0 = Release|Any CPU
{DE817AE7-F45E-415B-91AD-42DD73F462C9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{DE817AE7-F45E-415B-91AD-42DD73F462C9}.Debug|Any CPU.Build.0 = Debug|Any CPU
{DE817AE7-F45E-415B-91AD-42DD73F462C9}.Release|Any CPU.ActiveCfg = Release|Any CPU
{DE817AE7-F45E-415B-91AD-42DD73F462C9}.Release|Any CPU.Build.0 = Release|Any CPU
{B4FB10B4-08FF-4D38-A721-4F786C00E446}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{B4FB10B4-08FF-4D38-A721-4F786C00E446}.Debug|Any CPU.Build.0 = Debug|Any CPU
{B4FB10B4-08FF-4D38-A721-4F786C00E446}.Release|Any CPU.ActiveCfg = Release|Any CPU
{B4FB10B4-08FF-4D38-A721-4F786C00E446}.Release|Any CPU.Build.0 = Release|Any CPU
{877F4392-6F0A-49CA-88A6-9A390DAE8E59}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{877F4392-6F0A-49CA-88A6-9A390DAE8E59}.Debug|Any CPU.Build.0 = Debug|Any CPU
{877F4392-6F0A-49CA-88A6-9A390DAE8E59}.Release|Any CPU.ActiveCfg = Release|Any CPU
{877F4392-6F0A-49CA-88A6-9A390DAE8E59}.Release|Any CPU.Build.0 = Release|Any CPU
{ECF6407B-BCEE-4367-B6B3-C058C45E1D5C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{ECF6407B-BCEE-4367-B6B3-C058C45E1D5C}.Debug|Any CPU.Build.0 = Debug|Any CPU
{ECF6407B-BCEE-4367-B6B3-C058C45E1D5C}.Release|Any CPU.ActiveCfg = Release|Any CPU
{ECF6407B-BCEE-4367-B6B3-C058C45E1D5C}.Release|Any CPU.Build.0 = Release|Any CPU
{6BE75B1F-9CB5-4D07-8129-4FC9E30CB597}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{6BE75B1F-9CB5-4D07-8129-4FC9E30CB597}.Debug|Any CPU.Build.0 = Debug|Any CPU
{6BE75B1F-9CB5-4D07-8129-4FC9E30CB597}.Release|Any CPU.ActiveCfg = Release|Any CPU
{6BE75B1F-9CB5-4D07-8129-4FC9E30CB597}.Release|Any CPU.Build.0 = Release|Any CPU
{81DBB365-F608-4C19-961B-6D9D3A141A24}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{81DBB365-F608-4C19-961B-6D9D3A141A24}.Debug|Any CPU.Build.0 = Debug|Any CPU
{81DBB365-F608-4C19-961B-6D9D3A141A24}.Release|Any CPU.ActiveCfg = Release|Any CPU
{81DBB365-F608-4C19-961B-6D9D3A141A24}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(NestedProjects) = preSolution
{155227F0-A33B-40AA-A4B4-06F813EB921B} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F}
{C6658DE0-2B2F-47F0-BB61-2CA66D435C09} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F}
{F8604289-6D28-428E-9FA8-1480464964F0} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F}
{6FA9ED7D-2D32-46BC-B47F-E503B2A4C0A9} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09}
{042022EA-62BC-44BD-A93F-11ABDB39C791} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09}
{A34FD350-79C5-492A-8811-6DBD80830F8B} = {0354F050-3992-4DD4-B0EE-5FBA04AC72B6}
{873BFC3E-63C2-4495-A503-5EC05DCD84E4} = {155227F0-A33B-40AA-A4B4-06F813EB921B}
{B44A9DDA-8091-490A-BC0B-2EAE4A601B09} = {873BFC3E-63C2-4495-A503-5EC05DCD84E4}
@ -202,11 +219,9 @@ Global
{04584653-8E42-47DB-850B-7C6C37B307B7} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F}
{8C5894A4-F358-45CF-B198-BFBCA30B77F6} = {04584653-8E42-47DB-850B-7C6C37B307B7}
{C0226CDD-DF79-4767-A4A5-3D9037535929} = {9B7776E2-0014-4C87-B1F4-22FB0BC076F1}
{4A874C29-2915-4F5C-A01A-80F56B929541} = {04584653-8E42-47DB-850B-7C6C37B307B7}
{CE121DFD-72D4-4DB7-968F-C0DB320D08FB} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F}
{61F5C50C-C572-46AB-8F37-54D1F84CC31D} = {CE121DFD-72D4-4DB7-968F-C0DB320D08FB}
{48960E35-903C-4EF0-A8F4-3871B000B12D} = {873BFC3E-63C2-4495-A503-5EC05DCD84E4}
{9FE7C169-F639-4F6D-86EB-D97FE3E9D206} = {F8604289-6D28-428E-9FA8-1480464964F0}
{2633B8B9-4AEC-4A54-8832-1C940B54E041} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F}
{5D349665-3B5D-4DA4-A7B0-C6F11611B0C9} = {2633B8B9-4AEC-4A54-8832-1C940B54E041}
{E5A05E1D-0F7F-4030-B434-596C76486CDB} = {04584653-8E42-47DB-850B-7C6C37B307B7}
@ -216,7 +231,13 @@ Global
{EB1BACFB-3DF0-45D0-89AF-4571023DDE50} = {03A5D774-A546-4D4B-8687-C6F96A886C72}
{0BD81F36-D6B4-4DF1-8BAA-5F25B6779FC5} = {03A5D774-A546-4D4B-8687-C6F96A886C72}
{466E5047-4184-42C6-9C4D-14A5707E2F49} = {03A5D774-A546-4D4B-8687-C6F96A886C72}
{A335B51D-8219-4361-BAAF-5F5D1B74EF89} = {298BA226-B766-40AE-933F-EB9AFCDE051C}
{521365AE-A681-4E4B-87E6-B4D982CDD520} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09}
{33D18969-2522-4D82-86DF-74668CBA2AB7} = {F8604289-6D28-428E-9FA8-1480464964F0}
{DE817AE7-F45E-415B-91AD-42DD73F462C9} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09}
{B4FB10B4-08FF-4D38-A721-4F786C00E446} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09}
{877F4392-6F0A-49CA-88A6-9A390DAE8E59} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09}
{ECF6407B-BCEE-4367-B6B3-C058C45E1D5C} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09}
{6BE75B1F-9CB5-4D07-8129-4FC9E30CB597} = {9B7776E2-0014-4C87-B1F4-22FB0BC076F1}
{81DBB365-F608-4C19-961B-6D9D3A141A24} = {9B7776E2-0014-4C87-B1F4-22FB0BC076F1}
EndGlobalSection
EndGlobal

View file

@ -0,0 +1,14 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Models;
namespace Elsa.Activities.Scheduling.Contracts;
/// <summary>
/// Schedules jobs for the specified list of workflow bookmarks.
/// </summary>
public interface IWorkflowBookmarkScheduler
{
Task ScheduleBookmarksAsync(string workflowInstanceId, IEnumerable<Bookmark> bookmarks, CancellationToken cancellationToken = default);
}

View file

@ -15,6 +15,7 @@ public static class ServiceCollectionExtensions
{
services
.AddSingleton<IWorkflowTriggerScheduler, WorkflowTriggerScheduler>()
.AddSingleton<IWorkflowBookmarkScheduler, WorkflowBookmarkScheduler>()
.AddJobHandler<RunWorkflowJobHandler>()
.AddJobHandler<ResumeWorkflowJobHandler>()
.AddNotificationHandlersFrom<ScheduleWorkflowsHandler>()

View file

@ -1,9 +1,8 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.Scheduling.Contracts;
using Elsa.Mediator.Contracts;
using Elsa.Runtime.Notifications;
using Elsa.Scheduling.Contracts;
using IWorkflowTriggerScheduler = Elsa.Activities.Scheduling.Contracts.IWorkflowTriggerScheduler;
namespace Elsa.Activities.Scheduling.Handlers;
@ -11,11 +10,21 @@ namespace Elsa.Activities.Scheduling.Handlers;
public class ScheduleWorkflowsHandler : INotificationHandler<TriggerIndexingFinished>, INotificationHandler<WorkflowExecuted>
{
private readonly IWorkflowTriggerScheduler _workflowTriggerScheduler;
public ScheduleWorkflowsHandler(IWorkflowTriggerScheduler workflowTriggerScheduler) => _workflowTriggerScheduler = workflowTriggerScheduler;
public async Task HandleAsync(TriggerIndexingFinished notification, CancellationToken cancellationToken) => await _workflowTriggerScheduler.ScheduleTriggersAsync(notification.Triggers, cancellationToken);
public Task HandleAsync(WorkflowExecuted notification, CancellationToken cancellationToken)
private readonly IWorkflowBookmarkScheduler _workflowBookmarkScheduler;
public ScheduleWorkflowsHandler(IWorkflowTriggerScheduler workflowTriggerScheduler, IWorkflowBookmarkScheduler workflowBookmarkScheduler)
{
return Task.CompletedTask;
_workflowTriggerScheduler = workflowTriggerScheduler;
_workflowBookmarkScheduler = workflowBookmarkScheduler;
}
public async Task HandleAsync(TriggerIndexingFinished notification, CancellationToken cancellationToken) => await _workflowTriggerScheduler.ScheduleTriggersAsync(notification.Triggers, cancellationToken);
public async Task HandleAsync(WorkflowExecuted notification, CancellationToken cancellationToken)
{
var workflowExecutionContext = notification.WorkflowExecutionContext;
var workflowInstanceId = workflowExecutionContext.Id;
var bookmarks = workflowExecutionContext.Bookmarks;
await _workflowBookmarkScheduler.ScheduleBookmarksAsync(workflowInstanceId, bookmarks, cancellationToken);
}
}

View file

@ -1,9 +1,10 @@
using System.Collections.Immutable;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.Scheduling.Contracts;
using Elsa.Mediator.Contracts;
using Elsa.Persistence.Requests;
using Elsa.Scheduling.Contracts;
using Microsoft.Extensions.Hosting;
using IWorkflowTriggerScheduler = Elsa.Activities.Scheduling.Contracts.IWorkflowTriggerScheduler;
@ -16,16 +17,40 @@ public class ScheduleWorkflowsHostedService : BackgroundService
{
private readonly IRequestSender _requestSender;
private readonly IWorkflowTriggerScheduler _workflowTriggerScheduler;
private readonly IWorkflowBookmarkScheduler _workflowBookmarkScheduler;
public ScheduleWorkflowsHostedService(IRequestSender requestSender, IWorkflowTriggerScheduler workflowTriggerScheduler)
public ScheduleWorkflowsHostedService(
IRequestSender requestSender,
IWorkflowTriggerScheduler workflowTriggerScheduler,
IWorkflowBookmarkScheduler workflowBookmarkScheduler)
{
_requestSender = requestSender;
_workflowTriggerScheduler = workflowTriggerScheduler;
_workflowBookmarkScheduler = workflowBookmarkScheduler;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
var timerTriggers = (await _requestSender.RequestAsync(FindWorkflowTriggers.ForTrigger<Timer>(), stoppingToken)).ToImmutableList();
await _workflowTriggerScheduler.ScheduleTriggersAsync(timerTriggers, stoppingToken);
await ScheduleTriggersAsync(stoppingToken);
await ScheduleBookmarksAsync(stoppingToken);
}
private async Task ScheduleTriggersAsync(CancellationToken cancellationToken)
{
var timerTriggers = (await _requestSender.RequestAsync(FindWorkflowTriggers.ForTrigger<Timer>(), cancellationToken)).ToImmutableList();
await _workflowTriggerScheduler.ScheduleTriggersAsync(timerTriggers, cancellationToken);
}
private async Task ScheduleBookmarksAsync(CancellationToken cancellationToken)
{
var workflowBookmarks = (await _requestSender.RequestAsync(FindWorkflowBookmarks.ForActivity<Delay>(), cancellationToken)).ToImmutableList();
var groupedBookmarks = workflowBookmarks.GroupBy(x => x.WorkflowInstanceId);
foreach (var bookmarksGroup in groupedBookmarks)
{
var workflowInstanceId = bookmarksGroup.Key;
var bookmarks = bookmarksGroup.Select(x => x.ToBookmark()).ToList();
await _workflowBookmarkScheduler.ScheduleBookmarksAsync(workflowInstanceId, bookmarks, cancellationToken);
}
}
}

View file

@ -1,6 +1,6 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.Scheduling.Contracts;
using Elsa.Models;
using Elsa.Runtime.Contracts;
using Elsa.Runtime.Models;
using Elsa.Scheduling.Abstractions;
@ -8,9 +8,9 @@ using Elsa.Scheduling.Contracts;
namespace Elsa.Activities.Scheduling.Jobs;
public record ResumeWorkflowJob(string WorkflowInstanceId, string ActivityId) : IJob
public record ResumeWorkflowJob(string WorkflowInstanceId, Bookmark Bookmark) : IJob
{
public string JobId => $"workflow-instance:{WorkflowInstanceId}-{ActivityId}";
public string JobId => $"Bookmark:{Bookmark.Id}";
}
public class ResumeWorkflowJobHandler : JobHandler<ResumeWorkflowJob>
@ -20,7 +20,7 @@ public class ResumeWorkflowJobHandler : JobHandler<ResumeWorkflowJob>
protected override async Task HandleAsync(ResumeWorkflowJob job, CancellationToken cancellationToken)
{
var request = new ExecuteWorkflowInstanceRequest(job.WorkflowInstanceId);
await _workflowInvoker.ExecuteAsync(request, cancellationToken);
var request = new DispatchWorkflowInstanceRequest(job.WorkflowInstanceId, job.Bookmark);
await _workflowInvoker.DispatchAsync(request, cancellationToken);
}
}

View file

@ -1,6 +1,5 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.Scheduling.Contracts;
using Elsa.Runtime.Contracts;
using Elsa.Runtime.Models;
using Elsa.Scheduling.Abstractions;
@ -10,7 +9,7 @@ namespace Elsa.Activities.Scheduling.Jobs;
public record RunWorkflowJob(string WorkflowId) : IJob
{
public string JobId => $"workflow:{WorkflowId}";
public string JobId => WorkflowId;
}
public class RunWorkflowJobHandler : JobHandler<RunWorkflowJob>
@ -24,7 +23,7 @@ public class RunWorkflowJobHandler : JobHandler<RunWorkflowJob>
protected override async Task HandleAsync(RunWorkflowJob job, CancellationToken cancellationToken)
{
var request = new ExecuteWorkflowDefinitionRequest(job.WorkflowId, 1);
await _workflowInvoker.ExecuteAsync(request, cancellationToken);
var request = new DispatchWorkflowDefinitionRequest(job.WorkflowId, 1);
await _workflowInvoker.DispatchAsync(request, cancellationToken);
}
}

View file

@ -0,0 +1,44 @@
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.Scheduling.Contracts;
using Elsa.Activities.Scheduling.Jobs;
using Elsa.Extensions;
using Elsa.Models;
using Elsa.Scheduling.Contracts;
using Elsa.Scheduling.Schedules;
namespace Elsa.Activities.Scheduling.Services;
public class WorkflowBookmarkScheduler : IWorkflowBookmarkScheduler
{
private const string RootGroupKey = "WorkflowInstance";
private readonly IJobScheduler _jobScheduler;
public WorkflowBookmarkScheduler(IJobScheduler jobScheduler)
{
_jobScheduler = jobScheduler;
}
public async Task ScheduleBookmarksAsync(string workflowInstanceId, IEnumerable<Bookmark> bookmarks, CancellationToken cancellationToken = default)
{
var bookmarkList = bookmarks.ToList();
// Select all Delay bookmarks.
var delayBookmarks = bookmarkList.Filter<Delay>().ToList();
// Unschedule all triggers for the distinct set of affected workflows.
var groupKeys = new[] { RootGroupKey, workflowInstanceId };
await _jobScheduler.ClearAsync(groupKeys, cancellationToken);
// Schedule a trigger for each bookmark.
foreach (var bookmark in delayBookmarks)
{
var payload = JsonSerializer.Deserialize<DelayPayload>(bookmark.Payload!)!;
var resumeAt = payload.ResumeAt;
await _jobScheduler.ScheduleAsync(new ResumeWorkflowJob(workflowInstanceId, bookmark), new SpecificInstantSchedule(resumeAt), groupKeys, cancellationToken);
}
}
}

View file

@ -0,0 +1,14 @@
using Elsa.Contracts;
using Elsa.Helpers;
using Elsa.Models;
namespace Elsa.Extensions;
public static class BookmarkExtensions
{
public static IEnumerable<Bookmark> Filter<T>(this IEnumerable<Bookmark> bookmarks) where T : IActivity
{
var bookmarkName = TypeNameHelper.GenerateTypeName<T>();
return bookmarks.Where(x => x.Name == bookmarkName);
}
}

View file

@ -1,6 +1,5 @@
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using Elsa.Contracts;
namespace Elsa.Services;

View file

@ -1,2 +1 @@
using System;
[assembly: CLSCompliant(true)]

View file

@ -1,2 +1 @@
using System;
[assembly: CLSCompliant(true)]

View file

@ -3,7 +3,4 @@ using Elsa.Scheduling.Contracts;
namespace Elsa.Scheduling.Schedules;
public class SpecificInstantSchedule : ISchedule
{
public DateTime DateTime { get; set; }
}
public record SpecificInstantSchedule(DateTime DateTime) : ISchedule;

View file

@ -1,2 +1 @@
using System;
[assembly: CLSCompliant(true)]

View file

@ -50,6 +50,7 @@ public static class ServiceCollectionExtensions
private static IServiceCollectionQuartzConfigurator AddElsaJobs(this IServiceCollectionQuartzConfigurator quartz)
{
quartz.AddJob<RunWorkflowJob>();
quartz.AddJob<ResumeWorkflowJob>();
return quartz;
}
@ -61,12 +62,11 @@ public static class ServiceCollectionExtensions
}
private static IServiceCollectionQuartzConfigurator AddJob<TJob>(this IServiceCollectionQuartzConfigurator quartz) where TJob : IElsaJob =>
quartz.AddJob<QuartzJob<RunWorkflowJob>>(job => job.StoreDurably().WithIdentity(nameof(RunWorkflowJob)));
quartz.AddJob<QuartzJob<TJob>>(job => job.StoreDurably().WithIdentity(typeof(TJob).Name));
private static void ConfigureQuartz(IServiceCollectionQuartzConfigurator quartz, Action<IServiceCollectionQuartzConfigurator>? configureQuartz)
{
quartz.UseMicrosoftDependencyInjectionJobFactory();
quartz.AddJob<QuartzJob<RunWorkflowJob>>(job => job.StoreDurably().WithIdentity(nameof(RunWorkflowJob)));
quartz.UseSimpleTypeLoader();
quartz.UseInMemoryStore();
configureQuartz?.Invoke(quartz);

View file

@ -1,5 +1,4 @@
using Elsa.Modules.Quartz.Contracts;
using Elsa.Modules.Quartz.Jobs;
using Elsa.Scheduling.Contracts;
using Elsa.Scheduling.Schedules;
using Microsoft.Extensions.Logging;

View file

@ -1,2 +1 @@
using System;
[assembly: CLSCompliant(true)]

View file

@ -1,3 +1,5 @@
using Elsa.Models;
namespace Elsa.Persistence.Entities;
public class WorkflowBookmark : Entity
@ -11,4 +13,6 @@ public class WorkflowBookmark : Entity
public string ActivityId { get; init; } = default!;
public string ActivityInstanceId { get; init; } = default!;
public string? CallbackMethodName { get; set; }
public Bookmark ToBookmark() => new(Id, Name, Hash, Payload, ActivityId, ActivityInstanceId, CallbackMethodName);
}

View file

@ -65,7 +65,7 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware
// Get a copy of current bookmarks.
var existingBookmarks = await _requestSender.RequestAsync(new FindWorkflowBookmarks(workflowInstance.Id), cancellationToken);
var bookmarksSnapshot = existingBookmarks.Select(x => new Bookmark(x.Id, x.Name, x.Hash, x.Payload, x.ActivityId, x.ActivityInstanceId, x.CallbackMethodName)).ToList();
var bookmarksSnapshot = existingBookmarks.Select(x => x.ToBookmark()).ToList();
// Exclude the current bookmark that initiated the creation of the workflow context, if any.
var invokedBookmark = context.Bookmark;

View file

@ -1,3 +1,5 @@
using Elsa.Contracts;
using Elsa.Helpers;
using Elsa.Mediator.Contracts;
using Elsa.Persistence.Entities;
@ -19,4 +21,6 @@ public record FindWorkflowBookmarks : IRequest<IEnumerable<WorkflowBookmark>>
public string? WorkflowInstanceId { get; }
public string? Name { get; }
public string? Hash { get; }
public static FindWorkflowBookmarks ForActivity<T>(string? hash = default) where T : IActivity => new(TypeNameHelper.GenerateTypeName<T>(), hash);
}

View file

@ -19,5 +19,17 @@ public class FindWorkflowBookmarksHandler : IRequestHandler<FindWorkflowBookmark
: Enumerable.Empty<WorkflowBookmark>();
private async Task<IEnumerable<WorkflowBookmark>> FindByWorkflowInstanceIdAsync(string instanceId, CancellationToken cancellationToken) => await _store.FindManyAsync(x => x.WorkflowInstanceId == instanceId, cancellationToken);
private async Task<IEnumerable<WorkflowBookmark>> FindByNameAsync(string name, string? hash, CancellationToken cancellationToken) => await _store.FindManyAsync(x => x.Name == name && x.Hash == hash, cancellationToken);
private async Task<IEnumerable<WorkflowBookmark>> FindByNameAsync(string name, string? hash, CancellationToken cancellationToken)
{
return await _store.QueryAsync(query =>
{
query = query.Where(x => x.Name == name);
if (!string.IsNullOrWhiteSpace(hash))
query = query.Where(x => x.Hash == hash);
return query;
}, cancellationToken);
}
}

View file

@ -24,11 +24,24 @@ public class FindWorkflowBookmarksHandler : IRequestHandler<FindWorkflowBookmark
return Task.FromResult(bookmarks);
}
private IEnumerable<WorkflowBookmark> Find(FindWorkflowBookmarks request) => request.WorkflowInstanceId != null
? FindByWorkflowInstanceId(request.WorkflowInstanceId)
: request.Name != null ? FindByName(request.Name, request.Hash)
private IEnumerable<WorkflowBookmark> Find(FindWorkflowBookmarks request) => request.WorkflowInstanceId != null
? FindByWorkflowInstanceId(request.WorkflowInstanceId)
: request.Name != null
? FindByName(request.Name, request.Hash)
: Enumerable.Empty<WorkflowBookmark>();
private IEnumerable<WorkflowBookmark> FindByWorkflowInstanceId(string instanceId) => _store.FindMany(x => x.WorkflowInstanceId == instanceId).ToList();
private IEnumerable<WorkflowBookmark> FindByName(string name, string? hash) => _store.FindMany(x => x.Name == name && x.Hash == hash).ToList();
private IEnumerable<WorkflowBookmark> FindByName(string name, string? hash)
{
return _store.Query(query =>
{
query = query.Where(x => x.Name == name);
if (!string.IsNullOrWhiteSpace(hash))
query = query.Where(x => x.Hash == hash);
return query;
});
}
}

View file

@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading;

View file

@ -6,43 +6,43 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Google.Protobuf" Version="3.19.1" />
<PackageReference Include="Google.Protobuf" Version="3.19.1"/>
<PackageReference Include="Grpc.Tools" Version="2.41.1">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="6.0.0" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="6.0.0"/>
</ItemGroup>
<ItemGroup Label="ProtoActor">
<PackageReference Include="Proto.Actor" Version="0.26.1-alpha.0.19" />
<PackageReference Include="Proto.Cluster" Version="0.26.1-alpha.0.19" />
<PackageReference Include="Proto.Cluster.Consul" Version="0.26.1-alpha.0.19" />
<PackageReference Include="Proto.Cluster.Identity.Redis" Version="0.26.1-alpha.0.19" />
<PackageReference Include="Proto.Cluster.TestProvider" Version="0.26.1-alpha.0.19" />
<PackageReference Include="Proto.Remote" Version="0.26.1-alpha.0.19" />
<PackageReference Include="Proto.Remote.GrpcCore" Version="0.26.1-alpha.0.19" />
</ItemGroup>
<!-- <ItemGroup Label="ProtoActor">-->
<!-- <ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Actor\Proto.Actor.csproj"/>-->
<!-- <ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Cluster.TestProvider\Proto.Cluster.TestProvider.csproj"/>-->
<!-- <ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Cluster\Proto.Cluster.csproj"/>-->
<!-- <ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Remote.GrpcCore\Proto.Remote.GrpcCore.csproj"/>-->
<!-- <ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Remote\Proto.Remote.csproj"/>-->
<!-- <PackageReference Include="Proto.Actor" Version="0.26.1-alpha.0.19" />-->
<!-- <PackageReference Include="Proto.Cluster" Version="0.26.1-alpha.0.19" />-->
<!-- <PackageReference Include="Proto.Cluster.Consul" Version="0.26.1-alpha.0.19" />-->
<!-- <PackageReference Include="Proto.Cluster.Identity.Redis" Version="0.26.1-alpha.0.19" />-->
<!-- <PackageReference Include="Proto.Cluster.TestProvider" Version="0.26.1-alpha.0.19" />-->
<!-- <PackageReference Include="Proto.Remote" Version="0.26.1-alpha.0.19" />-->
<!-- <PackageReference Include="Proto.Remote.GrpcCore" Version="0.26.1-alpha.0.19" />-->
<!-- </ItemGroup>-->
<ItemGroup>
<ProjectReference Include="..\Elsa.Runtime\Elsa.Runtime.csproj" />
<ItemGroup Label="ProtoActor">
<ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Actor\Proto.Actor.csproj"/>
<ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Cluster.TestProvider\Proto.Cluster.TestProvider.csproj"/>
<ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Cluster\Proto.Cluster.csproj"/>
<ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Remote.GrpcCore\Proto.Remote.GrpcCore.csproj"/>
<ProjectReference Include="..\..\..\..\..\ProtoActor\protoactor-dotnet\src\Proto.Remote\Proto.Remote.csproj"/>
</ItemGroup>
<ItemGroup>
<Protobuf Include="messages.proto" />
<ProjectReference Include="..\Elsa.Runtime\Elsa.Runtime.csproj"/>
</ItemGroup>
<ItemGroup>
<Folder Include="Contracts" />
<Protobuf Include="messages.proto"/>
</ItemGroup>
<ItemGroup>
<Folder Include="Contracts"/>
</ItemGroup>
</Project>

View file

@ -23,22 +23,6 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker
_cluster = cluster;
}
public async Task<DispatchWorkflowResult> DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default)
{
var (definitionId, version) = request;
var name = GetActorName(definitionId, version);
var message = new DispatchWorkflowDefinition
{
Id = definitionId,
Version = version
};
await _cluster.RequestAsync<Unit>(name, GrainKinds.WorkflowDefinition, message, cancellationToken);
return new DispatchWorkflowResult();
}
public async Task<ExecuteWorkflowResult> ExecuteAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default)
{
var (definitionId, version) = request;
@ -57,23 +41,6 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker
return new ExecuteWorkflowResult(workflowState, bookmarks);
}
public async Task<DispatchWorkflowResult> DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
{
var (instanceId, bookmark) = request;
var name = GetActorName(instanceId);
var bookmarkMessage = MapBookmark(bookmark);
var message = new DispatchWorkflowInstance
{
Id = instanceId,
Bookmark = bookmarkMessage
};
await _cluster.RequestAsync<Unit>(name, GrainKinds.WorkflowInstance, message, cancellationToken);
return new DispatchWorkflowResult();
}
public async Task<ExecuteWorkflowResult> ExecuteAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
{
var (instanceId, bookmark) = request;
@ -92,6 +59,39 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker
return new ExecuteWorkflowResult(workflowState, bookmarks);
}
public async Task<DispatchWorkflowResult> DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default)
{
var (definitionId, version) = request;
var name = GetActorName(definitionId, version);
var message = new DispatchWorkflowDefinition
{
Id = definitionId,
Version = version
};
await _cluster.RequestAsync<Unit>(name, GrainKinds.WorkflowDefinition, message, cancellationToken);
return new DispatchWorkflowResult();
}
public async Task<DispatchWorkflowResult> DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
{
var (instanceId, bookmark) = request;
var name = GetActorName(instanceId);
var bookmarkMessage = MapBookmark(bookmark);
var message = new DispatchWorkflowInstance
{
Id = instanceId,
Bookmark = bookmarkMessage
};
await _cluster.RequestAsync<Unit>(name, GrainKinds.WorkflowInstance, message, cancellationToken);
return new DispatchWorkflowResult();
}
private BookmarkMessage? MapBookmark(Bookmark? bookmark)
{

View file

@ -1,2 +1 @@
using System;
[assembly: CLSCompliant(true)]

View file

@ -9,17 +9,17 @@ namespace Elsa.Runtime.Middleware;
public static class PersistWorkflowExecutionLogMiddlewareExtensions
{
public static IWorkflowExecutionBuilder UseWorkflowExecutionEvents(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware<WorkflowExecutionEventsMiddleware>();
public static IWorkflowExecutionBuilder UseWorkflowExecutionEvents(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware<PublishWorkflowExecutionEventsMiddleware>();
}
/// <summary>
/// Processes collected bookmarks.
/// </summary>
public class WorkflowExecutionEventsMiddleware : WorkflowExecutionMiddleware
public class PublishWorkflowExecutionEventsMiddleware : WorkflowExecutionMiddleware
{
private readonly IEventPublisher _eventPublisher;
public WorkflowExecutionEventsMiddleware(WorkflowMiddlewareDelegate next, IEventPublisher eventPublisher) : base(next)
public PublishWorkflowExecutionEventsMiddleware(WorkflowMiddlewareDelegate next, IEventPublisher eventPublisher) : base(next)
{
_eventPublisher = eventPublisher;
}

View file

@ -10,7 +10,6 @@ using Elsa.Extensions;
using Elsa.Management.Contracts;
using Elsa.Management.Extensions;
using Elsa.Mediator.Extensions;
using Elsa.Modules.Quartz.Extensions;
using Elsa.Modules.Quartz.Services;
using Elsa.Persistence.EntityFrameworkCore.Extensions;
using Elsa.Persistence.EntityFrameworkCore.Sqlite;
@ -36,12 +35,13 @@ services
.AddElsa()
.AddMediator()
.AddEntityFrameworkCorePersistence((_, ef) => ef.UseSqlite())
.AddProtoActorWorkflowHost()
.IndexWorkflowTriggers()
.AddElsaManagement()
.AddScheduling(new QuartzSchedulingServiceProvider())
.AddHttpActivityServices()
.AddSchedulingActivities()
.AddProtoActorWorkflowHost()
.ConfigureWorkflowRuntime(options =>
{
options.Workflows.Add("HelloWorldWorkflow", new HelloWorldWorkflow());
@ -74,9 +74,6 @@ services
.AddJavaScriptExpressions()
.AddLiquidExpressions();
// Register modules.
services.AddQuartzModule(); // Provides a scheduler implementation for Timer activities.
// Configure middleware pipeline.
var app = builder.Build();
var serviceProvider = app.Services;

View file

@ -1,5 +1,4 @@
using Elsa.Models;
using Fluid;
namespace Elsa.Scripting.Liquid.Contracts
{

View file

@ -2,7 +2,6 @@ using Elsa.Contracts;
using Elsa.Helpers;
using Elsa.Models;
using Elsa.Scripting.Liquid.Contracts;
using Elsa.Scripting.Liquid.Services;
namespace Elsa.Scripting.Liquid.Expressions;