From ef3431c57aab2b1ef56b6e62760b6ab431e5a3d2 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Fri, 16 Feb 2024 10:56:30 +0100 Subject: [PATCH] Implement dispatch channels (#4949) * Add DispatchWorkflowOptions and update MassTransit configuration Introduced a new class `DispatchWorkflowOptions` to provide workflow dispatch options. Updated MassTransit configuration to include NET6.0 and NET7.0 support, and ensure correct MassTransit version usage per target framework version. * Add support for configurable MassTransit message dispatching Implemented a feature which allows for configurable MassTransit message dispatching. Added support for specifying channels and message brokers. Message dispatching code was massively refactored and relevant endpoints and response models were updated to support new message dispatching features. * Add IEndpointChannelFormatter interface and implementation An interface for formatting channel queue names, 'IEndpointChannelFormatter', has been added, along with its default implementation 'DefaultEndpointChannelFormatter'. The code that uses hardcoded queue name formatting has been modified to use the new formatter instead, making it more configurable and reusable. The implementation of the formatter uses the 'Humanizer' library to kebab-case the channel names. * Add channel dispatch option to workflow activities Introduced `WorkflowDispatcherChannelOptionsProvider` to provide dropdown channel options for workflow dispatch-related activities. Updated `DispatchWorkflow` and `BulkDispatchWorkflows` activities to include a new dropdown input for specifying a dispatch channel. Also, included the selected channel name in the `DispatchWorkflowOptions` during workflow dispatching process. * Update MassTransit configurations and remove unused code The MassTransit setup in Elsa.MassTransit module has been simplified by removing conditional code for different .NET versions. Additionally, unused parameter '__X_Channel' in 'DispatchWorkflowDefinition' was removed. Lastly, the MassTransit broker in Elsa.Server.Web was switched from RabbitMq to AzureServiceBus. * Refactor dispatch workflow classes and methods Simplified the class, method and variable names related to workflow dispatching in the Elsa.Workflows.Runtime module. For example, the 'WorkflowDispatcherChannelDescriptor' class was renamed to 'DispatcherChannel'. This refactoring was performed to make code more readable and maintainable by removing redundant wording in the naming convention. * Update GitHub Workflow to support feature and issue branches The workflow changes add support for feature and issue branches. Now, it extracts the branch name and verifies the commit exists in the given branch rather than just 'main'. The versioning scheme is also modified to include the branch name and not just the run number. * Add 'bug/*' to triggering branches in packages workflow The 'bug/*' pattern was missing from the triggers that initiate the GitHub actions within our packages workflow. This update includes any branch with a 'bug/' prefix to the list, allowing bug-related branches to start jobs in our CI/CD pipeline. * Remove 'issue/*' and 'bug/*' branches from packages workflow The 'issue/*' and 'bug/*' branches have been removed from the GitHub action workflow for packages. This change was made to simplify the workflow and optimize the triggering of package building. * Update GitHub workflow to handle main branch versioning This commit modifies the GitHub workflow script to accommodate changes when the branch name is "main." If the branch name is "main", a preview version is used. It also updates script execution to print the branch name for easier debugging and verifies commit existence on the correct branch instead of dispatch channels. * Enclose branch names in quotes in packages.yml The update modifies the branch names in the packages.yml GitHub Actions workflow file. The change consists of enclosing the branch names 'main' and 'feature/*' in single quotes, ensuring compatibility and preventing potential string interpretation issues. * Update GitHub workflows package configuration The workflows package configuration has been updated to specifically watch for changes on 'feature/dispatch-channels' rather than on all feature branches. This change will prevent unnecessary builds on less relevant feature branches. * Update trigger branches in packages workflow The triggering branches in the packages workflow have been updated. Previously, only changes in the 'main' and 'feature/dispatch-channels' would trigger the workflow, now any 'feature/*' branch will. This will cause more frequent and comprehensive testing. * Add 'patch/*' to workflow trigger branches This update adds 'patch/*' to the list of branches in .github/workflows/packages.yml that can trigger the workflow. It will allow the workflow to be initiated not just for main and feature branches, but also for patches. * Add 'preview/*' to workflow triggers This commit adds a new trigger for the GitHub Actions workflow. It now also responds to push events on 'preview/*' branches, allowing for automated testing and building of these preview branches. * Update branch name extraction in GitHub Actions The extraction of the branch name has been slightly modified in the packages.yml GitHub workflow file. This alteration ensures the correct branch name is obtained for further processing within the workflow without any discrepancy. * Update branch name extraction in packages.yml Corrected the syntax for extracting the branch name within the packages.yml github workflow file. Added an extra line to print out the ref which might be useful for debugging. * Update branch name extraction in GitHub workflow The commit simplifies the way the branch name is being extracted from the GitHub ref in the packages.yml workflow file. The new method employs straightforward string manipulation, making it easier to understand and debug in case of potential issues. * Add extraction of branch name in workflow Added a new line in the GitHub workflow file (.github/workflows/packages.yml) to extract the last part after the final slash from the branch name. This enhancement allows cleaner naming conventions, especially in cases where branches are named feature/issue-123, as it will only retain 'issue-123'. * Update package naming in Github workflow The Github workflow has been updated to handle package naming more effectively. Previously, the branch name was used directly for package versioning. Now, the last part of the branch name is extracted and used as the package prefix. If the branch name is "main", the package prefix is set to "preview". * Move and add environment variable assignments The placement of the assignment for BRANCH_NAME environment variable was moved for better readability. Additionally, the PACKAGE_PREFIX environment variable was also added. These environment variables are crucial for subsequent steps in the GitHub workflow. * Add workflow dispatch validation and response handling Removed several specific dispatch response classes and consolidated all types of dispatch responses into a single DispatchWorkflowResponse class. Added a new ValidatingWorkflowDispatcher service to validate dispatch requests before they're sent. Updated several classes to work with these changes, including the BackgroundWorkflowDispatcher, MassTransitWorkflowDispatcher, and the API endpoint class. * Handle dispatch workflow failures with exceptions The DispatchWorkflow and BulkDispatchWorkflows activities now throw a FaultException when the dispatch operations fail. Previously, these operations were not checking for success and could fail silently. Now, an unsuccessful dispatch response results in a FaultException with an error message from the dispatch response. --- .../Elsa.Server.Web.csproj.DotSettings | 2 + .../Enums/MassTransitBroker.cs | 10 ++ src/bundles/Elsa.Server.Web/Program.cs | 11 ++- src/bundles/Elsa.Server.Web/appsettings.json | 13 +++ .../DefaultAlteredWorkflowScheduler.cs | 2 +- .../Endpoints/Workflows/Retry/Endpoint.cs | 2 +- .../Handlers/ResumeWorkflowInstance.cs | 2 +- .../Elsa.Hangfire/Jobs/ResumeWorkflowJob.cs | 2 +- .../Elsa.Hangfire/Jobs/RunWorkflowJob.cs | 2 +- .../Features/AzureServiceBusFeature.cs | 14 ++- .../Features/RabbitMqServiceBusFeature.cs | 14 +-- .../Activities/MessageReceived.cs | 2 +- .../Activities/PublishMessage.cs | 2 +- .../DispatchWorkflowRequestConsumer.cs | 2 + .../Consumers/WorkflowMessageConsumer.cs | 2 +- .../Contracts/IEndpointChannelFormatter.cs | 14 +++ .../MassTransitFeatureExtensions.cs | 2 +- ...rviceBusChannelsConfigurationExtensions.cs | 42 ++++++++ .../Features/MassTransitFeature.cs | 22 +++-- .../MassTransitWorkflowDispatcherFeature.cs | 27 ++--- .../DefaultEndpointChannelFormatter.cs | 18 ++++ .../MassTransitWorkflowDispatcher.cs | 80 --------------- .../MassTransitActivityTypeProvider.cs | 4 +- .../Services/MassTransitWorkflowDispatcher.cs | 98 +++++++++++++++++++ .../Elsa.Quartz/Jobs/ResumeWorkflowJob.cs | 2 +- .../Elsa.Quartz/Jobs/RunWorkflowJob.cs | 2 +- .../Tasks/ResumeWorkflowTask.cs | 2 +- .../Elsa.Scheduling/Tasks/RunWorkflowTask.cs | 2 +- .../BulkDispatch/Endpoint.cs | 2 +- .../WorkflowDefinitions/Dispatch/Endpoint.cs | 52 +++++----- .../WorkflowDefinitions/Dispatch/Models.cs | 2 + .../Activities/BulkDispatchWorkflows.cs | 28 +++++- .../Activities/DispatchWorkflow.cs | 26 ++++- .../Contracts/IWorkflowDispatcher.cs | 10 +- .../Features/WorkflowRuntimeFeature.cs | 43 +++++++- .../Models/DispatchWorkflowOptions.cs | 6 ++ .../Models/DispatcherChannel.cs | 13 +++ .../Options/WorkflowDispatcherOptions.cs | 17 ++++ .../DispatchResumeWorkflowsResponse.cs | 3 - .../DispatchTriggerWorkflowsResponse.cs | 3 - .../DispatchWorkflowDefinitionRequest.cs | 2 +- .../DispatchWorkflowDefinitionResponse.cs | 3 - .../DispatchWorkflowInstanceResponse.cs | 3 - .../DispatchWorkflowDefinitionResponse.cs | 18 ++++ .../Services/BackgroundWorkflowDispatcher.cs | 18 ++-- .../DefaultBackgroundActivityInvoker.cs | 2 +- .../Services/DefaultWorkflowInbox.cs | 4 +- .../Services/TaskReporter.cs | 2 +- .../Services/ValidatingWorkflowDispatcher.cs | 63 ++++++++++++ .../DispatcherChannelOptionsProvider.cs | 31 ++++++ 50 files changed, 558 insertions(+), 190 deletions(-) create mode 100644 src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj.DotSettings create mode 100644 src/bundles/Elsa.Server.Web/Enums/MassTransitBroker.cs create mode 100644 src/modules/Elsa.MassTransit/Contracts/IEndpointChannelFormatter.cs create mode 100644 src/modules/Elsa.MassTransit/Extensions/ServiceBusChannelsConfigurationExtensions.cs create mode 100644 src/modules/Elsa.MassTransit/Formatters/DefaultEndpointChannelFormatter.cs delete mode 100644 src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs rename src/modules/Elsa.MassTransit/{Implementations => Services}/MassTransitActivityTypeProvider.cs (98%) create mode 100644 src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Models/DispatchWorkflowOptions.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Models/DispatcherChannel.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs delete mode 100644 src/modules/Elsa.Workflows.Runtime/Requests/DispatchResumeWorkflowsResponse.cs delete mode 100644 src/modules/Elsa.Workflows.Runtime/Requests/DispatchTriggerWorkflowsResponse.cs delete mode 100644 src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionResponse.cs delete mode 100644 src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceResponse.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Responses/DispatchWorkflowDefinitionResponse.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Services/ValidatingWorkflowDispatcher.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/UIHints/DispatcherChannelOptionsProvider.cs diff --git a/src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj.DotSettings b/src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj.DotSettings new file mode 100644 index 000000000..3d80b14ab --- /dev/null +++ b/src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj.DotSettings @@ -0,0 +1,2 @@ + + True \ No newline at end of file diff --git a/src/bundles/Elsa.Server.Web/Enums/MassTransitBroker.cs b/src/bundles/Elsa.Server.Web/Enums/MassTransitBroker.cs new file mode 100644 index 000000000..a93bc8f91 --- /dev/null +++ b/src/bundles/Elsa.Server.Web/Enums/MassTransitBroker.cs @@ -0,0 +1,10 @@ +namespace Elsa.Server.Web; + +/// +/// Represents the type of messaging broker used in MassTransit. +/// +public enum MassTransitBroker +{ + AzureServiceBus, + RabbitMq +} \ No newline at end of file diff --git a/src/bundles/Elsa.Server.Web/Program.cs b/src/bundles/Elsa.Server.Web/Program.cs index 78f1468e5..3078e1fc5 100644 --- a/src/bundles/Elsa.Server.Web/Program.cs +++ b/src/bundles/Elsa.Server.Web/Program.cs @@ -15,6 +15,7 @@ using Elsa.MongoDb.Extensions; using Elsa.MongoDb.Modules.Identity; using Elsa.MongoDb.Modules.Management; using Elsa.MongoDb.Modules.Runtime; +using Elsa.Server.Web; using Elsa.Workflows.Management.Compression; using Microsoft.Data.Sqlite; using Microsoft.Extensions.Options; @@ -27,10 +28,9 @@ const bool useDapper = false; const bool useProtoActor = false; const bool useHangfire = false; const bool useQuartz = true; -const bool useMassTransit = false; -const bool useMassTransitAzureServiceBus = true; -const bool useMassTransitRabbitMq = false; +const bool useMassTransit = true; const bool useZipCompression = true; +const MassTransitBroker useMassTransitBroker = MassTransitBroker.AzureServiceBus; var builder = WebApplication.CreateBuilder(args); var services = builder.Services; @@ -142,6 +142,7 @@ services runtime.UseMassTransitDispatcher(); } runtime.WorkflowInboxCleanupOptions = options => configuration.GetSection("Runtime:WorkflowInboxCleanup").Bind(options); + runtime.WorkflowDispatcherOptions = options => configuration.GetSection("Runtime:WorkflowDispatcher").Bind(options); }) .UseEnvironments(environments => environments.EnvironmentsOptions = options => configuration.GetSection("Environments").Bind(options)) .UseScheduling(scheduling => @@ -210,7 +211,7 @@ services { elsa.UseMassTransit(massTransit => { - if (useMassTransitAzureServiceBus) + if (useMassTransitBroker == MassTransitBroker.AzureServiceBus) { massTransit.UseAzureServiceBus(azureServiceBusConnectionString, serviceBusFeature => serviceBusFeature.ConfigureServiceBus = bus => { @@ -222,7 +223,7 @@ services }); } - if (useMassTransitRabbitMq) + if (useMassTransitBroker == MassTransitBroker.RabbitMq) { massTransit.UseRabbitMq(rabbitMqConnectionString, rabbit => rabbit.ConfigureServiceBus = bus => { diff --git a/src/bundles/Elsa.Server.Web/appsettings.json b/src/bundles/Elsa.Server.Web/appsettings.json index 131ccbe59..396e4e0bd 100644 --- a/src/bundles/Elsa.Server.Web/appsettings.json +++ b/src/bundles/Elsa.Server.Web/appsettings.json @@ -97,6 +97,19 @@ "WorkflowInboxCleanup": { "SweepInterval": "00:00:10:00", "BatchSize": 1000 + }, + "WorkflowDispatcher": { + "Channels": [ + { + "Name": "Low" + }, + { + "Name": "Medium" + }, + { + "Name": "High" + } + ] } }, "Scripting": { diff --git a/src/modules/Elsa.Alterations.Core/Contracts/DefaultAlteredWorkflowScheduler.cs b/src/modules/Elsa.Alterations.Core/Contracts/DefaultAlteredWorkflowScheduler.cs index 0fa9e1706..f2b6be4d0 100644 --- a/src/modules/Elsa.Alterations.Core/Contracts/DefaultAlteredWorkflowScheduler.cs +++ b/src/modules/Elsa.Alterations.Core/Contracts/DefaultAlteredWorkflowScheduler.cs @@ -27,6 +27,6 @@ public class DefaultAlteredWorkflowDispatcher : IAlteredWorkflowDispatcher /// public async Task DispatchAsync(RunAlterationsResult result, CancellationToken cancellationToken = default) { - await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(result.WorkflowInstanceId), cancellationToken); + await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(result.WorkflowInstanceId), cancellationToken: cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Endpoints/Workflows/Retry/Endpoint.cs b/src/modules/Elsa.Alterations/Endpoints/Workflows/Retry/Endpoint.cs index b2a8e97bf..fa2c33cd6 100644 --- a/src/modules/Elsa.Alterations/Endpoints/Workflows/Retry/Endpoint.cs +++ b/src/modules/Elsa.Alterations/Endpoints/Workflows/Retry/Endpoint.cs @@ -56,7 +56,7 @@ public class Retry : ElsaEndpoint allResults.AddRange(results); // Schedule updated workflow. - await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(workflowInstance.Id), cancellationToken); + await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(workflowInstance.Id), cancellationToken: cancellationToken); } // Write response. diff --git a/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs b/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs index 22eba245a..da4604a12 100644 --- a/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs +++ b/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs @@ -37,6 +37,6 @@ public class ResumeWorkflowInstance : INotificationHandlerThe name of the job. /// The workflow request. /// The cancellation token. - public async Task ExecuteAsync(string name, DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken) => await _workflowDispatcher.DispatchAsync(request, cancellationToken); + public async Task ExecuteAsync(string name, DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken) => await _workflowDispatcher.DispatchAsync(request, cancellationToken: cancellationToken); } \ No newline at end of file diff --git a/src/modules/Elsa.Hangfire/Jobs/RunWorkflowJob.cs b/src/modules/Elsa.Hangfire/Jobs/RunWorkflowJob.cs index 0d656ec96..3d1fb829f 100644 --- a/src/modules/Elsa.Hangfire/Jobs/RunWorkflowJob.cs +++ b/src/modules/Elsa.Hangfire/Jobs/RunWorkflowJob.cs @@ -24,5 +24,5 @@ public class RunWorkflowJob /// The name of the job. /// The workflow request. /// The cancellation token. - public async Task ExecuteAsync(string name, DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken) => await _workflowDispatcher.DispatchAsync(request, cancellationToken); + public async Task ExecuteAsync(string name, DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken) => await _workflowDispatcher.DispatchAsync(request, cancellationToken: cancellationToken); } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs b/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs index 5050a7c16..81fed0955 100644 --- a/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs +++ b/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs @@ -1,7 +1,10 @@ using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; +using Elsa.MassTransit.Extensions; using Elsa.MassTransit.Features; +using Elsa.MassTransit.Messages; +using Elsa.Workflows.Runtime.Activities; using MassTransit; namespace Elsa.MassTransit.AzureServiceBus.Features; @@ -33,14 +36,15 @@ public class AzureServiceBusFeature : FeatureBase { configure.AddServiceBusMessageScheduler(); - configure.UsingAzureServiceBus((context, serviceBus) => + configure.UsingAzureServiceBus((context, configurator) => { if (ConnectionString != null) - serviceBus.Host(ConnectionString); + configurator.Host(ConnectionString); - serviceBus.UseServiceBusMessageScheduler(); - ConfigureServiceBus?.Invoke(serviceBus); - serviceBus.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); + configurator.UseServiceBusMessageScheduler(); + configurator.SetupWorkflowDispatcherEndpoints(context); + ConfigureServiceBus?.Invoke(configurator); + configurator.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); }); }; }); diff --git a/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs b/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs index af1271514..53283432e 100644 --- a/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs +++ b/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs @@ -1,6 +1,8 @@ using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; +using Elsa.MassTransit.Consumers; +using Elsa.MassTransit.Extensions; using Elsa.MassTransit.Features; using MassTransit; using Microsoft.Extensions.DependencyInjection; @@ -21,9 +23,8 @@ public class RabbitMqServiceBusFeature : FeatureBase /// A RabbitMQ connection string. public string? ConnectionString { get; set; } - /// Configures the RabbitMQ transport options. public Action? TransportOptions { get; set; } - + /// /// Configures the RabbitMQ bus. /// @@ -34,14 +35,15 @@ public class RabbitMqServiceBusFeature : FeatureBase { Module.Configure().BusConfigurator = configure => { - configure.UsingRabbitMq((context, serviceBus) => + configure.UsingRabbitMq((context, configurator) => { if (!string.IsNullOrEmpty(ConnectionString)) - serviceBus.Host(ConnectionString); + configurator.Host(ConnectionString); - ConfigureServiceBus?.Invoke(serviceBus); + ConfigureServiceBus?.Invoke(configurator); - serviceBus.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); + configurator.SetupWorkflowDispatcherEndpoints(context); + configurator.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); }); }; } diff --git a/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs b/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs index de1121995..d0a32890a 100644 --- a/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs +++ b/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs @@ -2,7 +2,7 @@ using System.ComponentModel; using System.Runtime.CompilerServices; using Elsa.Expressions.Models; using Elsa.Extensions; -using Elsa.MassTransit.Implementations; +using Elsa.MassTransit.Services; using Elsa.Workflows; namespace Elsa.MassTransit.Activities; diff --git a/src/modules/Elsa.MassTransit/Activities/PublishMessage.cs b/src/modules/Elsa.MassTransit/Activities/PublishMessage.cs index 1802f3bd8..2e1770159 100644 --- a/src/modules/Elsa.MassTransit/Activities/PublishMessage.cs +++ b/src/modules/Elsa.MassTransit/Activities/PublishMessage.cs @@ -2,7 +2,7 @@ using System.ComponentModel; using System.Runtime.CompilerServices; using Elsa.Expressions.Helpers; using Elsa.Extensions; -using Elsa.MassTransit.Implementations; +using Elsa.MassTransit.Services; using Elsa.Workflows; using Elsa.Workflows.Attributes; using Elsa.Workflows.Models; diff --git a/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs b/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs index fcfcf255a..4912ba6a6 100644 --- a/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs +++ b/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs @@ -1,6 +1,7 @@ using Elsa.MassTransit.Messages; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Options; +using JetBrains.Annotations; using MassTransit; namespace Elsa.MassTransit.Consumers; @@ -8,6 +9,7 @@ namespace Elsa.MassTransit.Consumers; /// /// A consumer of various dispatch message types to asynchronously execute workflows. /// +[UsedImplicitly] public class DispatchWorkflowRequestConsumer : IConsumer, IConsumer, diff --git a/src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs b/src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs index 17ca203bf..e8bc5afbb 100644 --- a/src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs +++ b/src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs @@ -36,6 +36,6 @@ public class WorkflowMessageConsumer : IConsumer where T : class CorrelationId = correlationId, Input = input }; - await _workflowRuntime.DispatchAsync(request, cancellationToken); + await _workflowRuntime.DispatchAsync(request, cancellationToken: cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Contracts/IEndpointChannelFormatter.cs b/src/modules/Elsa.MassTransit/Contracts/IEndpointChannelFormatter.cs new file mode 100644 index 000000000..1a2612397 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Contracts/IEndpointChannelFormatter.cs @@ -0,0 +1,14 @@ +namespace Elsa.MassTransit.Contracts; + +/// +/// Represents a contract for formatting channel queue names. +/// +public interface IEndpointChannelFormatter +{ + /// + /// Formats the queue name based on the provided channel name. + /// + /// The channel name to be formatted. + /// The formatted queue name. + string FormatEndpointName(string? channelName = default); +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs b/src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs index 80e03d6b1..ee05200e9 100644 --- a/src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs +++ b/src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs @@ -1,7 +1,7 @@ using Elsa.Features.Services; using Elsa.MassTransit.Features; -using Elsa.MassTransit.Implementations; using Elsa.MassTransit.Models; +using Elsa.MassTransit.Services; using MassTransit; // ReSharper disable once CheckNamespace diff --git a/src/modules/Elsa.MassTransit/Extensions/ServiceBusChannelsConfigurationExtensions.cs b/src/modules/Elsa.MassTransit/Extensions/ServiceBusChannelsConfigurationExtensions.cs new file mode 100644 index 000000000..45622b6d5 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Extensions/ServiceBusChannelsConfigurationExtensions.cs @@ -0,0 +1,42 @@ +using Elsa.MassTransit.Consumers; +using Elsa.MassTransit.Contracts; +using Elsa.Workflows.Runtime.Options; +using MassTransit; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Elsa.MassTransit.Extensions; + +/// +/// Provides extension methods for configuring service bus channel endpoints. +/// +public static class ServiceBusChannelsConfigurationExtensions +{ + /// + /// Sets up channel endpoints for receiving workflow dispatch requests. + /// + /// The MassTransit receive configurator. + /// The MassTransit bus registration context. + public static void SetupWorkflowDispatcherEndpoints(this IReceiveConfigurator configurator, IBusRegistrationContext context) + { + var dispatcherOptions = context.GetRequiredService>(); + var formatter = context.GetRequiredService(); + var channelDescriptors = dispatcherOptions.Value.Channels; + var defaultEndpointName = formatter.FormatEndpointName(); + + var endpointNames = new List + { + defaultEndpointName + }; + + endpointNames.AddRange(channelDescriptors.Select(x => formatter.FormatEndpointName(x.Name))); + + foreach (string endpointName in endpointNames) + { + configurator.ReceiveEndpoint(endpointName, endpoint => + { + endpoint.ConfigureConsumer(context); + }); + } + } +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs b/src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs index d30250b18..527a9d5bc 100644 --- a/src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs +++ b/src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs @@ -5,9 +5,10 @@ using Elsa.Extensions; using Elsa.Features.Abstractions; using Elsa.Features.Services; using Elsa.MassTransit.Consumers; -using Elsa.MassTransit.Implementations; +using Elsa.MassTransit.Extensions; using Elsa.MassTransit.Models; using Elsa.MassTransit.Options; +using Elsa.MassTransit.Services; using Elsa.Workflows.Attributes; using Elsa.Workflows.Management.Models; using Elsa.Workflows.Management.Options; @@ -49,20 +50,23 @@ public class MassTransitFeature : FeatureBase Services.Configure(x => { }); Services.AddActivityProvider(); - var busConfigurator = BusConfigurator ??= configure => + void Configurator(IBusRegistrationConfigurator configure) { configure.UsingInMemory((context, configurator) => { + configurator.SetupWorkflowDispatcherEndpoints(context); configurator.ConfigureEndpoints(context); - if (PrefetchCount != null) - configurator.PrefetchCount = PrefetchCount.Value; + if (PrefetchCount != null) configurator.PrefetchCount = PrefetchCount.Value; }); - }; + } + + var busConfigurator = BusConfigurator ??= Configurator; AddMassTransit(busConfigurator); // Add collected message types to options. - Services.Configure(options => options.MessageTypes = new HashSet(messageTypes)); + Services.Configure( + options => options.MessageTypes = new HashSet(messageTypes)); // Add collected message types as available variable types. Services.Configure(options => @@ -79,7 +83,8 @@ public class MassTransitFeature : FeatureBase }); // Configure message serializer. - SystemTextJsonMessageSerializer.Options.Converters.Add(new TypeJsonConverter(WellKnownTypeRegistry.CreateDefault())); + SystemTextJsonMessageSerializer.Options.Converters.Add( + new TypeJsonConverter(WellKnownTypeRegistry.CreateDefault())); } /// @@ -89,7 +94,8 @@ public class MassTransitFeature : FeatureBase { // For each message type, create a concrete WorkflowMessageConsumer. var workflowMessageConsumerType = typeof(WorkflowMessageConsumer<>); - var workflowMessageConsumers = this.GetMessages().Select(x => new ConsumerTypeDefinition(workflowMessageConsumerType.MakeGenericType(x))); + var workflowMessageConsumers = this.GetMessages() + .Select(x => new ConsumerTypeDefinition(workflowMessageConsumerType.MakeGenericType(x))); // Concatenate the manually registered consumers with the workflow message consumers. var consumerTypeDefinitions = this.GetConsumers().Concat(workflowMessageConsumers).ToArray(); diff --git a/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs b/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs index af2f2f4c6..30cba3ded 100644 --- a/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs +++ b/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs @@ -4,12 +4,13 @@ using Elsa.Features.Attributes; using Elsa.Features.Services; using Elsa.MassTransit.ConsumerDefinitions; using Elsa.MassTransit.Consumers; -using Elsa.MassTransit.Implementations; -using Elsa.MassTransit.Messages; +using Elsa.MassTransit.Contracts; +using Elsa.MassTransit.Formatters; using Elsa.MassTransit.Options; +using Elsa.MassTransit.Services; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Features; -using MassTransit; +using Elsa.Workflows.Runtime.Services; using Microsoft.Extensions.DependencyInjection; namespace Elsa.MassTransit.Features; @@ -31,11 +32,20 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase /// public Action? ConfigureDispatcherOptions { get; set; } + /// + /// A factory that creates a . + /// + public Func ChannelQueueFormatterFactory { get; set; } = _ => new DefaultEndpointChannelFormatter(); + /// public override void Configure() { Module.AddMassTransitConsumer(); - Module.Configure(f => f.WorkflowDispatcher = sp => sp.GetRequiredService()); + Module.Configure(f => f.WorkflowDispatcher = sp => + { + var decoratedService = ActivatorUtilities.CreateInstance(sp); + return ActivatorUtilities.CreateInstance(sp, decoratedService); + }); } /// @@ -46,13 +56,6 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase if (ConfigureDispatcherOptions != null) options.Configure(ConfigureDispatcherOptions); - var queueName = KebabCaseEndpointNameFormatter.Instance.Consumer(); - var queueAddress = new Uri($"queue:elsa-{queueName}"); - EndpointConvention.Map(queueAddress); - EndpointConvention.Map(queueAddress); - EndpointConvention.Map(queueAddress); - EndpointConvention.Map(queueAddress); - - Services.AddScoped(); + Services.AddSingleton(ChannelQueueFormatterFactory); } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Formatters/DefaultEndpointChannelFormatter.cs b/src/modules/Elsa.MassTransit/Formatters/DefaultEndpointChannelFormatter.cs new file mode 100644 index 000000000..beca6be16 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Formatters/DefaultEndpointChannelFormatter.cs @@ -0,0 +1,18 @@ +using Elsa.MassTransit.Contracts; +using Humanizer; + +namespace Elsa.MassTransit.Formatters; + +/// +/// The DefaultChannelQueueFormatter class is an implementation of the IChannelQueueFormatter interface. +/// It is responsible for formatting channel queue names. +/// +public class DefaultEndpointChannelFormatter : IEndpointChannelFormatter +{ + /// + public string FormatEndpointName(string? channelName) + { + var channelSuffix = !string.IsNullOrEmpty(channelName) ? "-" + channelName.Kebaberize() : ""; + return $"elsa-dispatch-workflow-request{channelSuffix}"; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs b/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs deleted file mode 100644 index c0ccee005..000000000 --- a/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs +++ /dev/null @@ -1,80 +0,0 @@ -using Elsa.MassTransit.Messages; -using Elsa.Workflows.Runtime.Contracts; -using Elsa.Workflows.Runtime.Requests; -using MassTransit; - -namespace Elsa.MassTransit.Implementations; - -/// -/// Implements by leveraging MassTransit. -/// -public class MassTransitWorkflowDispatcher : IWorkflowDispatcher -{ - private readonly IBus _bus; - - /// - /// Constructor. - /// - public MassTransitWorkflowDispatcher(IBus bus) - { - _bus = bus; - } - - /// - public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) - { - await _bus.Publish(new DispatchWorkflowDefinition( - request.DefinitionId, - request.VersionOptions, - request.Input, - request.Properties, - request.CorrelationId, - request.InstanceId, - request.TriggerActivityId - ), cancellationToken); - return new(); - } - - /// - public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default) - { - await _bus.Publish(new DispatchWorkflowInstance(request.InstanceId) - { - BookmarkId = request.BookmarkId, - ActivityId = request.ActivityId, - ActivityNodeId = request.ActivityNodeId, - ActivityInstanceId = request.ActivityInstanceId, - ActivityHash = request.ActivityHash, - Input = request.Input, - Properties = request.Properties, - CorrelationId = request.CorrelationId - }, cancellationToken); - return new(); - } - - /// - public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, CancellationToken cancellationToken = default) - { - await _bus.Publish(new DispatchTriggerWorkflows(request.ActivityTypeName, request.BookmarkPayload) - { - CorrelationId = request.CorrelationId, - WorkflowInstanceId = request.WorkflowInstanceId, - ActivityInstanceId = request.ActivityInstanceId, - Input = request.Input - }, cancellationToken); - return new(); - } - - /// - public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, CancellationToken cancellationToken = default) - { - await _bus.Publish(new DispatchResumeWorkflows(request.ActivityTypeName, request.BookmarkPayload) - { - CorrelationId = request.CorrelationId, - WorkflowInstanceId = request.WorkflowInstanceId, - ActivityInstanceId = request.ActivityInstanceId, - Input = request.Input - }, cancellationToken); - return new(); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Implementations/MassTransitActivityTypeProvider.cs b/src/modules/Elsa.MassTransit/Services/MassTransitActivityTypeProvider.cs similarity index 98% rename from src/modules/Elsa.MassTransit/Implementations/MassTransitActivityTypeProvider.cs rename to src/modules/Elsa.MassTransit/Services/MassTransitActivityTypeProvider.cs index e2dabe143..9ecfff2c7 100644 --- a/src/modules/Elsa.MassTransit/Implementations/MassTransitActivityTypeProvider.cs +++ b/src/modules/Elsa.MassTransit/Services/MassTransitActivityTypeProvider.cs @@ -10,13 +10,15 @@ using Elsa.Workflows.Helpers; using Elsa.Workflows.Management; using Elsa.Workflows.Models; using Humanizer; +using JetBrains.Annotations; using Microsoft.Extensions.Options; -namespace Elsa.MassTransit.Implementations; +namespace Elsa.MassTransit.Services; /// /// Provides activities to the system from the configured MassTransit message types. /// +[UsedImplicitly] public class MassTransitActivityTypeProvider(IActivityFactory activityFactory, IOptions options, IActivityDescriber activityDescriber) : IActivityProvider { /// diff --git a/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs new file mode 100644 index 000000000..37f17a8a2 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs @@ -0,0 +1,98 @@ +using Elsa.MassTransit.Contracts; +using Elsa.MassTransit.Messages; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; +using MassTransit; + +namespace Elsa.MassTransit.Services; + +/// +/// Implements by leveraging MassTransit. +/// +public class MassTransitWorkflowDispatcher : IWorkflowDispatcher +{ + private readonly IBus _bus; + private readonly IEndpointChannelFormatter _endpointChannelFormatter; + + /// + /// Initializes a new instance of the class. + /// + public MassTransitWorkflowDispatcher(IBus bus, IEndpointChannelFormatter endpointChannelFormatter) + { + _bus = bus; + _endpointChannelFormatter = endpointChannelFormatter; + } + + /// + public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + var sendEndpoint = await GetSendEndpointAsync(options); + + await sendEndpoint.Send(new DispatchWorkflowDefinition( + request.DefinitionId, + request.VersionOptions, + request.Input, + request.Properties, + request.CorrelationId, + request.InstanceId, + request.TriggerActivityId + ), cancellationToken); + return DispatchWorkflowResponse.Success(); + } + + /// + public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + var sendEndpoint = await GetSendEndpointAsync(options); + + await sendEndpoint.Send(new DispatchWorkflowInstance(request.InstanceId) + { + BookmarkId = request.BookmarkId, + ActivityId = request.ActivityId, + ActivityNodeId = request.ActivityNodeId, + ActivityInstanceId = request.ActivityInstanceId, + ActivityHash = request.ActivityHash, + Input = request.Input, + Properties = request.Properties, + CorrelationId = request.CorrelationId + }, cancellationToken); + return DispatchWorkflowResponse.Success(); + } + + /// + public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + var sendEndpoint = await GetSendEndpointAsync(options); + await sendEndpoint.Send(new DispatchTriggerWorkflows(request.ActivityTypeName, request.BookmarkPayload) + { + CorrelationId = request.CorrelationId, + WorkflowInstanceId = request.WorkflowInstanceId, + ActivityInstanceId = request.ActivityInstanceId, + Input = request.Input + }, cancellationToken); + return DispatchWorkflowResponse.Success(); + } + + /// + public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + var sendEndpoint = await GetSendEndpointAsync(options); + await sendEndpoint.Send(new DispatchResumeWorkflows(request.ActivityTypeName, request.BookmarkPayload) + { + CorrelationId = request.CorrelationId, + WorkflowInstanceId = request.WorkflowInstanceId, + ActivityInstanceId = request.ActivityInstanceId, + Input = request.Input + }, cancellationToken); + return DispatchWorkflowResponse.Success(); + } + + private async Task GetSendEndpointAsync(DispatchWorkflowOptions? options = default) + { + var endpointName = _endpointChannelFormatter.FormatEndpointName(options?.Channel); + var sendEndpoint = await _bus.GetSendEndpoint(new Uri($"queue:{endpointName}")); + return sendEndpoint; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Quartz/Jobs/ResumeWorkflowJob.cs b/src/modules/Elsa.Quartz/Jobs/ResumeWorkflowJob.cs index 6a51977d9..f4cd46fd9 100644 --- a/src/modules/Elsa.Quartz/Jobs/ResumeWorkflowJob.cs +++ b/src/modules/Elsa.Quartz/Jobs/ResumeWorkflowJob.cs @@ -39,6 +39,6 @@ public class ResumeWorkflowJob : IJob CorrelationId = (string?)map.Get(nameof(DispatchWorkflowInstanceRequest.CorrelationId)), Input = map.GetDictionary(nameof(DispatchWorkflowInstanceRequest.Input)) }; - await _workflowDispatcher.DispatchAsync(request, context.CancellationToken); + await _workflowDispatcher.DispatchAsync(request, cancellationToken: context.CancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Quartz/Jobs/RunWorkflowJob.cs b/src/modules/Elsa.Quartz/Jobs/RunWorkflowJob.cs index c81c2181c..fd4795f71 100644 --- a/src/modules/Elsa.Quartz/Jobs/RunWorkflowJob.cs +++ b/src/modules/Elsa.Quartz/Jobs/RunWorkflowJob.cs @@ -44,6 +44,6 @@ public class RunWorkflowJob : IJob CorrelationId = (string?)map.Get(nameof(DispatchWorkflowDefinitionRequest.CorrelationId)), Input = map.GetDictionary(nameof(DispatchWorkflowDefinitionRequest.Input)) }; - await _workflowDispatcher.DispatchAsync(request, context.CancellationToken); + await _workflowDispatcher.DispatchAsync(request, cancellationToken: context.CancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs b/src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs index ab3448f1f..a9298f9f1 100644 --- a/src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs +++ b/src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs @@ -26,6 +26,6 @@ public class ResumeWorkflowTask : ITask public async ValueTask ExecuteAsync(TaskExecutionContext context) { var workflowDispatcher = context.ServiceProvider.GetRequiredService(); - await workflowDispatcher.DispatchAsync(_dispatchWorkflowInstanceRequest, context.CancellationToken); + await workflowDispatcher.DispatchAsync(_dispatchWorkflowInstanceRequest, cancellationToken: context.CancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Scheduling/Tasks/RunWorkflowTask.cs b/src/modules/Elsa.Scheduling/Tasks/RunWorkflowTask.cs index fdf100098..15cb29916 100644 --- a/src/modules/Elsa.Scheduling/Tasks/RunWorkflowTask.cs +++ b/src/modules/Elsa.Scheduling/Tasks/RunWorkflowTask.cs @@ -25,6 +25,6 @@ public class RunWorkflowTask : ITask public async ValueTask ExecuteAsync(TaskExecutionContext context) { var workflowDispatcher = context.ServiceProvider.GetRequiredService(); - await workflowDispatcher.DispatchAsync(_request, context.CancellationToken); + await workflowDispatcher.DispatchAsync(_request, cancellationToken: context.CancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDispatch/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDispatch/Endpoint.cs index 903c88e2c..c269a6d33 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDispatch/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDispatch/Endpoint.cs @@ -64,7 +64,7 @@ internal class Endpoint : ElsaEndpoint TriggerActivityId = triggerActivityId }; - await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken); + await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken: cancellationToken); instanceIds.Add(instanceId); } diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Endpoint.cs index 3a83d8e18..bf33ed80c 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Endpoint.cs @@ -4,25 +4,15 @@ using Elsa.Workflows.Contracts; using Elsa.Workflows.Management.Contracts; using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; using JetBrains.Annotations; namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Dispatch; -[PublicAPI] -internal class Endpoint : ElsaEndpoint +[UsedImplicitly] +internal class Endpoint(IWorkflowDefinitionStore store, IWorkflowDispatcher workflowDispatcher, IIdentityGenerator identityGenerator) : ElsaEndpoint { - private readonly IWorkflowDefinitionStore _store; - private readonly IWorkflowDispatcher _workflowDispatcher; - private readonly IIdentityGenerator _identityGenerator; - - public Endpoint(IWorkflowDefinitionStore store, IWorkflowDispatcher workflowDispatcher, IIdentityGenerator identityGenerator) - { - _store = store; - _workflowDispatcher = workflowDispatcher; - _identityGenerator = identityGenerator; - } - public override void Configure() { Post("/workflow-definitions/{definitionId}/dispatch"); @@ -34,7 +24,7 @@ internal class Endpoint : ElsaEndpoint var definitionId = request.DefinitionId; var versionOptions = request.VersionOptions ?? VersionOptions.Published; - var exists = await _store.AnyAsync( + var exists = await store.AnyAsync( new WorkflowDefinitionFilter { DefinitionId = definitionId, @@ -47,23 +37,41 @@ internal class Endpoint : ElsaEndpoint await SendNotFoundAsync(cancellationToken); return; } + + var input = request.Input; + + if(input != null && input is not IDictionary) + { + AddError("Input must be a dictionary."); + await SendErrorsAsync(cancellation: cancellationToken); + return; + } - var instanceId = request.InstanceId ?? _identityGenerator.GenerateId(); - var correlationId = request.CorrelationId; - var triggerActivityId = request.TriggerActivityId; - var input = (IDictionary?)request.Input; + var instanceId = request.InstanceId ?? identityGenerator.GenerateId(); var dispatchRequest = new DispatchWorkflowDefinitionRequest { DefinitionId = definitionId, VersionOptions = versionOptions, - Input = input, + Input = input as IDictionary, InstanceId = instanceId, - CorrelationId = correlationId, - TriggerActivityId = triggerActivityId + CorrelationId = request.CorrelationId, + TriggerActivityId = request.TriggerActivityId, }; - await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken); + var options = new DispatchWorkflowOptions + { + Channel = request.Channel + }; + var result = await workflowDispatcher.DispatchAsync(dispatchRequest, options, cancellationToken); + + if(!result.Succeeded) + { + AddError(result.ErrorMessage!); + await SendErrorsAsync(cancellation: cancellationToken); + return; + } + await SendOkAsync(new Response(instanceId), cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Models.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Models.cs index 2acb4f36c..0a10a7a21 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Models.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Models.cs @@ -15,6 +15,8 @@ internal class Request [JsonConverter(typeof(ExpandoObjectConverterFactory))] public object? Input { get; set; } + + public string? Channel { get; set; } } internal record Response(string WorkflowInstanceId); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs index 2a5cbe656..f45cc34f2 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs @@ -7,13 +7,16 @@ using Elsa.Extensions; using Elsa.Workflows.Activities.Flowchart.Attributes; using Elsa.Workflows.Attributes; using Elsa.Workflows.Contracts; +using Elsa.Workflows.Exceptions; using Elsa.Workflows.UIHints; using Elsa.Workflows.Memory; using Elsa.Workflows.Models; using Elsa.Workflows.Options; using Elsa.Workflows.Runtime.Bookmarks; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.UIHints; using Elsa.Workflows.Services; using JetBrains.Annotations; @@ -73,6 +76,17 @@ public class BulkDispatchWorkflows : Activity DefaultValue = true)] public Input WaitForCompletion { get; set; } = new(true); + /// + /// The channel to dispatch the workflow to. + /// + [Input( + DisplayName = "Channel", + Description = "The channel to dispatch the workflow to.", + UIHint = InputUIHints.DropDown, + UIHandler = typeof(DispatcherChannelOptionsProvider) + )] + public Input ChannelName { get; set; } = default!; + /// /// An activity to execute when the child workflow finishes. /// @@ -97,7 +111,7 @@ public class BulkDispatchWorkflows : Activity await ProcessItem(context, item); dispatchedInstancesCount++; } - + context.SetProperty(DispatchedInstancesCountKey, dispatchedInstancesCount); // If we need to wait for the child workflows to complete (if any), create a bookmark. @@ -144,6 +158,7 @@ public class BulkDispatchWorkflows : Activity var workflowDefinitionId = WorkflowDefinitionId.Get(context); var parentInstanceId = context.WorkflowExecutionContext.Id; var input = Input.GetOrDefault(context) ?? new Dictionary(); + var channelName = ChannelName.GetOrDefault(context); var properties = new Dictionary { ["ParentInstanceId"] = parentInstanceId @@ -173,12 +188,19 @@ public class BulkDispatchWorkflows : Activity CorrelationId = correlationId, InstanceId = instanceId }; + var options = new DispatchWorkflowOptions + { + Channel = channelName + }; - await workflowDispatcher.DispatchAsync(request, context.CancellationToken); + var dispatchResponse = await workflowDispatcher.DispatchAsync(request, options, context.CancellationToken); + + if (!dispatchResponse.Succeeded) + throw new FaultException(dispatchResponse.ErrorMessage); return instanceId; } - + private async ValueTask OnChildWorkflowCompletedAsync(ActivityExecutionContext context) { var input = context.WorkflowInput; diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs index dfba4ac65..eff1280c3 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs @@ -3,11 +3,14 @@ using Elsa.Common.Models; using Elsa.Extensions; using Elsa.Workflows.Attributes; using Elsa.Workflows.Contracts; +using Elsa.Workflows.Exceptions; using Elsa.Workflows.UIHints; using Elsa.Workflows.Models; using Elsa.Workflows.Runtime.Bookmarks; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.UIHints; using JetBrains.Annotations; namespace Elsa.Workflows.Runtime.Activities; @@ -55,6 +58,17 @@ public class DispatchWorkflow : Activity [Input(Description = "Wait for the child workflow to complete before completing this activity.")] public Input WaitForCompletion { get; set; } = default!; + /// + /// The channel to dispatch the workflow to. + /// + [Input( + DisplayName = "Channel", + Description = "The channel to dispatch the workflow to.", + UIHint = InputUIHints.DropDown, + UIHandler = typeof(DispatcherChannelOptionsProvider) + )] + public Input ChannelName { get; set; } = default!; + /// protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { @@ -85,6 +99,7 @@ public class DispatchWorkflow : Activity { var workflowDefinitionId = WorkflowDefinitionId.Get(context); var input = Input.GetOrDefault(context) ?? new Dictionary(); + var channelName = ChannelName.GetOrDefault(context); input["ParentInstanceId"] = context.WorkflowExecutionContext.Id; @@ -98,11 +113,18 @@ public class DispatchWorkflow : Activity VersionOptions = VersionOptions.Published, Input = input, CorrelationId = correlationId, - InstanceId = instanceId + InstanceId = instanceId, + }; + var options = new DispatchWorkflowOptions + { + Channel = channelName }; // Dispatch the child workflow. - await workflowDispatcher.DispatchAsync(request, context.CancellationToken); + var dispatchResponse = await workflowDispatcher.DispatchAsync(request, options, context.CancellationToken); + + if (!dispatchResponse.Succeeded) + throw new FaultException(dispatchResponse.ErrorMessage); return instanceId; } diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDispatcher.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDispatcher.cs index ca20040af..744fccf7f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDispatcher.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDispatcher.cs @@ -1,4 +1,6 @@ +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; namespace Elsa.Workflows.Runtime.Contracts; @@ -10,20 +12,20 @@ public interface IWorkflowDispatcher /// /// Dispatches a request to execute the specified workflow definition. /// - Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default); + Task DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default); /// /// Dispatches a request to execute the specified workflow instance. /// - Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default); + Task DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default); /// /// Starts all workflows and resumes existing workflow instances based on the specified activity type and bookmark payload. /// - Task DispatchAsync(DispatchTriggerWorkflowsRequest request, CancellationToken cancellationToken = default); + Task DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default); /// /// Resumes the workflow waiting for the specified bookmark. /// - Task DispatchAsync(DispatchResumeWorkflowsRequest request, CancellationToken cancellationToken = default); + Task DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index 23919dcde..744b73be2 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -12,6 +12,7 @@ using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Entities; using Elsa.Workflows.Runtime.Handlers; using Elsa.Workflows.Runtime.HostedServices; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Providers; using Elsa.Workflows.Runtime.Services; @@ -33,6 +34,8 @@ public class WorkflowRuntimeFeature : FeatureBase { } + private IDictionary WorkflowDispatcherChannels { get; set; } = new Dictionary(); + /// /// A list of workflow builders configured during application startup. /// @@ -46,7 +49,11 @@ public class WorkflowRuntimeFeature : FeatureBase /// /// A factory that instantiates an . /// - public Func WorkflowDispatcher { get; set; } = sp => ActivatorUtilities.CreateInstance(sp); + public Func WorkflowDispatcher { get; set; } = sp => + { + var decoratedService = ActivatorUtilities.CreateInstance(sp); + return ActivatorUtilities.CreateInstance(sp, decoratedService); + }; /// /// A factory that instantiates an . @@ -102,6 +109,11 @@ public class WorkflowRuntimeFeature : FeatureBase /// A delegate to configure the . /// public Action WorkflowInboxCleanupOptions { get; set; } = _ => { }; + + /// + /// A delegate to configure the . + /// + public Action WorkflowDispatcherOptions { get; set; } = _ => { }; /// /// Register the specified workflow type. @@ -127,6 +139,26 @@ public class WorkflowRuntimeFeature : FeatureBase return this; } + /// + /// Adds a dispatcher channel. + /// + public WorkflowRuntimeFeature AddDispatcherChannel(string channel) + { + return AddDispatcherChannel(new DispatcherChannel + { + Name = channel + }); + } + + /// + /// Adds a dispatcher channel. + /// + public WorkflowRuntimeFeature AddDispatcherChannel(DispatcherChannel channel) + { + WorkflowDispatcherChannels[channel.Name] = channel; + return this; + } + /// public override void Configure() { @@ -146,8 +178,13 @@ public class WorkflowRuntimeFeature : FeatureBase { // Options. Services.Configure(DistributedLockingOptions); - Services.Configure(options => { options.Workflows = Workflows; }); Services.Configure(WorkflowInboxCleanupOptions); + Services.Configure(WorkflowDispatcherOptions); + Services.Configure(options => { options.Workflows = Workflows; }); + Services.Configure(options => + { + options.Channels.AddRange(WorkflowDispatcherChannels.Values); + }); Services // Core. @@ -214,7 +251,7 @@ public class WorkflowRuntimeFeature : FeatureBase .AddNotificationHandler() .AddNotificationHandler() .AddNotificationHandler() - .AddNotificationHandler() + .AddNotificationHandler() .AddNotificationHandler() // Workflow activation strategies. diff --git a/src/modules/Elsa.Workflows.Runtime/Models/DispatchWorkflowOptions.cs b/src/modules/Elsa.Workflows.Runtime/Models/DispatchWorkflowOptions.cs new file mode 100644 index 000000000..cd276011b --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Models/DispatchWorkflowOptions.cs @@ -0,0 +1,6 @@ +namespace Elsa.Workflows.Runtime.Models; + +public class DispatchWorkflowOptions +{ + public string? Channel { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Models/DispatcherChannel.cs b/src/modules/Elsa.Workflows.Runtime/Models/DispatcherChannel.cs new file mode 100644 index 000000000..ee785c8aa --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Models/DispatcherChannel.cs @@ -0,0 +1,13 @@ +namespace Elsa.Workflows.Runtime.Models; + +/// +/// Describes a workflow dispatcher channel. +/// +public class DispatcherChannel +{ + /// + /// Gets or sets the name of the channel. + /// + public string Name { get; set; } = default!; + +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs new file mode 100644 index 000000000..385f13acc --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs @@ -0,0 +1,17 @@ +using Elsa.Workflows.Runtime.Models; + +namespace Elsa.Workflows.Runtime.Options; + +/// +/// Provides options to the workflow dispatcher. +/// +public class WorkflowDispatcherOptions +{ + /// + /// A list of available channels to dispatch workflows to. + /// + /// + /// Channels are used to dispatch workflows to different queues or endpoints. + /// + public ICollection Channels { get; set; } = new List(); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchResumeWorkflowsResponse.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchResumeWorkflowsResponse.cs deleted file mode 100644 index 0d63e9fe2..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchResumeWorkflowsResponse.cs +++ /dev/null @@ -1,3 +0,0 @@ -namespace Elsa.Workflows.Runtime.Requests; - -public record DispatchResumeWorkflowsResponse; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchTriggerWorkflowsResponse.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchTriggerWorkflowsResponse.cs deleted file mode 100644 index 601460c67..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchTriggerWorkflowsResponse.cs +++ /dev/null @@ -1,3 +0,0 @@ -namespace Elsa.Workflows.Runtime.Requests; - -public record DispatchTriggerWorkflowsResponse; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionRequest.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionRequest.cs index bd88eaf9f..103b0f760 100644 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionRequest.cs +++ b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionRequest.cs @@ -60,5 +60,5 @@ public class DispatchWorkflowDefinitionRequest /// /// The ID of the activity that triggered the workflow. /// - public string? TriggerActivityId { get; init; } + public string? TriggerActivityId { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionResponse.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionResponse.cs deleted file mode 100644 index 6320f22ef..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionResponse.cs +++ /dev/null @@ -1,3 +0,0 @@ -namespace Elsa.Workflows.Runtime.Requests; - -public record DispatchWorkflowDefinitionResponse(); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceResponse.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceResponse.cs deleted file mode 100644 index 79f91af3b..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceResponse.cs +++ /dev/null @@ -1,3 +0,0 @@ -namespace Elsa.Workflows.Runtime.Requests; - -public record DispatchWorkflowInstanceResponse; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Responses/DispatchWorkflowDefinitionResponse.cs b/src/modules/Elsa.Workflows.Runtime/Responses/DispatchWorkflowDefinitionResponse.cs new file mode 100644 index 000000000..744eeb8a2 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Responses/DispatchWorkflowDefinitionResponse.cs @@ -0,0 +1,18 @@ +namespace Elsa.Workflows.Runtime.Responses; + +/// +/// Represents the response of a dispatch action for a workflow definition. +/// +public record DispatchWorkflowResponse(bool Succeeded, string? ErrorMessage) +{ + /// + /// Creates a response indicating that the dispatch action was successful. + /// + /// + public static DispatchWorkflowResponse Success() => new DispatchWorkflowResponse(true, default); + + /// + /// Creates a response indicating that the specified channel does not exist. + /// + public static DispatchWorkflowResponse UnknownChannel() => new DispatchWorkflowResponse(false, "The specified channel does not exist."); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs index e08df6bef..cd8176a2e 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs @@ -2,7 +2,9 @@ using Elsa.Mediator; using Elsa.Mediator.Contracts; using Elsa.Workflows.Runtime.Commands; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; namespace Elsa.Workflows.Runtime.Services; @@ -22,7 +24,7 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher } /// - public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) + public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) { var command = new DispatchWorkflowDefinitionCommand(request.DefinitionId, request.VersionOptions) { @@ -34,11 +36,11 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher }; await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken); - return new DispatchWorkflowDefinitionResponse(); + return DispatchWorkflowResponse.Success(); } /// - public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default) + public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) { var command = new DispatchWorkflowInstanceCommand(request.InstanceId){ BookmarkId = request.BookmarkId, @@ -51,11 +53,11 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher CorrelationId = request.CorrelationId}; await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken); - return new DispatchWorkflowInstanceResponse(); + return DispatchWorkflowResponse.Success(); } /// - public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, CancellationToken cancellationToken = default) + public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) { var command = new DispatchTriggerWorkflowsCommand(request.ActivityTypeName, request.BookmarkPayload) { @@ -66,11 +68,11 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher Properties = request.Properties }; await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken); - return new DispatchTriggerWorkflowsResponse(); + return DispatchWorkflowResponse.Success(); } /// - public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, CancellationToken cancellationToken = default) + public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) { var command = new DispatchResumeWorkflowsCommand(request.ActivityTypeName, request.BookmarkPayload) { @@ -80,6 +82,6 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher Input = request.Input }; await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken); - return new DispatchResumeWorkflowsResponse(); + return DispatchWorkflowResponse.Success(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs index 7da481ea0..75932fb66 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs @@ -142,6 +142,6 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker return; } - await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken); + await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken: cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs index 1b9259f69..af952d15d 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs @@ -133,7 +133,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox WorkflowInstanceId = workflowInstanceId, ActivityInstanceId = activityInstanceId, Input = input - }, cancellationToken); + }, cancellationToken: cancellationToken); return new DeliverWorkflowInboxMessageResult(new List()); } @@ -153,7 +153,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox WorkflowInstanceId = workflowInstanceId, ActivityInstanceId = activityInstanceId, Input = input - }, cancellationToken); + }, cancellationToken: cancellationToken); } private async Task> ResumeWorkflowsSynchronouslyAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) diff --git a/src/modules/Elsa.Workflows.Runtime/Services/TaskReporter.cs b/src/modules/Elsa.Workflows.Runtime/Services/TaskReporter.cs index 4ceeadcb2..ff00e451a 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/TaskReporter.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/TaskReporter.cs @@ -31,6 +31,6 @@ public class TaskReporter : ITaskReporter var activityTypeName = ActivityTypeNameHelper.GenerateTypeName(); var request = new DispatchResumeWorkflowsRequest(activityTypeName, bookmarkPayload, Input: input); - await _workflowDispatcher.DispatchAsync(request, cancellationToken); + await _workflowDispatcher.DispatchAsync(request, cancellationToken: cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/ValidatingWorkflowDispatcher.cs b/src/modules/Elsa.Workflows.Runtime/Services/ValidatingWorkflowDispatcher.cs new file mode 100644 index 000000000..d0ddfad58 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Services/ValidatingWorkflowDispatcher.cs @@ -0,0 +1,63 @@ +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; +using Microsoft.Extensions.Options; + +namespace Elsa.Workflows.Runtime.Services; + +/// +/// Validates the workflow request before dispatching it to the workflow dispatcher. +/// +/// The workflow dispatcher to decorate. +public class ValidatingWorkflowDispatcher(IWorkflowDispatcher decoratedService, IOptions dispatcherOptions) : IWorkflowDispatcher +{ + private IWorkflowDispatcher DecoratedService { get; set; } = decoratedService; + + /// + public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + if (!ValidateChannel(options?.Channel)) + return DispatchWorkflowResponse.UnknownChannel(); + + return await DecoratedService.DispatchAsync(request, options, cancellationToken); + } + + /// + public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + if (!ValidateChannel(options?.Channel)) + return DispatchWorkflowResponse.UnknownChannel(); + + return await DecoratedService.DispatchAsync(request, options, cancellationToken); + } + + /// + public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + if (!ValidateChannel(options?.Channel)) + return DispatchWorkflowResponse.UnknownChannel(); + + return await DecoratedService.DispatchAsync(request, options, cancellationToken); + } + + /// + public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + if (!ValidateChannel(options?.Channel)) + return DispatchWorkflowResponse.UnknownChannel(); + + return await DecoratedService.DispatchAsync(request, options, cancellationToken); + } + + private bool ValidateChannel(string? channelName) + { + return string.IsNullOrEmpty(channelName) || GetChannelExists(channelName); + } + + private bool GetChannelExists(string channelName) + { + return dispatcherOptions.Value.Channels.Any(x => x.Name == channelName); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/UIHints/DispatcherChannelOptionsProvider.cs b/src/modules/Elsa.Workflows.Runtime/UIHints/DispatcherChannelOptionsProvider.cs new file mode 100644 index 000000000..f094903d9 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/UIHints/DispatcherChannelOptionsProvider.cs @@ -0,0 +1,31 @@ +using System.Reflection; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.UIHints.Dropdown; +using Microsoft.Extensions.Options; + +namespace Elsa.Workflows.Runtime.UIHints; + +/// +/// Provides options for activities that dispatch workflows to channels. +/// +public class DispatcherChannelOptionsProvider : DropDownOptionsProviderBase +{ + private readonly WorkflowDispatcherOptions _options; + + /// + public DispatcherChannelOptionsProvider(IOptions options) + { + _options = options.Value; + } + + /// + protected override ValueTask> GetItemsAsync(PropertyInfo propertyInfo, object? context, CancellationToken cancellationToken) + { + var channelNames = _options.Channels.Select(x => x.Name).ToList(); + var selectListItems = new List { new("Default", "") }; + + selectListItems.AddRange(channelNames.Select(x => new SelectListItem(x, x))); + + return new(selectListItems); + } +} \ No newline at end of file