From 78a799edb97616f4d71f74dca94e008cc421c4dd Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Sun, 10 Apr 2022 14:35:53 +0200 Subject: [PATCH] Incremental work on proto actor and service bus --- Elsa.sln | 16 ++- .../ApiResults/DispatchWorkflowResult.cs | 4 +- .../ApiResults/ExecuteWorkflowResult.cs | 4 +- .../Elsa.Core/Contracts/IWorkflowRunner.cs | 8 +- .../Models/DispatchWorkflowResult.cs | 3 - .../Elsa.Core/Models/ExecuteWorkflowResult.cs | 5 - .../Elsa.Core/Models/InvokeWorkflowResult.cs | 5 + src/core/Elsa.Core/Services/WorkflowRunner.cs | 10 +- .../Abstractions/Job.cs | 0 .../AssemblyInfo.cs | 0 .../Contracts/IJob.cs | 0 .../Contracts/IJobQueue.cs | 0 .../Contracts/IJobQueueProvider.cs | 0 .../Contracts/IJobRunner.cs | 0 .../Contracts/IJobScheduler.cs | 0 .../Contracts/IJobSchedulerProvider.cs | 0 .../Contracts/IJobSerializer.cs | 0 .../Contracts/ISchedule.cs | 0 .../Elsa.Jobs.Abstractions.csproj} | 1 + .../Extensions/ServiceCollectionExtensions.cs | 0 .../Models/JobExecutionContext.cs | 0 .../Schedules/CronSchedule.cs | 0 .../Schedules/RecurringSchedule.cs | 0 .../Schedules/SpecificInstantSchedule.cs | 0 .../Services/JobRunner.cs | 0 .../Services/JobSerializer.cs | 0 .../Contracts/IServiceBus.cs | 7 ++ .../Contracts/IServiceBusProvider.cs | 8 ++ .../Elsa.ServiceBus.Abstractions.csproj | 13 +++ .../Elsa.Modules.Activities.csproj | 2 +- .../Elsa.Modules.Hangfire.csproj | 2 +- .../Middleware/HttpTriggerMiddleware.cs | 2 +- .../Elsa.Modules.MassTransit.csproj | 19 +++ .../Extensions/ServiceCollectionExtensions.cs | 13 +++ .../Services/MassTransitServiceBus.cs | 12 ++ .../Services/MassTransitServiceBusProvider.cs | 16 +++ .../Services/MassTransitWorkflowDispatcher.cs | 27 +++++ .../Elsa.Modules.Scheduling.csproj | 2 +- .../Jobs/ResumeWorkflowJob.cs | 4 +- .../Jobs/RunWorkflowJob.cs | 4 +- .../Actors/WorkflowInstanceActor.cs | 60 ---------- .../Elsa.Runtime.ProtoActor.csproj | 17 ++- .../Extensions/ProtoInputExtensions.cs | 2 +- .../Extensions/ServiceCollectionExtensions.cs | 53 +++++---- .../Elsa.Runtime.ProtoActor/GrainKinds.cs | 7 -- .../WorkflowDefinitionGrain.cs} | 64 +++++------ .../WorkflowInstanceGrain.cs} | 62 ++++++---- .../HostedServices/WorkflowServerHost.cs | 18 ++- .../Protos/Grains.proto | 14 +++ .../{messages.proto => Protos/Messages.proto} | 34 ++---- .../Services/GrainClientFactory.cs | 18 +++ .../Services/ProtoActorWorkflowInvoker.cs | 108 +++++++----------- .../ExecuteWorkflowInstructionResult.cs | 2 +- .../Contracts/IWorkflowDispatcher.cs | 13 +++ .../Contracts/IWorkflowInvoker.cs | 16 ++- .../Contracts/IWorkflowService.cs | 9 +- .../Extensions/ServiceCollectionExtensions.cs | 25 +++- .../DispatchedWorkflowDefinitionWorker.cs | 42 +++++++ .../DispatchedWorkflowInstanceWorker.cs | 41 +++++++ .../TriggerWorkflowInstructionInterpreter.cs | 14 ++- .../DispatchWorkflowDefinitionResponse.cs | 3 + .../DispatchWorkflowInstanceResponse.cs | 3 + .../ExecuteWorkflowDefinitionRequest.cs | 5 - .../Models/ExecuteWorkflowInstanceRequest.cs | 5 - .../Models/InvokeWorkflowDefinitionRequest.cs | 5 + .../Models/InvokeWorkflowInstanceRequest.cs | 5 + .../Options/WorkflowRuntimeOptions.cs | 14 ++- .../Services/DefaultWorkflowInvoker.cs | 21 ++++ .../Services/TaskBasedWorkflowDispatcher.cs | 34 ++++++ .../Elsa.Runtime/Services/WorkflowService.cs | 28 +++-- .../Elsa.Samples.Web1.csproj | 2 + 71 files changed, 612 insertions(+), 319 deletions(-) delete mode 100644 src/core/Elsa.Core/Models/DispatchWorkflowResult.cs delete mode 100644 src/core/Elsa.Core/Models/ExecuteWorkflowResult.cs create mode 100644 src/core/Elsa.Core/Models/InvokeWorkflowResult.cs rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Abstractions/Job.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/AssemblyInfo.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Contracts/IJob.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Contracts/IJobQueue.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Contracts/IJobQueueProvider.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Contracts/IJobRunner.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Contracts/IJobScheduler.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Contracts/IJobSchedulerProvider.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Contracts/IJobSerializer.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Contracts/ISchedule.cs (100%) rename src/core/{Elsa.Jobs/Elsa.Jobs.csproj => Elsa.Jobs.Abstractions/Elsa.Jobs.Abstractions.csproj} (91%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Extensions/ServiceCollectionExtensions.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Models/JobExecutionContext.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Schedules/CronSchedule.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Schedules/RecurringSchedule.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Schedules/SpecificInstantSchedule.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Services/JobRunner.cs (100%) rename src/core/{Elsa.Jobs => Elsa.Jobs.Abstractions}/Services/JobSerializer.cs (100%) create mode 100644 src/core/Elsa.ServiceBus.Abstractions/Contracts/IServiceBus.cs create mode 100644 src/core/Elsa.ServiceBus.Abstractions/Contracts/IServiceBusProvider.cs create mode 100644 src/core/Elsa.ServiceBus.Abstractions/Elsa.ServiceBus.Abstractions.csproj create mode 100644 src/modules/Elsa.Modules.MassTransit/Elsa.Modules.MassTransit.csproj create mode 100644 src/modules/Elsa.Modules.MassTransit/Extensions/ServiceCollectionExtensions.cs create mode 100644 src/modules/Elsa.Modules.MassTransit/Services/MassTransitServiceBus.cs create mode 100644 src/modules/Elsa.Modules.MassTransit/Services/MassTransitServiceBusProvider.cs create mode 100644 src/modules/Elsa.Modules.MassTransit/Services/MassTransitWorkflowDispatcher.cs delete mode 100644 src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowInstanceActor.cs delete mode 100644 src/runtime/Elsa.Runtime.ProtoActor/GrainKinds.cs rename src/runtime/Elsa.Runtime.ProtoActor/{Actors/WorkflowDefinitionActor.cs => Grains/WorkflowDefinitionGrain.cs} (51%) rename src/runtime/Elsa.Runtime.ProtoActor/{Actors/WorkflowOperatorActor.cs => Grains/WorkflowInstanceGrain.cs} (61%) create mode 100644 src/runtime/Elsa.Runtime.ProtoActor/Protos/Grains.proto rename src/runtime/Elsa.Runtime.ProtoActor/{messages.proto => Protos/Messages.proto} (65%) create mode 100644 src/runtime/Elsa.Runtime.ProtoActor/Services/GrainClientFactory.cs create mode 100644 src/runtime/Elsa.Runtime/Contracts/IWorkflowDispatcher.cs create mode 100644 src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowDefinitionWorker.cs create mode 100644 src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowInstanceWorker.cs create mode 100644 src/runtime/Elsa.Runtime/Models/DispatchWorkflowDefinitionResponse.cs create mode 100644 src/runtime/Elsa.Runtime/Models/DispatchWorkflowInstanceResponse.cs delete mode 100644 src/runtime/Elsa.Runtime/Models/ExecuteWorkflowDefinitionRequest.cs delete mode 100644 src/runtime/Elsa.Runtime/Models/ExecuteWorkflowInstanceRequest.cs create mode 100644 src/runtime/Elsa.Runtime/Models/InvokeWorkflowDefinitionRequest.cs create mode 100644 src/runtime/Elsa.Runtime/Models/InvokeWorkflowInstanceRequest.cs create mode 100644 src/runtime/Elsa.Runtime/Services/DefaultWorkflowInvoker.cs create mode 100644 src/runtime/Elsa.Runtime/Services/TaskBasedWorkflowDispatcher.cs diff --git a/Elsa.sln b/Elsa.sln index 7b9706cba..ae79c21f2 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -70,7 +70,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "modules", "modules", "{5BA4 EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Modules.Quartz", "src\modules\Elsa.Modules.Quartz\Elsa.Modules.Quartz.csproj", "{7D5A49B4-9A9B-496E-803B-DEB85B2C3132}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Jobs", "src\core\Elsa.Jobs\Elsa.Jobs.csproj", "{50697882-E38B-4CE7-B041-8009C18490E0}" +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Jobs.Abstractions", "src\core\Elsa.Jobs.Abstractions\Elsa.Jobs.Abstractions.csproj", "{50697882-E38B-4CE7-B041-8009C18490E0}" EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Modules.Http", "src\modules\Elsa.Modules.Http\Elsa.Modules.Http.csproj", "{82E26BF9-5F3A-4365-899A-AB1FFD54AA45}" EndProject @@ -90,6 +90,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Modules.SetName", "src EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Modules.Activities", "src\modules\Elsa.Modules.Activities\Elsa.Modules.Activities.csproj", "{05249066-9125-47D8-847B-A86887B1D269}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.ServiceBus.Abstractions", "src\core\Elsa.ServiceBus.Abstractions\Elsa.ServiceBus.Abstractions.csproj", "{61FAFEED-A477-4D8F-A293-C0196F11EA0F}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Modules.MassTransit", "src\modules\Elsa.Modules.MassTransit\Elsa.Modules.MassTransit.csproj", "{BB983D0B-A939-4008-B9E3-C7C172BCF83A}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -212,6 +216,14 @@ Global {05249066-9125-47D8-847B-A86887B1D269}.Debug|Any CPU.Build.0 = Debug|Any CPU {05249066-9125-47D8-847B-A86887B1D269}.Release|Any CPU.ActiveCfg = Release|Any CPU {05249066-9125-47D8-847B-A86887B1D269}.Release|Any CPU.Build.0 = Release|Any CPU + {61FAFEED-A477-4D8F-A293-C0196F11EA0F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {61FAFEED-A477-4D8F-A293-C0196F11EA0F}.Debug|Any CPU.Build.0 = Debug|Any CPU + {61FAFEED-A477-4D8F-A293-C0196F11EA0F}.Release|Any CPU.ActiveCfg = Release|Any CPU + {61FAFEED-A477-4D8F-A293-C0196F11EA0F}.Release|Any CPU.Build.0 = Release|Any CPU + {BB983D0B-A939-4008-B9E3-C7C172BCF83A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {BB983D0B-A939-4008-B9E3-C7C172BCF83A}.Debug|Any CPU.Build.0 = Debug|Any CPU + {BB983D0B-A939-4008-B9E3-C7C172BCF83A}.Release|Any CPU.ActiveCfg = Release|Any CPU + {BB983D0B-A939-4008-B9E3-C7C172BCF83A}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(NestedProjects) = preSolution {155227F0-A33B-40AA-A4B4-06F813EB921B} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F} @@ -254,5 +266,7 @@ Global {302BFC43-ED2F-43AE-8AD4-FCD481B0AC67} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79} {E4EB94A0-E6F6-4EFA-BE55-7CEED1D8431A} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79} {05249066-9125-47D8-847B-A86887B1D269} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79} + {61FAFEED-A477-4D8F-A293-C0196F11EA0F} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09} + {BB983D0B-A939-4008-B9E3-C7C172BCF83A} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79} EndGlobalSection EndGlobal diff --git a/src/api/Elsa.Api/ApiResults/DispatchWorkflowResult.cs b/src/api/Elsa.Api/ApiResults/DispatchWorkflowResult.cs index cf66d4400..931362e63 100644 --- a/src/api/Elsa.Api/ApiResults/DispatchWorkflowResult.cs +++ b/src/api/Elsa.Api/ApiResults/DispatchWorkflowResult.cs @@ -17,9 +17,9 @@ public class DispatchWorkflowResult : IResult public async Task ExecuteAsync(HttpContext httpContext) { var response = httpContext.Response; - var workflowInvoker = httpContext.RequestServices.GetRequiredService(); + var workflowDispatcher = httpContext.RequestServices.GetRequiredService(); var definitionId = Workflow.Identity.DefinitionId; - var result = await workflowInvoker.DispatchAsync(new DispatchWorkflowDefinitionRequest(definitionId, VersionOptions.Published)); + var result = await workflowDispatcher.DispatchAsync(new DispatchWorkflowDefinitionRequest(definitionId, VersionOptions.Published)); response.StatusCode = (int)HttpStatusCode.OK; await response.WriteAsJsonAsync(result, httpContext.RequestAborted); diff --git a/src/api/Elsa.Api/ApiResults/ExecuteWorkflowResult.cs b/src/api/Elsa.Api/ApiResults/ExecuteWorkflowResult.cs index 32f0c3e42..73574a767 100644 --- a/src/api/Elsa.Api/ApiResults/ExecuteWorkflowResult.cs +++ b/src/api/Elsa.Api/ApiResults/ExecuteWorkflowResult.cs @@ -19,8 +19,8 @@ public class ExecuteWorkflowResult : IResult var response = httpContext.Response; var workflowInvoker = httpContext.RequestServices.GetRequiredService(); var definitionId = Workflow.Identity.DefinitionId; - var executeRequest = new ExecuteWorkflowDefinitionRequest(definitionId, VersionOptions.Published); - var result = await workflowInvoker.ExecuteAsync(executeRequest, CancellationToken.None); + var executeRequest = new InvokeWorkflowDefinitionRequest(definitionId, VersionOptions.Published); + var result = await workflowInvoker.InvokeAsync(executeRequest, CancellationToken.None); if (!response.HasStarted) await response.WriteAsJsonAsync(result, httpContext.RequestAborted); diff --git a/src/core/Elsa.Core/Contracts/IWorkflowRunner.cs b/src/core/Elsa.Core/Contracts/IWorkflowRunner.cs index 7b9b2f4cf..cd82e9aed 100644 --- a/src/core/Elsa.Core/Contracts/IWorkflowRunner.cs +++ b/src/core/Elsa.Core/Contracts/IWorkflowRunner.cs @@ -5,8 +5,8 @@ namespace Elsa.Contracts; public interface IWorkflowRunner { - Task RunAsync(Workflow workflow, IDictionary? input = default, CancellationToken cancellationToken = default); - Task RunAsync(Workflow workflow, WorkflowState workflowState, IDictionary? input = default, CancellationToken cancellationToken = default); - Task RunAsync(Workflow workflow, WorkflowState workflowState, Bookmark? bookmark, IDictionary? input = default, CancellationToken cancellationToken = default); - Task RunAsync(WorkflowExecutionContext workflowExecutionContext); + Task RunAsync(Workflow workflow, IDictionary? input = default, CancellationToken cancellationToken = default); + Task RunAsync(Workflow workflow, WorkflowState workflowState, IDictionary? input = default, CancellationToken cancellationToken = default); + Task RunAsync(Workflow workflow, WorkflowState workflowState, Bookmark? bookmark, IDictionary? input = default, CancellationToken cancellationToken = default); + Task RunAsync(WorkflowExecutionContext workflowExecutionContext); } \ No newline at end of file diff --git a/src/core/Elsa.Core/Models/DispatchWorkflowResult.cs b/src/core/Elsa.Core/Models/DispatchWorkflowResult.cs deleted file mode 100644 index 050a1e470..000000000 --- a/src/core/Elsa.Core/Models/DispatchWorkflowResult.cs +++ /dev/null @@ -1,3 +0,0 @@ -namespace Elsa.Models; - -public record DispatchWorkflowResult(); \ No newline at end of file diff --git a/src/core/Elsa.Core/Models/ExecuteWorkflowResult.cs b/src/core/Elsa.Core/Models/ExecuteWorkflowResult.cs deleted file mode 100644 index 08352a0cb..000000000 --- a/src/core/Elsa.Core/Models/ExecuteWorkflowResult.cs +++ /dev/null @@ -1,5 +0,0 @@ -using Elsa.State; - -namespace Elsa.Models; - -public record ExecuteWorkflowResult(WorkflowState WorkflowState, IReadOnlyCollection Bookmarks); \ No newline at end of file diff --git a/src/core/Elsa.Core/Models/InvokeWorkflowResult.cs b/src/core/Elsa.Core/Models/InvokeWorkflowResult.cs new file mode 100644 index 000000000..c8fe7d3be --- /dev/null +++ b/src/core/Elsa.Core/Models/InvokeWorkflowResult.cs @@ -0,0 +1,5 @@ +using Elsa.State; + +namespace Elsa.Models; + +public record InvokeWorkflowResult(WorkflowState WorkflowState, IReadOnlyCollection Bookmarks); \ No newline at end of file diff --git a/src/core/Elsa.Core/Services/WorkflowRunner.cs b/src/core/Elsa.Core/Services/WorkflowRunner.cs index 05fc485c6..829ba16d4 100644 --- a/src/core/Elsa.Core/Services/WorkflowRunner.cs +++ b/src/core/Elsa.Core/Services/WorkflowRunner.cs @@ -30,7 +30,7 @@ public class WorkflowRunner : IWorkflowRunner _schedulerFactory = schedulerFactory; } - public async Task RunAsync(Workflow workflow, IDictionary? input, CancellationToken cancellationToken = default) + public async Task RunAsync(Workflow workflow, IDictionary? input, CancellationToken cancellationToken = default) { // Create a child scope. using var scope = _serviceScopeFactory.CreateScope(); @@ -44,7 +44,7 @@ public class WorkflowRunner : IWorkflowRunner return await RunAsync(workflowExecutionContext); } - public async Task RunAsync(Workflow workflow, WorkflowState workflowState, IDictionary? input, CancellationToken cancellationToken = default) + public async Task RunAsync(Workflow workflow, WorkflowState workflowState, IDictionary? input, CancellationToken cancellationToken = default) { // Create a child scope. using var scope = _serviceScopeFactory.CreateScope(); @@ -58,7 +58,7 @@ public class WorkflowRunner : IWorkflowRunner return await RunAsync(workflowExecutionContext); } - public async Task RunAsync(Workflow workflow, WorkflowState workflowState, Bookmark? bookmark, IDictionary? input, CancellationToken cancellationToken = default) + public async Task RunAsync(Workflow workflow, WorkflowState workflowState, Bookmark? bookmark, IDictionary? input, CancellationToken cancellationToken = default) { // Create a child scope. using var scope = _serviceScopeFactory.CreateScope(); @@ -72,7 +72,7 @@ public class WorkflowRunner : IWorkflowRunner return await RunAsync(workflowExecutionContext); } - public async Task RunAsync(WorkflowExecutionContext workflowExecutionContext) + public async Task RunAsync(WorkflowExecutionContext workflowExecutionContext) { // Execute the activity execution pipeline. await _pipeline.ExecuteAsync(workflowExecutionContext); @@ -81,7 +81,7 @@ public class WorkflowRunner : IWorkflowRunner var workflowState = _workflowStateSerializer.ReadState(workflowExecutionContext); // Return workflow execution result containing state + bookmarks. - return new ExecuteWorkflowResult(workflowState, workflowExecutionContext.Bookmarks); + return new InvokeWorkflowResult(workflowState, workflowExecutionContext.Bookmarks); } private WorkflowExecutionContext CreateWorkflowExecutionContext( diff --git a/src/core/Elsa.Jobs/Abstractions/Job.cs b/src/core/Elsa.Jobs.Abstractions/Abstractions/Job.cs similarity index 100% rename from src/core/Elsa.Jobs/Abstractions/Job.cs rename to src/core/Elsa.Jobs.Abstractions/Abstractions/Job.cs diff --git a/src/core/Elsa.Jobs/AssemblyInfo.cs b/src/core/Elsa.Jobs.Abstractions/AssemblyInfo.cs similarity index 100% rename from src/core/Elsa.Jobs/AssemblyInfo.cs rename to src/core/Elsa.Jobs.Abstractions/AssemblyInfo.cs diff --git a/src/core/Elsa.Jobs/Contracts/IJob.cs b/src/core/Elsa.Jobs.Abstractions/Contracts/IJob.cs similarity index 100% rename from src/core/Elsa.Jobs/Contracts/IJob.cs rename to src/core/Elsa.Jobs.Abstractions/Contracts/IJob.cs diff --git a/src/core/Elsa.Jobs/Contracts/IJobQueue.cs b/src/core/Elsa.Jobs.Abstractions/Contracts/IJobQueue.cs similarity index 100% rename from src/core/Elsa.Jobs/Contracts/IJobQueue.cs rename to src/core/Elsa.Jobs.Abstractions/Contracts/IJobQueue.cs diff --git a/src/core/Elsa.Jobs/Contracts/IJobQueueProvider.cs b/src/core/Elsa.Jobs.Abstractions/Contracts/IJobQueueProvider.cs similarity index 100% rename from src/core/Elsa.Jobs/Contracts/IJobQueueProvider.cs rename to src/core/Elsa.Jobs.Abstractions/Contracts/IJobQueueProvider.cs diff --git a/src/core/Elsa.Jobs/Contracts/IJobRunner.cs b/src/core/Elsa.Jobs.Abstractions/Contracts/IJobRunner.cs similarity index 100% rename from src/core/Elsa.Jobs/Contracts/IJobRunner.cs rename to src/core/Elsa.Jobs.Abstractions/Contracts/IJobRunner.cs diff --git a/src/core/Elsa.Jobs/Contracts/IJobScheduler.cs b/src/core/Elsa.Jobs.Abstractions/Contracts/IJobScheduler.cs similarity index 100% rename from src/core/Elsa.Jobs/Contracts/IJobScheduler.cs rename to src/core/Elsa.Jobs.Abstractions/Contracts/IJobScheduler.cs diff --git a/src/core/Elsa.Jobs/Contracts/IJobSchedulerProvider.cs b/src/core/Elsa.Jobs.Abstractions/Contracts/IJobSchedulerProvider.cs similarity index 100% rename from src/core/Elsa.Jobs/Contracts/IJobSchedulerProvider.cs rename to src/core/Elsa.Jobs.Abstractions/Contracts/IJobSchedulerProvider.cs diff --git a/src/core/Elsa.Jobs/Contracts/IJobSerializer.cs b/src/core/Elsa.Jobs.Abstractions/Contracts/IJobSerializer.cs similarity index 100% rename from src/core/Elsa.Jobs/Contracts/IJobSerializer.cs rename to src/core/Elsa.Jobs.Abstractions/Contracts/IJobSerializer.cs diff --git a/src/core/Elsa.Jobs/Contracts/ISchedule.cs b/src/core/Elsa.Jobs.Abstractions/Contracts/ISchedule.cs similarity index 100% rename from src/core/Elsa.Jobs/Contracts/ISchedule.cs rename to src/core/Elsa.Jobs.Abstractions/Contracts/ISchedule.cs diff --git a/src/core/Elsa.Jobs/Elsa.Jobs.csproj b/src/core/Elsa.Jobs.Abstractions/Elsa.Jobs.Abstractions.csproj similarity index 91% rename from src/core/Elsa.Jobs/Elsa.Jobs.csproj rename to src/core/Elsa.Jobs.Abstractions/Elsa.Jobs.Abstractions.csproj index 801c56a86..be8687453 100644 --- a/src/core/Elsa.Jobs/Elsa.Jobs.csproj +++ b/src/core/Elsa.Jobs.Abstractions/Elsa.Jobs.Abstractions.csproj @@ -4,6 +4,7 @@ net6.0 enable Latest + Elsa.Jobs diff --git a/src/core/Elsa.Jobs/Extensions/ServiceCollectionExtensions.cs b/src/core/Elsa.Jobs.Abstractions/Extensions/ServiceCollectionExtensions.cs similarity index 100% rename from src/core/Elsa.Jobs/Extensions/ServiceCollectionExtensions.cs rename to src/core/Elsa.Jobs.Abstractions/Extensions/ServiceCollectionExtensions.cs diff --git a/src/core/Elsa.Jobs/Models/JobExecutionContext.cs b/src/core/Elsa.Jobs.Abstractions/Models/JobExecutionContext.cs similarity index 100% rename from src/core/Elsa.Jobs/Models/JobExecutionContext.cs rename to src/core/Elsa.Jobs.Abstractions/Models/JobExecutionContext.cs diff --git a/src/core/Elsa.Jobs/Schedules/CronSchedule.cs b/src/core/Elsa.Jobs.Abstractions/Schedules/CronSchedule.cs similarity index 100% rename from src/core/Elsa.Jobs/Schedules/CronSchedule.cs rename to src/core/Elsa.Jobs.Abstractions/Schedules/CronSchedule.cs diff --git a/src/core/Elsa.Jobs/Schedules/RecurringSchedule.cs b/src/core/Elsa.Jobs.Abstractions/Schedules/RecurringSchedule.cs similarity index 100% rename from src/core/Elsa.Jobs/Schedules/RecurringSchedule.cs rename to src/core/Elsa.Jobs.Abstractions/Schedules/RecurringSchedule.cs diff --git a/src/core/Elsa.Jobs/Schedules/SpecificInstantSchedule.cs b/src/core/Elsa.Jobs.Abstractions/Schedules/SpecificInstantSchedule.cs similarity index 100% rename from src/core/Elsa.Jobs/Schedules/SpecificInstantSchedule.cs rename to src/core/Elsa.Jobs.Abstractions/Schedules/SpecificInstantSchedule.cs diff --git a/src/core/Elsa.Jobs/Services/JobRunner.cs b/src/core/Elsa.Jobs.Abstractions/Services/JobRunner.cs similarity index 100% rename from src/core/Elsa.Jobs/Services/JobRunner.cs rename to src/core/Elsa.Jobs.Abstractions/Services/JobRunner.cs diff --git a/src/core/Elsa.Jobs/Services/JobSerializer.cs b/src/core/Elsa.Jobs.Abstractions/Services/JobSerializer.cs similarity index 100% rename from src/core/Elsa.Jobs/Services/JobSerializer.cs rename to src/core/Elsa.Jobs.Abstractions/Services/JobSerializer.cs diff --git a/src/core/Elsa.ServiceBus.Abstractions/Contracts/IServiceBus.cs b/src/core/Elsa.ServiceBus.Abstractions/Contracts/IServiceBus.cs new file mode 100644 index 000000000..d357c4519 --- /dev/null +++ b/src/core/Elsa.ServiceBus.Abstractions/Contracts/IServiceBus.cs @@ -0,0 +1,7 @@ +namespace Elsa.ServiceBus.Abstractions.Contracts; + +public interface IServiceBus +{ + Task SendAsync(object message, CancellationToken cancellationToken = default); + Task PublishAsync(object message, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/core/Elsa.ServiceBus.Abstractions/Contracts/IServiceBusProvider.cs b/src/core/Elsa.ServiceBus.Abstractions/Contracts/IServiceBusProvider.cs new file mode 100644 index 000000000..b4a5d104c --- /dev/null +++ b/src/core/Elsa.ServiceBus.Abstractions/Contracts/IServiceBusProvider.cs @@ -0,0 +1,8 @@ +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.ServiceBus.Abstractions.Contracts; + +public interface IServiceBusProvider +{ + void ConfigureServices(IServiceCollection services); +} \ No newline at end of file diff --git a/src/core/Elsa.ServiceBus.Abstractions/Elsa.ServiceBus.Abstractions.csproj b/src/core/Elsa.ServiceBus.Abstractions/Elsa.ServiceBus.Abstractions.csproj new file mode 100644 index 000000000..168c16fd2 --- /dev/null +++ b/src/core/Elsa.ServiceBus.Abstractions/Elsa.ServiceBus.Abstractions.csproj @@ -0,0 +1,13 @@ + + + + net6.0 + enable + enable + + + + + + + diff --git a/src/modules/Elsa.Modules.Activities/Elsa.Modules.Activities.csproj b/src/modules/Elsa.Modules.Activities/Elsa.Modules.Activities.csproj index 45d71909b..fd61fa3e2 100644 --- a/src/modules/Elsa.Modules.Activities/Elsa.Modules.Activities.csproj +++ b/src/modules/Elsa.Modules.Activities/Elsa.Modules.Activities.csproj @@ -7,7 +7,7 @@ - + diff --git a/src/modules/Elsa.Modules.Hangfire/Elsa.Modules.Hangfire.csproj b/src/modules/Elsa.Modules.Hangfire/Elsa.Modules.Hangfire.csproj index 42028df37..4109d664f 100644 --- a/src/modules/Elsa.Modules.Hangfire/Elsa.Modules.Hangfire.csproj +++ b/src/modules/Elsa.Modules.Hangfire/Elsa.Modules.Hangfire.csproj @@ -11,7 +11,7 @@ - + diff --git a/src/modules/Elsa.Modules.Http/Middleware/HttpTriggerMiddleware.cs b/src/modules/Elsa.Modules.Http/Middleware/HttpTriggerMiddleware.cs index e01db1908..48cf9ee13 100644 --- a/src/modules/Elsa.Modules.Http/Middleware/HttpTriggerMiddleware.cs +++ b/src/modules/Elsa.Modules.Http/Middleware/HttpTriggerMiddleware.cs @@ -53,7 +53,7 @@ public class HttpTriggerMiddleware var model = new { - workflowInstanceIds = executionResults.Select(x => x.ExecuteWorkflowResult.WorkflowState.Id).ToArray() + workflowInstanceIds = executionResults.Select(x => x.InvokeWorkflowResult.WorkflowState.Id).ToArray() }; var json = JsonSerializer.Serialize(model); diff --git a/src/modules/Elsa.Modules.MassTransit/Elsa.Modules.MassTransit.csproj b/src/modules/Elsa.Modules.MassTransit/Elsa.Modules.MassTransit.csproj new file mode 100644 index 000000000..a087d15a0 --- /dev/null +++ b/src/modules/Elsa.Modules.MassTransit/Elsa.Modules.MassTransit.csproj @@ -0,0 +1,19 @@ + + + + net6.0 + enable + enable + + + + + + + + + + + + + diff --git a/src/modules/Elsa.Modules.MassTransit/Extensions/ServiceCollectionExtensions.cs b/src/modules/Elsa.Modules.MassTransit/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 000000000..36b8a37e5 --- /dev/null +++ b/src/modules/Elsa.Modules.MassTransit/Extensions/ServiceCollectionExtensions.cs @@ -0,0 +1,13 @@ +using Elsa.Modules.MassTransit.Services; +using Elsa.Runtime.Contracts; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Modules.MassTransit.Extensions; + +public static class ServiceCollectionExtensions +{ + public static IServiceCollection AddMassTransitServices(this IServiceCollection services) + { + return services.AddSingleton(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.MassTransit/Services/MassTransitServiceBus.cs b/src/modules/Elsa.Modules.MassTransit/Services/MassTransitServiceBus.cs new file mode 100644 index 000000000..8d6f425b2 --- /dev/null +++ b/src/modules/Elsa.Modules.MassTransit/Services/MassTransitServiceBus.cs @@ -0,0 +1,12 @@ +using Elsa.ServiceBus.Abstractions.Contracts; +using MassTransit; + +namespace Elsa.Modules.MassTransit.Services; + +public class MassTransitServiceBus : IServiceBus +{ + private readonly IBus _bus; + public MassTransitServiceBus(IBus bus) => _bus = bus; + public async Task SendAsync(object message, CancellationToken cancellationToken = default) => await _bus.Send(message, cancellationToken); + public async Task PublishAsync(object message, CancellationToken cancellationToken = default) => await _bus.Publish(message, cancellationToken); +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.MassTransit/Services/MassTransitServiceBusProvider.cs b/src/modules/Elsa.Modules.MassTransit/Services/MassTransitServiceBusProvider.cs new file mode 100644 index 000000000..61b1ebf3f --- /dev/null +++ b/src/modules/Elsa.Modules.MassTransit/Services/MassTransitServiceBusProvider.cs @@ -0,0 +1,16 @@ +using Elsa.ServiceBus.Abstractions.Contracts; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Modules.MassTransit.Services; + +/// +/// A MassTransit implementation for . +/// +public class MassTransitServiceBusProvider : IServiceBusProvider +{ + public void ConfigureServices(IServiceCollection services) + { + services + .AddSingleton(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.MassTransit/Services/MassTransitWorkflowDispatcher.cs b/src/modules/Elsa.Modules.MassTransit/Services/MassTransitWorkflowDispatcher.cs new file mode 100644 index 000000000..0bb2af170 --- /dev/null +++ b/src/modules/Elsa.Modules.MassTransit/Services/MassTransitWorkflowDispatcher.cs @@ -0,0 +1,27 @@ +using Elsa.Runtime.Contracts; +using Elsa.Runtime.Models; +using MassTransit; + +namespace Elsa.Modules.MassTransit.Services; + +public class MassTransitWorkflowDispatcher : IWorkflowDispatcher +{ + private readonly IBus _bus; + + public MassTransitWorkflowDispatcher(IBus bus) + { + _bus = bus; + } + + public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) + { + await _bus.Send(request, cancellationToken); + return new(); + } + + public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default) + { + await _bus.Send(request, cancellationToken); + return new(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.Scheduling/Elsa.Modules.Scheduling.csproj b/src/modules/Elsa.Modules.Scheduling/Elsa.Modules.Scheduling.csproj index 18211728a..f87d1c304 100644 --- a/src/modules/Elsa.Modules.Scheduling/Elsa.Modules.Scheduling.csproj +++ b/src/modules/Elsa.Modules.Scheduling/Elsa.Modules.Scheduling.csproj @@ -7,7 +7,7 @@ - + diff --git a/src/modules/Elsa.Modules.Scheduling/Jobs/ResumeWorkflowJob.cs b/src/modules/Elsa.Modules.Scheduling/Jobs/ResumeWorkflowJob.cs index 97937e17a..4f3067d8d 100644 --- a/src/modules/Elsa.Modules.Scheduling/Jobs/ResumeWorkflowJob.cs +++ b/src/modules/Elsa.Modules.Scheduling/Jobs/ResumeWorkflowJob.cs @@ -27,7 +27,7 @@ public class ResumeWorkflowJob : Job protected override async ValueTask ExecuteAsync(JobExecutionContext context) { var request = new DispatchWorkflowInstanceRequest(WorkflowInstanceId, Bookmark); - var workflowInvoker = context.GetRequiredService(); - await workflowInvoker.DispatchAsync(request, context.CancellationToken); + var workflowDispatcher = context.GetRequiredService(); + await workflowDispatcher.DispatchAsync(request, context.CancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Modules.Scheduling/Jobs/RunWorkflowJob.cs b/src/modules/Elsa.Modules.Scheduling/Jobs/RunWorkflowJob.cs index 7c17bdf60..2b30ce513 100644 --- a/src/modules/Elsa.Modules.Scheduling/Jobs/RunWorkflowJob.cs +++ b/src/modules/Elsa.Modules.Scheduling/Jobs/RunWorkflowJob.cs @@ -25,7 +25,7 @@ public class RunWorkflowJob : Job protected override async ValueTask ExecuteAsync(JobExecutionContext context) { var request = new DispatchWorkflowDefinitionRequest(WorkflowId, VersionOptions.Published); - var workflowInvoker = context.GetRequiredService(); - await workflowInvoker.DispatchAsync(request, context.CancellationToken); + var workflowDispatcher = context.GetRequiredService(); + await workflowDispatcher.DispatchAsync(request, context.CancellationToken); } } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowInstanceActor.cs b/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowInstanceActor.cs deleted file mode 100644 index 86bb2b627..000000000 --- a/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowInstanceActor.cs +++ /dev/null @@ -1,60 +0,0 @@ -using System.Linq; -using System.Threading.Tasks; -using Elsa.Runtime.ProtoActor.Messages; -using Proto; -using Proto.DependencyInjection; - -namespace Elsa.Runtime.ProtoActor.Actors; - -/// -/// Dispatches a workflow instance for execution. The workflow instance will be executed by . -/// -public class WorkflowInstanceActor : IActor -{ - public Task ReceiveAsync(IContext context) => context.Message switch - { - ExecuteWorkflowInstance m => OnExecuteWorkflowInstanceAsync(context, m), - DispatchWorkflowInstance m => OnDispatchWorkflowInstanceAsync(context, m), - _ => Task.CompletedTask - }; - - private async Task OnExecuteWorkflowInstanceAsync(IContext context, ExecuteWorkflowInstance message) - { - var workflowInstanceId = message.Id; - var pid = GetWorkflowOperatorPid(context, workflowInstanceId); - var cancellationToken = context.CancellationToken; - var response = await context.RequestAsync(pid, message, cancellationToken); - - context.Respond(response); - } - - private Task OnDispatchWorkflowInstanceAsync(IContext context, DispatchWorkflowInstance message) - { - var workflowInstanceId = message.Id; - var pid = GetWorkflowOperatorPid(context, workflowInstanceId); - - var executeWorkflowInstanceMessage = new ExecuteWorkflowInstance - { - Id = workflowInstanceId, - Bookmark = message.Bookmark, - Input = message.Input - }; - - context.Send(pid, executeWorkflowInstanceMessage); - context.Respond(new Unit()); - return Task.CompletedTask; - } - - private PID GetWorkflowOperatorPid(IContext context, string workflowInstanceId) - { - var pid = context.System.ProcessRegistry.Find($"partition-activator/workflow-instance:{workflowInstanceId}").FirstOrDefault(); - - if (pid != null) - return pid; - - var props = context.System.DI().PropsFor(); - pid = context.SpawnNamed(props, workflowInstanceId); - - return pid; - } -} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Elsa.Runtime.ProtoActor.csproj b/src/runtime/Elsa.Runtime.ProtoActor/Elsa.Runtime.ProtoActor.csproj index 4c7581323..338c5e31d 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Elsa.Runtime.ProtoActor.csproj +++ b/src/runtime/Elsa.Runtime.ProtoActor/Elsa.Runtime.ProtoActor.csproj @@ -18,6 +18,7 @@ + @@ -37,12 +38,16 @@ - - - - - - + + + Both + Public + True + True + obj\Debug\net6.0\ + MSBuild:Compile + + diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ProtoInputExtensions.cs b/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ProtoInputExtensions.cs index 0ef641b98..c4bfdee96 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ProtoInputExtensions.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ProtoInputExtensions.cs @@ -1,7 +1,7 @@ using System.Collections.Generic; using System.Linq; using System.Text.Json; -using Elsa.Runtime.ProtoActor.Messages; +using Elsa.Runtime.Protos; namespace Elsa.Runtime.ProtoActor.Extensions; diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ServiceCollectionExtensions.cs b/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ServiceCollectionExtensions.cs index 29c3a1f39..a583718fc 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ServiceCollectionExtensions.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ServiceCollectionExtensions.cs @@ -1,8 +1,12 @@ using System; +using Elsa.Helpers; +using Elsa.Persistence.Models; using Elsa.Runtime.Contracts; -using Elsa.Runtime.ProtoActor.Actors; +using Elsa.Runtime.Extensions; +using Elsa.Runtime.ProtoActor.Grains; using Elsa.Runtime.ProtoActor.HostedServices; using Elsa.Runtime.ProtoActor.Services; +using Elsa.Runtime.Protos; using Microsoft.Extensions.DependencyInjection; using Proto; using Proto.Cluster; @@ -20,12 +24,13 @@ public static class ServiceCollectionExtensions { var systemConfig = GetSystemConfig(); + // Actor System. services.AddSingleton(sp => { var system = new ActorSystem(systemConfig).WithServiceProvider(sp); var remoteConfig = GetRemoteConfig(); var clusterConfig = GetClusterConfig(system, "my-cluster"); - + system .WithRemote(remoteConfig) .WithCluster(clusterConfig); @@ -33,37 +38,43 @@ public static class ServiceCollectionExtensions return system; }); + // Cluster. services.AddSingleton(sp => sp.GetRequiredService().Cluster()); - return services - .AddHostedService() - .AddSingleton() - .AddTransient() - .AddTransient() - .AddTransient(); - } - - private static ActorSystemConfig GetSystemConfig() => + // Actors. + services + .AddSingleton(sp => new WorkflowDefinitionGrainActor((context, _) => ActivatorUtilities.CreateInstance(sp, context))) + .AddSingleton(sp => new WorkflowInstanceGrainActor((context, _) => ActivatorUtilities.CreateInstance(sp, context))); + // Client factory. + services.AddSingleton(); + + // Configure runtime with ProtoActor workflow invoker. + services.ConfigureWorkflowRuntime(options => options.WorkflowInvokerFactory = sp => ActivatorUtilities.CreateInstance(sp)); + + return services + .AddHostedService(); + } + + private static ActorSystemConfig GetSystemConfig() => ActorSystemConfig .Setup() .WithDeadLetterThrottleCount(3) .WithDeadLetterThrottleInterval(TimeSpan.FromSeconds(10000)) .WithDeveloperSupervisionLogging(true) .WithDeadLetterRequestLogging(true); - + private static GrpcCoreRemoteConfig GetRemoteConfig() => GrpcCoreRemoteConfig .BindToLocalhost() - .WithProtoMessages(Messages.MessagesReflection.Descriptor) - ; - + .WithProtoMessages(Protos.MessagesReflection.Descriptor); + private static ClusterConfig GetClusterConfig(ActorSystem system, string clusterName) { //var clusterProvider = new ConsulProvider(new ConsulProviderConfig{}); var clusterProvider = new TestProvider(new TestProviderOptions(), new InMemAgent()); - - var workflowDefinitionProps = system.DI().PropsFor(); - var workflowInstanceProps = system.DI().PropsFor(); + + var workflowDefinitionProps = system.DI().PropsFor(); + var workflowInstanceProps = system.DI().PropsFor(); var clusterConfig = ClusterConfig @@ -73,12 +84,12 @@ public static class ServiceCollectionExtensions .WithActorRequestTimeout(TimeSpan.FromHours(1)) .WithActorActivationTimeout(TimeSpan.FromHours(1)) .WithActorSpawnTimeout(TimeSpan.FromHours(1)) - .WithClusterKind(GrainKinds.WorkflowDefinition, workflowDefinitionProps) - .WithClusterKind(GrainKinds.WorkflowInstance, workflowInstanceProps) + .WithClusterKind(WorkflowDefinitionGrainActor.Kind, workflowDefinitionProps) + .WithClusterKind(WorkflowInstanceGrainActor.Kind, workflowInstanceProps) ; return clusterConfig; } - + // private static IIdentityStorage GetIdentityLookup(string clusterName) => // new RedisIdentityStorage(clusterName, ConnectionMultiplexer // .Connect("localhost:6379" /* use proper config */) diff --git a/src/runtime/Elsa.Runtime.ProtoActor/GrainKinds.cs b/src/runtime/Elsa.Runtime.ProtoActor/GrainKinds.cs deleted file mode 100644 index c73ef3284..000000000 --- a/src/runtime/Elsa.Runtime.ProtoActor/GrainKinds.cs +++ /dev/null @@ -1,7 +0,0 @@ -namespace Elsa.Runtime.ProtoActor; - -public static class GrainKinds -{ - public const string WorkflowDefinition = "WorkflowDefinition"; - public const string WorkflowInstance = "WorkflowInstance"; -} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowDefinitionActor.cs b/src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowDefinitionGrain.cs similarity index 51% rename from src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowDefinitionActor.cs rename to src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowDefinitionGrain.cs index 399b39708..6f3b3717f 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowDefinitionActor.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowDefinitionGrain.cs @@ -1,3 +1,4 @@ +using System; using System.Threading; using System.Threading.Tasks; using Elsa.Contracts; @@ -6,69 +7,64 @@ using Elsa.Persistence.Commands; using Elsa.Persistence.Entities; using Elsa.Persistence.Models; using Elsa.Runtime.Contracts; -using Elsa.Runtime.ProtoActor.Messages; +using Elsa.Runtime.ProtoActor.Services; +using Elsa.Runtime.Protos; using Elsa.State; using Proto; -using Proto.Cluster; -using Unit = Elsa.Runtime.ProtoActor.Messages.Unit; -namespace Elsa.Runtime.ProtoActor.Actors; +namespace Elsa.Runtime.ProtoActor.Grains; /// -/// Instantiates and dispatches a workflow instance for execution. The workflow instance will be executed by . +/// Instantiates and executes a workflow instance for execution. /// -public class WorkflowDefinitionActor : IActor +public class WorkflowDefinitionGrain : WorkflowDefinitionGrainBase { private readonly IWorkflowRegistry _workflowRegistry; private readonly ICommandSender _commandSender; private readonly IIdentityGenerator _identityGenerator; private readonly ISystemClock _systemClock; + private readonly GrainClientFactory _grainClientFactory; - public WorkflowDefinitionActor(IWorkflowRegistry workflowRegistry, ICommandSender commandSender, IIdentityGenerator identityGenerator, ISystemClock systemClock) + public WorkflowDefinitionGrain( + IWorkflowRegistry workflowRegistry, + ICommandSender commandSender, + IIdentityGenerator identityGenerator, + ISystemClock systemClock, + GrainClientFactory grainClientFactory, + IContext context) : base(context) { _workflowRegistry = workflowRegistry; _commandSender = commandSender; _identityGenerator = identityGenerator; _systemClock = systemClock; + _grainClientFactory = grainClientFactory; } - - public Task ReceiveAsync(IContext context) => context.Message switch + + public override async Task Execute(ExecuteWorkflowDefinitionRequest request) { - ExecuteWorkflowDefinition i => OnExecuteWorkflowDefinitionAsync(context, i), - DispatchWorkflowDefinition i => OnDispatchWorkflowDefinitionAsync(context, i), - _ => Task.CompletedTask - }; - - private async Task OnExecuteWorkflowDefinitionAsync(IContext context, ExecuteWorkflowDefinition message) - { - var workflowDefinitionId = message.Id; - var cancellationToken = context.CancellationToken; + var workflowDefinitionId = request.Id; + var cancellationToken = Context.CancellationToken; var workflowInstance = await CreateWorkflowInstanceAsync(workflowDefinitionId, cancellationToken); - var executeWorkflowInstanceMessage = new ExecuteWorkflowInstance + var executeWorkflowInstanceMessage = new ExecuteWorkflowInstanceRequest { Id = workflowInstance.Id, - Input = message.Input + Input = request.Input }; - var response = await context.ClusterRequestAsync(workflowInstance.Id, GrainKinds.WorkflowInstance, executeWorkflowInstanceMessage, cancellationToken); - context.Respond(response); - } + var workflowInstanceGrainClient = Context.GetWorkflowInstanceGrain(workflowInstance.Id); + var workflowInstanceResponse = await workflowInstanceGrainClient.Execute(executeWorkflowInstanceMessage, cancellationToken); + + if (workflowInstanceResponse == null) + throw new TimeoutException("Did not receive a response from the WorkflowInstance actor within the configured amount of time."); - private async Task OnDispatchWorkflowDefinitionAsync(IContext context, DispatchWorkflowDefinition message) - { - var workflowDefinitionId = message.Id; - var cancellationToken = context.CancellationToken; - var workflowInstance = await CreateWorkflowInstanceAsync(workflowDefinitionId, cancellationToken); - - var dispatchWorkflowInstanceMessage = new DispatchWorkflowInstance + var response = new ExecuteWorkflowDefinitionResponse { - Id = workflowInstance.Id, - Input = message.Input + WorkflowState = workflowInstanceResponse.WorkflowState, }; - await context.ClusterRequestAsync(workflowInstance.Id, GrainKinds.WorkflowInstance, dispatchWorkflowInstanceMessage, cancellationToken); - context.Respond(new Unit()); + response.Bookmarks.AddRange(workflowInstanceResponse.Bookmarks); + return response; } private async Task CreateWorkflowInstanceAsync(string workflowDefinitionId, CancellationToken cancellationToken) diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowOperatorActor.cs b/src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowInstanceGrain.cs similarity index 61% rename from src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowOperatorActor.cs rename to src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowInstanceGrain.cs index 95b204317..9dad2cb80 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowOperatorActor.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowInstanceGrain.cs @@ -11,42 +11,45 @@ using Elsa.Persistence.Models; using Elsa.Persistence.Requests; using Elsa.Runtime.Contracts; using Elsa.Runtime.ProtoActor.Extensions; -using Elsa.Runtime.ProtoActor.Messages; +using Elsa.Runtime.ProtoActor.Services; +using Elsa.Runtime.Protos; using Elsa.Serialization; using Elsa.State; using Proto; -using Bookmark = Elsa.Runtime.ProtoActor.Messages.Bookmark; +using Bookmark = Elsa.Runtime.Protos.Bookmark; -namespace Elsa.Runtime.ProtoActor.Actors; +namespace Elsa.Runtime.ProtoActor.Grains; /// /// Executes a workflow instance. /// -public class WorkflowOperatorActor : IActor +public class WorkflowInstanceGrain : WorkflowInstanceGrainBase { private readonly IRequestSender _requestSender; private readonly IWorkflowRegistry _workflowRegistry; + private readonly GrainClientFactory _grainClientFactory; private readonly IWorkflowRunner _workflowRunner; private readonly WorkflowSerializerOptionsProvider _workflowSerializerOptionsProvider; - public WorkflowOperatorActor(IRequestSender requestSender, IWorkflowRegistry workflowRegistry, IWorkflowRunner workflowRunner, WorkflowSerializerOptionsProvider workflowSerializerOptionsProvider) + public WorkflowInstanceGrain( + IRequestSender requestSender, + IWorkflowRegistry workflowRegistry, + GrainClientFactory grainClientFactory, + IWorkflowRunner workflowRunner, + WorkflowSerializerOptionsProvider workflowSerializerOptionsProvider, + IContext context) : base(context) { _requestSender = requestSender; _workflowRegistry = workflowRegistry; + _grainClientFactory = grainClientFactory; _workflowRunner = workflowRunner; _workflowSerializerOptionsProvider = workflowSerializerOptionsProvider; } - public Task ReceiveAsync(IContext context) => context.Message switch + public override async Task Execute(ExecuteWorkflowInstanceRequest request) { - ExecuteWorkflowInstance m => OnExecuteWorkflowInstance(context, m), - _ => Task.CompletedTask - }; - - private async Task OnExecuteWorkflowInstance(IContext context, ExecuteWorkflowInstance message) - { - var workflowInstanceId = message.Id; - var cancellationToken = context.CancellationToken; + var workflowInstanceId = request.Id; + var cancellationToken = Context.CancellationToken; var workflowInstance = await _requestSender.RequestAsync(new FindWorkflowInstance(workflowInstanceId), cancellationToken); if (workflowInstance == null) @@ -59,15 +62,30 @@ public class WorkflowOperatorActor : IActor throw new Exception($"No workflow definition found with ID {workflowDefinitionId}"); var workflowState = workflowInstance.WorkflowState; - var bookmarkMessage = message.Bookmark; - var input = message.Input?.Deserialize(); - var executionResult = await ExecuteAsync(workflow, workflowState, bookmarkMessage, input, cancellationToken); + var bookmarkMessage = request.Bookmark; + var input = request.Input?.Deserialize(); + var executionResult = await ExecuteAsync(workflow, workflowState, bookmarkMessage, input!, cancellationToken); var response = MapResult(executionResult); - context.Respond(response); + return response; } - private ExecuteWorkflowResponse MapResult(ExecuteWorkflowResult result) + // public override async Task Dispatch(DispatchWorkflowInstanceRequest request) + // { + // var workflowInstanceId = request.Id; + // //var pid = GetWorkflowOperatorPid(Context, workflowInstanceId); + // + // var executeWorkflowInstanceMessage = new ExecuteWorkflowInstanceRequest + // { + // Id = workflowInstanceId, + // Bookmark = request.Bookmark, + // Input = request.Input + // }; + + //Context.Send(pid, executeWorkflowInstanceMessage); + //} + + private ExecuteWorkflowInstanceResponse MapResult(InvokeWorkflowResult result) { var bookmarks = result.Bookmarks.Select(x => new Bookmark { @@ -79,10 +97,10 @@ public class WorkflowOperatorActor : IActor ActivityInstanceId = x.ActivityInstanceId, CallbackMethodName = x.CallbackMethodName }); - + var options = _workflowSerializerOptionsProvider.CreatePersistenceOptions(); - var response = new ExecuteWorkflowResponse + var response = new ExecuteWorkflowInstanceResponse { WorkflowState = new Json { @@ -95,7 +113,7 @@ public class WorkflowOperatorActor : IActor return response; } - private async Task ExecuteAsync(Workflow workflow, WorkflowState workflowState, Bookmark? bookmarkMessage, IDictionary? input, CancellationToken cancellationToken) + private async Task ExecuteAsync(Workflow workflow, WorkflowState workflowState, Bookmark? bookmarkMessage, IDictionary? input, CancellationToken cancellationToken) { if (bookmarkMessage == null) return await _workflowRunner.RunAsync(workflow, workflowState, input, cancellationToken); diff --git a/src/runtime/Elsa.Runtime.ProtoActor/HostedServices/WorkflowServerHost.cs b/src/runtime/Elsa.Runtime.ProtoActor/HostedServices/WorkflowServerHost.cs index 4efeb91c7..cc2a8743f 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/HostedServices/WorkflowServerHost.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/HostedServices/WorkflowServerHost.cs @@ -6,17 +6,13 @@ using Proto.Cluster; namespace Elsa.Runtime.ProtoActor.HostedServices; -public class WorkflowServerHost : BackgroundService +/// +/// Starts the current member in the cluster. +/// +public class WorkflowServerHost : IHostedService { private readonly ActorSystem _actorSystem; - - public WorkflowServerHost(ActorSystem actorSystem) - { - _actorSystem = actorSystem; - } - - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - await _actorSystem.Cluster().StartMemberAsync(); - } + public WorkflowServerHost(ActorSystem actorSystem) => _actorSystem = actorSystem; + public async Task StartAsync(CancellationToken cancellationToken) => await _actorSystem.Cluster().StartMemberAsync(); + public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Protos/Grains.proto b/src/runtime/Elsa.Runtime.ProtoActor/Protos/Grains.proto new file mode 100644 index 000000000..59f05910f --- /dev/null +++ b/src/runtime/Elsa.Runtime.ProtoActor/Protos/Grains.proto @@ -0,0 +1,14 @@ +syntax = "proto3"; + +option csharp_namespace = "Elsa.Runtime.Protos"; + +import "google/protobuf/empty.proto"; +import "Messages.proto"; + +service WorkflowDefinitionGrain { + rpc Execute (ExecuteWorkflowDefinitionRequest) returns (ExecuteWorkflowDefinitionResponse); +} + +service WorkflowInstanceGrain { + rpc Execute (ExecuteWorkflowInstanceRequest) returns (ExecuteWorkflowInstanceResponse); +} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime.ProtoActor/messages.proto b/src/runtime/Elsa.Runtime.ProtoActor/Protos/Messages.proto similarity index 65% rename from src/runtime/Elsa.Runtime.ProtoActor/messages.proto rename to src/runtime/Elsa.Runtime.ProtoActor/Protos/Messages.proto index c0218c38f..1898e0a95 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/messages.proto +++ b/src/runtime/Elsa.Runtime.ProtoActor/Protos/Messages.proto @@ -1,41 +1,31 @@ syntax = "proto3"; import "google/protobuf/wrappers.proto"; package Elsa.Runtime.ProtoActor.Messages; -option csharp_namespace = "Elsa.Runtime.ProtoActor.Messages"; +option csharp_namespace = "Elsa.Runtime.Protos"; // Empty response type. message Unit {} -message ExecuteWorkflowDefinition { +message ExecuteWorkflowDefinitionRequest { string id = 1; string versionOptions = 2; optional Input input = 3; } -message DispatchWorkflowDefinition { - string id = 1; - string versionOptions = 2; - optional Input input = 3; -} - -message ExecuteWorkflowInstance { - string id = 1; - optional Bookmark bookmark = 2; - optional Input input = 3; -} - -message DispatchWorkflowInstance { - string id = 1; - optional Bookmark bookmark = 2; - optional Input input = 3; -} - -message ExecuteWorkflowResponse { +message ExecuteWorkflowDefinitionResponse { Json workflowState = 1; repeated Bookmark bookmarks = 2; } -message DispatchWorkflowResponse { +message ExecuteWorkflowInstanceRequest { + string id = 1; + optional Bookmark bookmark = 2; + optional Input input = 3; +} + +message ExecuteWorkflowInstanceResponse { + Json workflowState = 1; + repeated Bookmark bookmarks = 2; } message HandleStimulusRequest { diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Services/GrainClientFactory.cs b/src/runtime/Elsa.Runtime.ProtoActor/Services/GrainClientFactory.cs new file mode 100644 index 000000000..f31cb58da --- /dev/null +++ b/src/runtime/Elsa.Runtime.ProtoActor/Services/GrainClientFactory.cs @@ -0,0 +1,18 @@ +using Elsa.Persistence.Models; +using Elsa.Runtime.Protos; +using Proto.Cluster; + +namespace Elsa.Runtime.ProtoActor.Services; + +public class GrainClientFactory +{ + private readonly Cluster _cluster; + + public GrainClientFactory(Cluster cluster) + { + _cluster = cluster; + } + + public WorkflowDefinitionGrainClient CreateWorkflowDefinitionGrainClient(string workflowDefinitionId, VersionOptions versionOptions) => _cluster.GetWorkflowDefinitionGrain($"workflow-definition:{workflowDefinitionId}:{versionOptions.ToString()}"); + public WorkflowInstanceGrainClient CreateWorkflowInstanceGrainClient(string workflowInstanceId) => _cluster.GetWorkflowInstanceGrain($"workflow-instance:{workflowInstanceId}"); +} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs b/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs index 4dd8e11f3..d3a4c181e 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs @@ -1,110 +1,85 @@ +using System; using System.Linq; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Elsa.Models; -using Elsa.Persistence.Models; using Elsa.Runtime.Contracts; using Elsa.Runtime.Models; using Elsa.Runtime.ProtoActor.Extensions; -using Elsa.Runtime.ProtoActor.Messages; +using Elsa.Runtime.Protos; using Elsa.State; +using Proto; using Proto.Cluster; +using Bookmark = Elsa.Models.Bookmark; +using ProtoBookmark = Elsa.Runtime.Protos.Bookmark; namespace Elsa.Runtime.ProtoActor.Services; -using Bookmark = Elsa.Models.Bookmark; -using BookmarkMessage = Messages.Bookmark; - public class ProtoActorWorkflowInvoker : IWorkflowInvoker { private readonly Cluster _cluster; + private readonly GrainClientFactory _grainClientFactory; - public ProtoActorWorkflowInvoker(Cluster cluster) + public ProtoActorWorkflowInvoker(Cluster cluster, GrainClientFactory grainClientFactory) { _cluster = cluster; + _grainClientFactory = grainClientFactory; } - public async Task ExecuteAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) + public async Task InvokeAsync(InvokeWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) { var (definitionId, versionOptions, input) = request; - var name = GetActorName(definitionId, versionOptions); - - var message = new ExecuteWorkflowDefinition + + var message = new ExecuteWorkflowDefinitionRequest { Id = definitionId, VersionOptions = versionOptions.ToString(), Input = input!?.Serialize() }; - var response = await _cluster.RequestAsync(name, GrainKinds.WorkflowDefinition, message, cancellationToken); + var client = _grainClientFactory.CreateWorkflowDefinitionGrainClient(definitionId, versionOptions); + var response = await client.Execute(message, CancellationTokens.FromSeconds(100)); + + if (response == null) + throw new TimeoutException("Did not receive a response from the WorkflowDefinition actor within the configured amount of time."); + var workflowState = JsonSerializer.Deserialize(response.WorkflowState.Text)!; var bookmarks = response.Bookmarks.Select(MapBookmark).ToList(); - return new ExecuteWorkflowResult(workflowState, bookmarks); + return new InvokeWorkflowResult(workflowState, bookmarks); } - public async Task ExecuteAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default) + public async Task InvokeAsync(InvokeWorkflowInstanceRequest request, CancellationToken cancellationToken = default) { var (instanceId, bookmark, input) = request; - var name = GetActorName(instanceId); var bookmarkMessage = MapBookmark(bookmark); - var message = new ExecuteWorkflowInstance - { - Id = instanceId, - Bookmark = bookmarkMessage, - Input = input?.Serialize() - }; - - var response = await _cluster.RequestAsync(name, GrainKinds.WorkflowInstance, message, cancellationToken); - var workflowState = JsonSerializer.Deserialize(response.WorkflowState.Text)!; - var bookmarks = response.Bookmarks.Select(MapBookmark).ToList(); - - return new ExecuteWorkflowResult(workflowState, bookmarks); - } - - public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) - { - var (definitionId, versionOptions, input) = request; - var name = GetActorName(definitionId, versionOptions); - - var message = new DispatchWorkflowDefinition - { - Id = definitionId, - VersionOptions = versionOptions.ToString(), - Input = input!?.Serialize() - }; - - await _cluster.RequestAsync(name, GrainKinds.WorkflowDefinition, message, cancellationToken); - - return new DispatchWorkflowResult(); - } - - public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default) - { - var (instanceId, bookmark, input) = request; - var name = GetActorName(instanceId); - var bookmarkMessage = MapBookmark(bookmark); - - var message = new DispatchWorkflowInstance + var message = new ExecuteWorkflowInstanceRequest { Id = instanceId, Bookmark = bookmarkMessage, Input = input!?.Serialize() }; - await _cluster.RequestAsync(name, GrainKinds.WorkflowInstance, message, cancellationToken); + var client = _grainClientFactory.CreateWorkflowInstanceGrainClient(instanceId); + var response = await client.Execute(message, cancellationToken); + + if (response == null) + throw new TimeoutException("Did not receive a response from the WorkflowInstance actor within the configured amount of time."); + + var workflowState = JsonSerializer.Deserialize(response.WorkflowState.Text)!; + var bookmarks = response.Bookmarks.Select(MapBookmark).ToList(); - return new DispatchWorkflowResult(); + return new InvokeWorkflowResult(workflowState, bookmarks); } - - private BookmarkMessage? MapBookmark(Bookmark? bookmark) + + private ProtoBookmark? MapBookmark(Bookmark? bookmark) { if (bookmark == null) return null; - return new BookmarkMessage + return new ProtoBookmark { Id = bookmark.Id, Name = bookmark.Name, @@ -116,19 +91,16 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker }; } - private Bookmark MapBookmark(BookmarkMessage bookmarkMessage) + private Bookmark MapBookmark(ProtoBookmark protoBookmark) { return new Bookmark( - bookmarkMessage.Id, - bookmarkMessage.Name, - bookmarkMessage.Hash, - bookmarkMessage.Payload, - bookmarkMessage.ActivityId, - bookmarkMessage.ActivityInstanceId, - bookmarkMessage.CallbackMethodName + protoBookmark.Id, + protoBookmark.Name, + protoBookmark.Hash, + protoBookmark.Payload, + protoBookmark.ActivityId, + protoBookmark.ActivityInstanceId, + protoBookmark.CallbackMethodName ); } - - private static string GetActorName(string definitionId, VersionOptions versionOptions) => $"workflow-definition:{definitionId}:{versionOptions.ToString()}"; - private static string GetActorName(string instanceId) => $"workflow-instance:{instanceId}"; } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Contracts/ExecuteWorkflowInstructionResult.cs b/src/runtime/Elsa.Runtime/Contracts/ExecuteWorkflowInstructionResult.cs index aee1cfebd..f2b900ac6 100644 --- a/src/runtime/Elsa.Runtime/Contracts/ExecuteWorkflowInstructionResult.cs +++ b/src/runtime/Elsa.Runtime/Contracts/ExecuteWorkflowInstructionResult.cs @@ -2,4 +2,4 @@ using Elsa.Models; namespace Elsa.Runtime.Contracts; -public record ExecuteWorkflowInstructionResult(Workflow Workflow, ExecuteWorkflowResult ExecuteWorkflowResult); \ No newline at end of file +public record ExecuteWorkflowInstructionResult(Workflow Workflow, InvokeWorkflowResult InvokeWorkflowResult); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Contracts/IWorkflowDispatcher.cs b/src/runtime/Elsa.Runtime/Contracts/IWorkflowDispatcher.cs new file mode 100644 index 000000000..d7c6eed17 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Contracts/IWorkflowDispatcher.cs @@ -0,0 +1,13 @@ +using Elsa.Models; +using Elsa.Runtime.Models; + +namespace Elsa.Runtime.Contracts; + +/// +/// Posts a message to a queue to invoke a specified workflow. +/// +public interface IWorkflowDispatcher +{ + Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default); + Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Contracts/IWorkflowInvoker.cs b/src/runtime/Elsa.Runtime/Contracts/IWorkflowInvoker.cs index 1388a817d..a72e76838 100644 --- a/src/runtime/Elsa.Runtime/Contracts/IWorkflowInvoker.cs +++ b/src/runtime/Elsa.Runtime/Contracts/IWorkflowInvoker.cs @@ -3,10 +3,18 @@ using Elsa.Runtime.Models; namespace Elsa.Runtime.Contracts; +/// +/// Invokes a workflow using a configured runtime (e.g. ProtoActor). +/// public interface IWorkflowInvoker { - Task ExecuteAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default); - Task ExecuteAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default); - Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default); - Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default); + /// + /// Creates a workflow instance of the specified workflow definition and then invokes the workflow instance. + /// + Task InvokeAsync(InvokeWorkflowDefinitionRequest request, CancellationToken cancellationToken = default); + + /// + /// Invokes the specified workflow instance. + /// + Task InvokeAsync(InvokeWorkflowInstanceRequest request, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Contracts/IWorkflowService.cs b/src/runtime/Elsa.Runtime/Contracts/IWorkflowService.cs index 62b1c27f7..85baed93f 100644 --- a/src/runtime/Elsa.Runtime/Contracts/IWorkflowService.cs +++ b/src/runtime/Elsa.Runtime/Contracts/IWorkflowService.cs @@ -1,5 +1,6 @@ using Elsa.Models; using Elsa.Persistence.Models; +using Elsa.Runtime.Models; namespace Elsa.Runtime.Contracts; @@ -8,10 +9,10 @@ namespace Elsa.Runtime.Contracts; /// public interface IWorkflowService { - Task ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default); - Task ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default); - Task DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default); - Task DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default); + Task ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default); + Task ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default); + Task DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default); + Task DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default); Task> ExecuteStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken); Task> DispatchStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken); diff --git a/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs b/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs index 91d56f83e..6efef1d13 100644 --- a/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs +++ b/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs @@ -1,13 +1,16 @@ +using System.Threading.Channels; using Elsa.Expressions; using Elsa.Mediator.Extensions; using Elsa.Runtime.Contracts; using Elsa.Runtime.HostedServices; using Elsa.Runtime.Interpreters; +using Elsa.Runtime.Models; using Elsa.Runtime.Options; using Elsa.Runtime.Services; using Elsa.Runtime.Stimuli.Handlers; using Elsa.Runtime.WorkflowProviders; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; namespace Elsa.Runtime.Extensions; @@ -24,6 +27,8 @@ public static class ServiceCollectionExtensions .AddSingleton() .AddSingleton() .AddSingleton() + .AddSingleton(sp => sp.GetRequiredService>().Value.WorkflowInvokerFactory(sp)) + .AddSingleton(sp => sp.GetRequiredService>().Value.WorkflowDispatcherFactory(sp)) // Stimulus handlers. .AddStimulusHandler() @@ -42,13 +47,29 @@ public static class ServiceCollectionExtensions // Domain event handlers. .AddNotificationHandlersFrom(typeof(ServiceCollectionExtensions)) - + // Hosted Services. .AddHostedService() - .AddHostedService(); + .AddHostedService() + .AddHostedService() + .AddHostedService() + + // Channels for dispatching workflows in-memory. + .CreateChannel() + .CreateChannel() ; } + private static IServiceCollection CreateChannel(this IServiceCollection services) => + services + .AddSingleton(CreateChannel()) + .AddSingleton(CreateChannelReader) + .AddSingleton(CreateChannelWriter); + + private static Channel CreateChannel() => Channel.CreateUnbounded(new UnboundedChannelOptions()); + private static ChannelReader CreateChannelReader(IServiceProvider serviceProvider) => serviceProvider.GetRequiredService>().Reader; + private static ChannelWriter CreateChannelWriter(IServiceProvider serviceProvider) => serviceProvider.GetRequiredService>().Writer; + private static IServiceCollection AddDefaultExpressionHandlers(this IServiceCollection services) => services .AddExpressionHandler() diff --git a/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowDefinitionWorker.cs b/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowDefinitionWorker.cs new file mode 100644 index 000000000..62eca7554 --- /dev/null +++ b/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowDefinitionWorker.cs @@ -0,0 +1,42 @@ +using System.Threading.Channels; +using Elsa.Runtime.Contracts; +using Elsa.Runtime.Models; +using Elsa.Runtime.Services; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace Elsa.Runtime.HostedServices; + +public class DispatchedWorkflowDefinitionWorker : BackgroundService +{ + private readonly ChannelReader _channelReader; + private readonly IWorkflowInvoker _workflowInvoker; + private readonly ILogger _logger; + + public DispatchedWorkflowDefinitionWorker( + ChannelReader channelReader, + IWorkflowInvoker workflowInvoker, + ILogger logger) + { + _channelReader = channelReader; + _workflowInvoker = workflowInvoker; + _logger = logger; + } + + protected override async Task ExecuteAsync(CancellationToken cancellationToken) + { + await foreach (var (definitionId, versionOptions, input) in _channelReader.ReadAllAsync(cancellationToken)) + { + var request = new InvokeWorkflowDefinitionRequest(definitionId, versionOptions, input); + + try + { + await _workflowInvoker.InvokeAsync(request, cancellationToken); + } + catch (Exception e) + { + _logger.LogError(e, "An unhandled exception occured while processing the queue"); + } + } + } +} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowInstanceWorker.cs b/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowInstanceWorker.cs new file mode 100644 index 000000000..c1f9614e8 --- /dev/null +++ b/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowInstanceWorker.cs @@ -0,0 +1,41 @@ +using System.Threading.Channels; +using Elsa.Runtime.Contracts; +using Elsa.Runtime.Models; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace Elsa.Runtime.HostedServices; + +public class DispatchedWorkflowInstanceWorker : BackgroundService +{ + private readonly ChannelReader _channelReader; + private readonly IWorkflowInvoker _workflowInvoker; + private readonly ILogger _logger; + + public DispatchedWorkflowInstanceWorker( + ChannelReader channelReader, + IWorkflowInvoker workflowInvoker, + ILogger logger) + { + _channelReader = channelReader; + _workflowInvoker = workflowInvoker; + _logger = logger; + } + + protected override async Task ExecuteAsync(CancellationToken cancellationToken) + { + await foreach (var (instanceId, bookmark, input) in _channelReader.ReadAllAsync(cancellationToken)) + { + var request = new InvokeWorkflowInstanceRequest(instanceId, bookmark, input); + + try + { + await _workflowInvoker.InvokeAsync(request, cancellationToken); + } + catch (Exception e) + { + _logger.LogError(e, "An unhandled exception occured while processing the queue"); + } + } + } +} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Interpreters/TriggerWorkflowInstructionInterpreter.cs b/src/runtime/Elsa.Runtime/Interpreters/TriggerWorkflowInstructionInterpreter.cs index 1888fc36b..6855df43c 100644 --- a/src/runtime/Elsa.Runtime/Interpreters/TriggerWorkflowInstructionInterpreter.cs +++ b/src/runtime/Elsa.Runtime/Interpreters/TriggerWorkflowInstructionInterpreter.cs @@ -13,12 +13,18 @@ public record TriggerWorkflowInstruction(WorkflowTrigger WorkflowTrigger, IDicti public class TriggerWorkflowInstructionInterpreter : WorkflowInstructionInterpreter { private readonly IWorkflowInvoker _workflowInvoker; + private readonly IWorkflowDispatcher _workflowDispatcher; private readonly IWorkflowRegistry _workflowRegistry; private readonly ILogger _logger; - public TriggerWorkflowInstructionInterpreter(IWorkflowInvoker workflowInvoker, IWorkflowRegistry workflowRegistry, ILogger logger) + public TriggerWorkflowInstructionInterpreter( + IWorkflowInvoker workflowInvoker, + IWorkflowDispatcher workflowDispatcher, + IWorkflowRegistry workflowRegistry, + ILogger logger) { _workflowInvoker = workflowInvoker; + _workflowDispatcher = workflowDispatcher; _workflowRegistry = workflowRegistry; _logger = logger; } @@ -35,8 +41,8 @@ public class TriggerWorkflowInstructionInterpreter : WorkflowInstructionInterpre return null; // Execute workflow. - var executeRequest = new ExecuteWorkflowDefinitionRequest(workflowId, VersionOptions.Published, instruction.Input); - var workflowExecutionResult = await _workflowInvoker.ExecuteAsync(executeRequest, cancellationToken); + var executeRequest = new InvokeWorkflowDefinitionRequest(workflowId, VersionOptions.Published, instruction.Input); + var workflowExecutionResult = await _workflowInvoker.InvokeAsync(executeRequest, cancellationToken); return new ExecuteWorkflowInstructionResult(workflow, workflowExecutionResult); } @@ -54,7 +60,7 @@ public class TriggerWorkflowInstructionInterpreter : WorkflowInstructionInterpre // Execute workflow. var dispatchRequest = new DispatchWorkflowDefinitionRequest(definitionId, VersionOptions.Published, instruction.Input); - await _workflowInvoker.DispatchAsync(dispatchRequest, cancellationToken); + await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken); return new DispatchWorkflowInstructionResult(); } diff --git a/src/runtime/Elsa.Runtime/Models/DispatchWorkflowDefinitionResponse.cs b/src/runtime/Elsa.Runtime/Models/DispatchWorkflowDefinitionResponse.cs new file mode 100644 index 000000000..802d31a7e --- /dev/null +++ b/src/runtime/Elsa.Runtime/Models/DispatchWorkflowDefinitionResponse.cs @@ -0,0 +1,3 @@ +namespace Elsa.Runtime.Models; + +public record DispatchWorkflowDefinitionResponse(); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/DispatchWorkflowInstanceResponse.cs b/src/runtime/Elsa.Runtime/Models/DispatchWorkflowInstanceResponse.cs new file mode 100644 index 000000000..222905525 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Models/DispatchWorkflowInstanceResponse.cs @@ -0,0 +1,3 @@ +namespace Elsa.Runtime.Models; + +public record DispatchWorkflowInstanceResponse(); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/ExecuteWorkflowDefinitionRequest.cs b/src/runtime/Elsa.Runtime/Models/ExecuteWorkflowDefinitionRequest.cs deleted file mode 100644 index 5c0e6eb95..000000000 --- a/src/runtime/Elsa.Runtime/Models/ExecuteWorkflowDefinitionRequest.cs +++ /dev/null @@ -1,5 +0,0 @@ -using Elsa.Persistence.Models; - -namespace Elsa.Runtime.Models; - -public record ExecuteWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary? Input = default); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/ExecuteWorkflowInstanceRequest.cs b/src/runtime/Elsa.Runtime/Models/ExecuteWorkflowInstanceRequest.cs deleted file mode 100644 index 5f24d12cf..000000000 --- a/src/runtime/Elsa.Runtime/Models/ExecuteWorkflowInstanceRequest.cs +++ /dev/null @@ -1,5 +0,0 @@ -using Elsa.Models; - -namespace Elsa.Runtime.Models; - -public record ExecuteWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary? Input = default); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/InvokeWorkflowDefinitionRequest.cs b/src/runtime/Elsa.Runtime/Models/InvokeWorkflowDefinitionRequest.cs new file mode 100644 index 000000000..f0fcd25f0 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Models/InvokeWorkflowDefinitionRequest.cs @@ -0,0 +1,5 @@ +using Elsa.Persistence.Models; + +namespace Elsa.Runtime.Models; + +public record InvokeWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary? Input = default); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/InvokeWorkflowInstanceRequest.cs b/src/runtime/Elsa.Runtime/Models/InvokeWorkflowInstanceRequest.cs new file mode 100644 index 000000000..7a730f93b --- /dev/null +++ b/src/runtime/Elsa.Runtime/Models/InvokeWorkflowInstanceRequest.cs @@ -0,0 +1,5 @@ +using Elsa.Models; + +namespace Elsa.Runtime.Models; + +public record InvokeWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary? Input = default); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Options/WorkflowRuntimeOptions.cs b/src/runtime/Elsa.Runtime/Options/WorkflowRuntimeOptions.cs index e2c20e47b..115c309d7 100644 --- a/src/runtime/Elsa.Runtime/Options/WorkflowRuntimeOptions.cs +++ b/src/runtime/Elsa.Runtime/Options/WorkflowRuntimeOptions.cs @@ -1,11 +1,23 @@ using Elsa.Runtime.Contracts; +using Elsa.Runtime.Services; +using Microsoft.Extensions.DependencyInjection; namespace Elsa.Runtime.Options; public class WorkflowRuntimeOptions { /// - /// A list of workflow builders configured at application startup. + /// A list of workflow builders configured during application startup. /// public IDictionary> Workflows { get; set; } = new Dictionary>(); + + /// + /// A factory that instantiates a concrete . + /// + public Func WorkflowInvokerFactory { get; set; } = sp => ActivatorUtilities.CreateInstance(sp); + + /// + /// A factory that instantiates a concrete . + /// + public Func WorkflowDispatcherFactory { get; set; } = sp => ActivatorUtilities.CreateInstance(sp); } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Services/DefaultWorkflowInvoker.cs b/src/runtime/Elsa.Runtime/Services/DefaultWorkflowInvoker.cs new file mode 100644 index 000000000..95d03e967 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Services/DefaultWorkflowInvoker.cs @@ -0,0 +1,21 @@ +using Elsa.Models; +using Elsa.Runtime.Contracts; +using Elsa.Runtime.Models; + +namespace Elsa.Runtime.Services; + +/// +/// A basic implementation that directly executes the specified workflow in local memory (as opposed elsewhere in some cluster). +/// +public class DefaultWorkflowInvoker : IWorkflowInvoker +{ + public Task InvokeAsync(InvokeWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task InvokeAsync(InvokeWorkflowInstanceRequest request, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } +} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Services/TaskBasedWorkflowDispatcher.cs b/src/runtime/Elsa.Runtime/Services/TaskBasedWorkflowDispatcher.cs new file mode 100644 index 000000000..382ef4905 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Services/TaskBasedWorkflowDispatcher.cs @@ -0,0 +1,34 @@ +using System.Threading.Channels; +using Elsa.Runtime.Contracts; +using Elsa.Runtime.Models; + +namespace Elsa.Runtime.Services; + +/// +/// A simple implementation that queues the specified request for workflow execution on a non-durable background worker. +/// +public class TaskBasedWorkflowDispatcher : IWorkflowDispatcher +{ + private readonly ChannelWriter _workflowDefinitionChannelWriter; + private readonly ChannelWriter _workflowInstanceChannelWriter; + + public TaskBasedWorkflowDispatcher( + ChannelWriter workflowDefinitionChannelWriter, + ChannelWriter workflowInstanceChannelWriter) + { + _workflowDefinitionChannelWriter = workflowDefinitionChannelWriter; + _workflowInstanceChannelWriter = workflowInstanceChannelWriter; + } + + public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) + { + await _workflowDefinitionChannelWriter.WriteAsync(request, cancellationToken); + return new DispatchWorkflowDefinitionResponse(); + } + + public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default) + { + await _workflowInstanceChannelWriter.WriteAsync(request, cancellationToken); + return new DispatchWorkflowInstanceResponse(); + } +} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Services/WorkflowService.cs b/src/runtime/Elsa.Runtime/Services/WorkflowService.cs index 2d8f62bb2..9c1ea931c 100644 --- a/src/runtime/Elsa.Runtime/Services/WorkflowService.cs +++ b/src/runtime/Elsa.Runtime/Services/WorkflowService.cs @@ -8,38 +8,44 @@ namespace Elsa.Runtime.Services; public class WorkflowService : IWorkflowService { private readonly IWorkflowInvoker _workflowInvoker; + private readonly IWorkflowDispatcher _workflowDispatcher; private readonly IWorkflowInstructionExecutor _workflowInstructionExecutor; private readonly IStimulusInterpreter _stimulusInterpreter; - public WorkflowService(IWorkflowInvoker workflowInvoker, IWorkflowInstructionExecutor workflowInstructionExecutor, IStimulusInterpreter stimulusInterpreter) + public WorkflowService( + IWorkflowInvoker workflowInvoker, + IWorkflowDispatcher workflowDispatcher, + IWorkflowInstructionExecutor workflowInstructionExecutor, + IStimulusInterpreter stimulusInterpreter) { _workflowInvoker = workflowInvoker; + _workflowDispatcher = workflowDispatcher; _workflowInstructionExecutor = workflowInstructionExecutor; _stimulusInterpreter = stimulusInterpreter; } - public async Task ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default) + public async Task ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default) { - var executeRequest = new ExecuteWorkflowDefinitionRequest(definitionId, versionOptions, input); - return await _workflowInvoker.ExecuteAsync(executeRequest, cancellationToken); + var executeRequest = new InvokeWorkflowDefinitionRequest(definitionId, versionOptions, input); + return await _workflowInvoker.InvokeAsync(executeRequest, cancellationToken); } - public async Task ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default) + public async Task ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default) { - var request = new ExecuteWorkflowInstanceRequest(instanceId, bookmark, input); - return await _workflowInvoker.ExecuteAsync(request, cancellationToken); + var request = new InvokeWorkflowInstanceRequest(instanceId, bookmark, input); + return await _workflowInvoker.InvokeAsync(request, cancellationToken); } - public async Task DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default) + public async Task DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default) { var executeRequest = new DispatchWorkflowDefinitionRequest(definitionId, versionOptions, input); - return await _workflowInvoker.DispatchAsync(executeRequest, cancellationToken); + return await _workflowDispatcher.DispatchAsync(executeRequest, cancellationToken); } - public async Task DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default) + public async Task DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default) { var request = new DispatchWorkflowInstanceRequest(instanceId, bookmark, input); - return await _workflowInvoker.DispatchAsync(request, cancellationToken); + return await _workflowDispatcher.DispatchAsync(request, cancellationToken); } public async Task> ExecuteStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken = default) diff --git a/src/samples/aspnet/Elsa.Samples.Web1/Elsa.Samples.Web1.csproj b/src/samples/aspnet/Elsa.Samples.Web1/Elsa.Samples.Web1.csproj index b7641f64b..0d3354af3 100644 --- a/src/samples/aspnet/Elsa.Samples.Web1/Elsa.Samples.Web1.csproj +++ b/src/samples/aspnet/Elsa.Samples.Web1/Elsa.Samples.Web1.csproj @@ -6,11 +6,13 @@ + +