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

Implement dispatch channels (#4949)

Browse source 
* 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.
This commit is contained in:
Sipke Schoorstra 2024-02-16 10:56:30 +01:00 committed by GitHub
parent ff7c5dce1e
commit ef3431c57a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
50 changed files with 558 additions and 190 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj.DotSettings Normal file
View file

@ -0,0 +1,2 @@
<wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=enums/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

10
src/bundles/Elsa.Server.Web/Enums/MassTransitBroker.cs Normal file
View file

@ -0,0 +1,10 @@
namespace Elsa.Server.Web;
/// <summary>
/// Represents the type of messaging broker used in MassTransit.
/// </summary>
public enum MassTransitBroker
{
AzureServiceBus,
RabbitMq
}

11
src/bundles/Elsa.Server.Web/Program.cs
View file

@ -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 =>
{

13
src/bundles/Elsa.Server.Web/appsettings.json
View file

@ -97,6 +97,19 @@
"WorkflowInboxCleanup": {
"SweepInterval": "00:00:10:00",
"BatchSize": 1000
},
"WorkflowDispatcher": {
"Channels": [
{
"Name": "Low"
},
{
"Name": "Medium"
},
{
"Name": "High"
}
]
}
},
"Scripting": {

2
src/modules/Elsa.Alterations.Core/Contracts/DefaultAlteredWorkflowScheduler.cs
View file

@ -27,6 +27,6 @@ public class DefaultAlteredWorkflowDispatcher : IAlteredWorkflowDispatcher
/// <inheritdoc />
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);
}
}

2
src/modules/Elsa.Alterations/Endpoints/Workflows/Retry/Endpoint.cs
View file

@ -56,7 +56,7 @@ public class Retry : ElsaEndpoint<Request, Response>
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.

2
src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs
View file

@ -37,6 +37,6 @@ public class ResumeWorkflowInstance : INotificationHandler<AlterationJobComplete
return;
// Resume workflow instance.
await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(job.WorkflowInstanceId), cancellationToken);
await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(job.WorkflowInstanceId), cancellationToken: cancellationToken);
}
}

2
src/modules/Elsa.Hangfire/Jobs/ResumeWorkflowJob.cs
View file

@ -24,5 +24,5 @@ public class ResumeWorkflowJob
/// <param name="name">The name of the job.</param>
/// <param name="request">The workflow request.</param>
/// <param name="cancellationToken">The cancellation token.</param>
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);
}

2
src/modules/Elsa.Hangfire/Jobs/RunWorkflowJob.cs
View file

@ -24,5 +24,5 @@ public class RunWorkflowJob
/// <param name="name">The name of the job.</param>
/// <param name="request">The workflow request.</param>
/// <param name="cancellationToken">The cancellation token.</param>
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);
}

14
src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs
View file

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

14
src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs
View file

@ -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<RabbitMqTransportOptions>? TransportOptions { get; set; }
/// <summary>
/// Configures the RabbitMQ bus.
/// </summary>
@ -34,14 +35,15 @@ public class RabbitMqServiceBusFeature : FeatureBase
{
Module.Configure<MassTransitFeature>().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));
});
};
}

2
src/modules/Elsa.MassTransit/Activities/MessageReceived.cs
View file

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

2
src/modules/Elsa.MassTransit/Activities/PublishMessage.cs
View file

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

2
src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs
View file

@ -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;
/// <summary>
/// A consumer of various dispatch message types to asynchronously execute workflows.
/// </summary>
[UsedImplicitly]
public class DispatchWorkflowRequestConsumer :
IConsumer<DispatchWorkflowDefinition>,
IConsumer<DispatchWorkflowInstance>,

2
src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs
View file

@ -36,6 +36,6 @@ public class WorkflowMessageConsumer<T> : IConsumer<T> where T : class
CorrelationId = correlationId,
Input = input
};
await _workflowRuntime.DispatchAsync(request, cancellationToken);
await _workflowRuntime.DispatchAsync(request, cancellationToken: cancellationToken);
}
}

14
src/modules/Elsa.MassTransit/Contracts/IEndpointChannelFormatter.cs Normal file
View file

@ -0,0 +1,14 @@
namespace Elsa.MassTransit.Contracts;
/// <summary>
/// Represents a contract for formatting channel queue names.
/// </summary>
public interface IEndpointChannelFormatter
{
/// <summary>
/// Formats the queue name based on the provided channel name.
/// </summary>
/// <param name="channelName">The channel name to be formatted.</param>
/// <returns>The formatted queue name.</returns>
string FormatEndpointName(string? channelName = default);
}

2
src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs
View file

@ -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

42
src/modules/Elsa.MassTransit/Extensions/ServiceBusChannelsConfigurationExtensions.cs Normal file
View file

@ -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;
/// <summary>
/// Provides extension methods for configuring service bus channel endpoints.
/// </summary>
public static class ServiceBusChannelsConfigurationExtensions
{
/// <summary>
/// Sets up channel endpoints for receiving workflow dispatch requests.
/// </summary>
/// <param name="configurator">The MassTransit receive configurator.</param>
/// <param name="context">The MassTransit bus registration context.</param>
public static void SetupWorkflowDispatcherEndpoints(this IReceiveConfigurator configurator, IBusRegistrationContext context)
{
var dispatcherOptions = context.GetRequiredService<IOptions<WorkflowDispatcherOptions>>();
var formatter = context.GetRequiredService<IEndpointChannelFormatter>();
var channelDescriptors = dispatcherOptions.Value.Channels;
var defaultEndpointName = formatter.FormatEndpointName();
var endpointNames = new List<string>
{
defaultEndpointName
};
endpointNames.AddRange(channelDescriptors.Select(x => formatter.FormatEndpointName(x.Name)));
foreach (string endpointName in endpointNames)
{
configurator.ReceiveEndpoint(endpointName, endpoint =>
{
endpoint.ConfigureConsumer<DispatchWorkflowRequestConsumer>(context);
});
}
}
}

22
src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs
View file

@ -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<MassTransitWorkflowDispatcherOptions>(x => { });
Services.AddActivityProvider<MassTransitActivityTypeProvider>();
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<MassTransitActivityOptions>(options => options.MessageTypes = new HashSet<Type>(messageTypes));
Services.Configure<MassTransitActivityOptions>(
options => options.MessageTypes = new HashSet<Type>(messageTypes));
// Add collected message types as available variable types.
Services.Configure<ManagementOptions>(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()));
}
/// <summary>
@ -89,7 +94,8 @@ public class MassTransitFeature : FeatureBase
{
// For each message type, create a concrete WorkflowMessageConsumer<T>.
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();

27
src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs
View file

@ -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
/// </summary>
public Action<MassTransitWorkflowDispatcherOptions>? ConfigureDispatcherOptions { get; set; }
/// <summary>
/// A factory that creates a <see cref="IEndpointChannelFormatter"/>.
/// </summary>
public Func<IServiceProvider, IEndpointChannelFormatter> ChannelQueueFormatterFactory { get; set; } = _ => new DefaultEndpointChannelFormatter();
/// <inheritdoc />
public override void Configure()
{
Module.AddMassTransitConsumer<DispatchWorkflowRequestConsumer, DispatchWorkflowRequestConsumerDefinition>();
Module.Configure<WorkflowRuntimeFeature>(f => f.WorkflowDispatcher = sp => sp.GetRequiredService<MassTransitWorkflowDispatcher>());
Module.Configure<WorkflowRuntimeFeature>(f => f.WorkflowDispatcher = sp =>
{
var decoratedService = ActivatorUtilities.CreateInstance<MassTransitWorkflowDispatcher>(sp);
return ActivatorUtilities.CreateInstance<ValidatingWorkflowDispatcher>(sp, decoratedService);
});
}
/// <inheritdoc />
@ -46,13 +56,6 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase
if (ConfigureDispatcherOptions != null)
options.Configure(ConfigureDispatcherOptions);
var queueName = KebabCaseEndpointNameFormatter.Instance.Consumer<DispatchWorkflowRequestConsumer>();
var queueAddress = new Uri($"queue:elsa-{queueName}");
EndpointConvention.Map<DispatchWorkflowDefinition>(queueAddress);
EndpointConvention.Map<DispatchWorkflowInstance>(queueAddress);
EndpointConvention.Map<DispatchTriggerWorkflows>(queueAddress);
EndpointConvention.Map<DispatchResumeWorkflows>(queueAddress);
Services.AddScoped<MassTransitWorkflowDispatcher>();
Services.AddSingleton(ChannelQueueFormatterFactory);
}
}

18
src/modules/Elsa.MassTransit/Formatters/DefaultEndpointChannelFormatter.cs Normal file
View file

@ -0,0 +1,18 @@
using Elsa.MassTransit.Contracts;
using Humanizer;
namespace Elsa.MassTransit.Formatters;
/// <summary>
/// The DefaultChannelQueueFormatter class is an implementation of the IChannelQueueFormatter interface.
/// It is responsible for formatting channel queue names.
/// </summary>
public class DefaultEndpointChannelFormatter : IEndpointChannelFormatter
{
/// <inheritdoc />
public string FormatEndpointName(string? channelName)
{
var channelSuffix = !string.IsNullOrEmpty(channelName) ? "-" + channelName.Kebaberize() : "";
return $"elsa-dispatch-workflow-request{channelSuffix}";
}
}

80
src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs
View file

@ -1,80 +0,0 @@
using Elsa.MassTransit.Messages;
using Elsa.Workflows.Runtime.Contracts;
using Elsa.Workflows.Runtime.Requests;
using MassTransit;
namespace Elsa.MassTransit.Implementations;
/// <summary>
/// Implements <see cref="IWorkflowDispatcher"/> by leveraging MassTransit.
/// </summary>
public class MassTransitWorkflowDispatcher : IWorkflowDispatcher
{
private readonly IBus _bus;
/// <summary>
/// Constructor.
/// </summary>
public MassTransitWorkflowDispatcher(IBus bus)
{
_bus = bus;
}
/// <inheritdoc />
public async Task<DispatchWorkflowDefinitionResponse> 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();
}
/// <inheritdoc />
public async Task<DispatchWorkflowInstanceResponse> 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();
}
/// <inheritdoc />
public async Task<DispatchTriggerWorkflowsResponse> 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();
}
/// <inheritdoc />
public async Task<DispatchResumeWorkflowsResponse> 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();
}
}

4
src/modules/Elsa.MassTransit/Implementations/MassTransitActivityTypeProvider.cs → src/modules/Elsa.MassTransit/Services/MassTransitActivityTypeProvider.cs
View file

@ -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;
/// <summary>
/// Provides activities to the system from the configured MassTransit message types.
/// </summary>
[UsedImplicitly]
public class MassTransitActivityTypeProvider(IActivityFactory activityFactory, IOptions<MassTransitActivityOptions> options, IActivityDescriber activityDescriber) : IActivityProvider
{
/// <inheritdoc />

98
src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs Normal file
View file

@ -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;
/// <summary>
/// Implements <see cref="IWorkflowDispatcher"/> by leveraging MassTransit.
/// </summary>
public class MassTransitWorkflowDispatcher : IWorkflowDispatcher
{
private readonly IBus _bus;
private readonly IEndpointChannelFormatter _endpointChannelFormatter;
/// <summary>
/// Initializes a new instance of the <see cref="MassTransitWorkflowDispatcher"/> class.
/// </summary>
public MassTransitWorkflowDispatcher(IBus bus, IEndpointChannelFormatter endpointChannelFormatter)
{
_bus = bus;
_endpointChannelFormatter = endpointChannelFormatter;
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> 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();
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> 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();
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> 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();
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> 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<ISendEndpoint> GetSendEndpointAsync(DispatchWorkflowOptions? options = default)
{
var endpointName = _endpointChannelFormatter.FormatEndpointName(options?.Channel);
var sendEndpoint = await _bus.GetSendEndpoint(new Uri($"queue:{endpointName}"));
return sendEndpoint;
}
}

2
src/modules/Elsa.Quartz/Jobs/ResumeWorkflowJob.cs
View file

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

2
src/modules/Elsa.Quartz/Jobs/RunWorkflowJob.cs
View file

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

2
src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs
View file

@ -26,6 +26,6 @@ public class ResumeWorkflowTask : ITask
public async ValueTask ExecuteAsync(TaskExecutionContext context)
{
var workflowDispatcher = context.ServiceProvider.GetRequiredService<IWorkflowDispatcher>();
await workflowDispatcher.DispatchAsync(_dispatchWorkflowInstanceRequest, context.CancellationToken);
await workflowDispatcher.DispatchAsync(_dispatchWorkflowInstanceRequest, cancellationToken: context.CancellationToken);
}
}

2
src/modules/Elsa.Scheduling/Tasks/RunWorkflowTask.cs
View file

@ -25,6 +25,6 @@ public class RunWorkflowTask : ITask
public async ValueTask ExecuteAsync(TaskExecutionContext context)
{
var workflowDispatcher = context.ServiceProvider.GetRequiredService<IWorkflowDispatcher>();
await workflowDispatcher.DispatchAsync(_request, context.CancellationToken);
await workflowDispatcher.DispatchAsync(_request, cancellationToken: context.CancellationToken);
}
}

2
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDispatch/Endpoint.cs
View file

@ -64,7 +64,7 @@ internal class Endpoint : ElsaEndpoint<Request, Response>
TriggerActivityId = triggerActivityId
};
await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken);
await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken: cancellationToken);
instanceIds.Add(instanceId);
}

52
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Endpoint.cs
View file

@ -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<Request, Response>
[UsedImplicitly]
internal class Endpoint(IWorkflowDefinitionStore store, IWorkflowDispatcher workflowDispatcher, IIdentityGenerator identityGenerator) : ElsaEndpoint<Request, Response>
{
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<Request, Response>
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<Request, Response>
await SendNotFoundAsync(cancellationToken);
return;
}
var input = request.Input;
if(input != null && input is not IDictionary<string, object>)
{
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<string, object>?)request.Input;
var instanceId = request.InstanceId ?? identityGenerator.GenerateId();
var dispatchRequest = new DispatchWorkflowDefinitionRequest
{
DefinitionId = definitionId,
VersionOptions = versionOptions,
Input = input,
Input = input as IDictionary<string, object>,
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);
}
}

2
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Models.cs
View file

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

28
src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs
View file

@ -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<bool> WaitForCompletion { get; set; } = new(true);
/// <summary>
/// The channel to dispatch the workflow to.
/// </summary>
[Input(
DisplayName = "Channel",
Description = "The channel to dispatch the workflow to.",
UIHint = InputUIHints.DropDown,
UIHandler = typeof(DispatcherChannelOptionsProvider)
)]
public Input<string?> ChannelName { get; set; } = default!;
/// <summary>
/// An activity to execute when the child workflow finishes.
/// </summary>
@ -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<string, object>();
var channelName = ChannelName.GetOrDefault(context);
var properties = new Dictionary<string, object>
{
["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;

26
src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs
View file

@ -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<object>
[Input(Description = "Wait for the child workflow to complete before completing this activity.")]
public Input<bool> WaitForCompletion { get; set; } = default!;
/// <summary>
/// The channel to dispatch the workflow to.
/// </summary>
[Input(
DisplayName = "Channel",
Description = "The channel to dispatch the workflow to.",
UIHint = InputUIHints.DropDown,
UIHandler = typeof(DispatcherChannelOptionsProvider)
)]
public Input<string?> ChannelName { get; set; } = default!;
/// <inheritdoc />
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
@ -85,6 +99,7 @@ public class DispatchWorkflow : Activity<object>
{
var workflowDefinitionId = WorkflowDefinitionId.Get(context);
var input = Input.GetOrDefault(context) ?? new Dictionary<string, object>();
var channelName = ChannelName.GetOrDefault(context);
input["ParentInstanceId"] = context.WorkflowExecutionContext.Id;
@ -98,11 +113,18 @@ public class DispatchWorkflow : Activity<object>
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;
}

10
src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDispatcher.cs
View file

@ -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
/// <summary>
/// Dispatches a request to execute the specified workflow definition.
/// </summary>
Task<DispatchWorkflowDefinitionResponse> DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default);
Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default);
/// <summary>
/// Dispatches a request to execute the specified workflow instance.
/// </summary>
Task<DispatchWorkflowInstanceResponse> DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default);
Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default);
/// <summary>
/// Starts all workflows and resumes existing workflow instances based on the specified activity type and bookmark payload.
/// </summary>
Task<DispatchTriggerWorkflowsResponse> DispatchAsync(DispatchTriggerWorkflowsRequest request, CancellationToken cancellationToken = default);
Task<DispatchWorkflowResponse> DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default);
/// <summary>
/// Resumes the workflow waiting for the specified bookmark.
/// </summary>
Task<DispatchResumeWorkflowsResponse> DispatchAsync(DispatchResumeWorkflowsRequest request, CancellationToken cancellationToken = default);
Task<DispatchWorkflowResponse> DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default);
}

43
src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs
View file

@ -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<string, DispatcherChannel> WorkflowDispatcherChannels { get; set; } = new Dictionary<string, DispatcherChannel>();
/// <summary>
/// A list of workflow builders configured during application startup.
/// </summary>
@ -46,7 +49,11 @@ public class WorkflowRuntimeFeature : FeatureBase
/// <summary>
/// A factory that instantiates an <see cref="IWorkflowDispatcher"/>.
/// </summary>
public Func<IServiceProvider, IWorkflowDispatcher> WorkflowDispatcher { get; set; } = sp => ActivatorUtilities.CreateInstance<BackgroundWorkflowDispatcher>(sp);
public Func<IServiceProvider, IWorkflowDispatcher> WorkflowDispatcher { get; set; } = sp =>
{
var decoratedService = ActivatorUtilities.CreateInstance<BackgroundWorkflowDispatcher>(sp);
return ActivatorUtilities.CreateInstance<ValidatingWorkflowDispatcher>(sp, decoratedService);
};
/// <summary>
/// A factory that instantiates an <see cref="IBookmarkStore"/>.
@ -102,6 +109,11 @@ public class WorkflowRuntimeFeature : FeatureBase
/// A delegate to configure the <see cref="WorkflowInboxCleanupOptions"/>.
/// </summary>
public Action<WorkflowInboxCleanupOptions> WorkflowInboxCleanupOptions { get; set; } = _ => { };
/// <summary>
/// A delegate to configure the <see cref="WorkflowDispatcherOptions"/>.
/// </summary>
public Action<WorkflowDispatcherOptions> WorkflowDispatcherOptions { get; set; } = _ => { };
/// <summary>
/// Register the specified workflow type.
@ -127,6 +139,26 @@ public class WorkflowRuntimeFeature : FeatureBase
return this;
}
/// <summary>
/// Adds a dispatcher channel.
/// </summary>
public WorkflowRuntimeFeature AddDispatcherChannel(string channel)
{
return AddDispatcherChannel(new DispatcherChannel
{
Name = channel
});
}
/// <summary>
/// Adds a dispatcher channel.
/// </summary>
public WorkflowRuntimeFeature AddDispatcherChannel(DispatcherChannel channel)
{
WorkflowDispatcherChannels[channel.Name] = channel;
return this;
}
/// <inheritdoc />
public override void Configure()
{
@ -146,8 +178,13 @@ public class WorkflowRuntimeFeature : FeatureBase
{
// Options.
Services.Configure(DistributedLockingOptions);
Services.Configure<RuntimeOptions>(options => { options.Workflows = Workflows; });
Services.Configure(WorkflowInboxCleanupOptions);
Services.Configure(WorkflowDispatcherOptions);
Services.Configure<RuntimeOptions>(options => { options.Workflows = Workflows; });
Services.Configure<WorkflowDispatcherOptions>(options =>
{
options.Channels.AddRange(WorkflowDispatcherChannels.Values);
});
Services
// Core.
@ -214,7 +251,7 @@ public class WorkflowRuntimeFeature : FeatureBase
.AddNotificationHandler<DeleteActivityExecutionLogRecords>()
.AddNotificationHandler<ReadWorkflowInboxMessage>()
.AddNotificationHandler<DeliverWorkflowMessagesFromInbox>()
.AddNotificationHandler<DeleteWorkflowExecutionLogRecords>()
.AddNotificationHandler<DeleteWorkflowExecutionLogRecords>()
.AddNotificationHandler<WorkflowExecutionContextNotificationsHandler>()
// Workflow activation strategies.

6
src/modules/Elsa.Workflows.Runtime/Models/DispatchWorkflowOptions.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Workflows.Runtime.Models;
public class DispatchWorkflowOptions
{
public string? Channel { get; set; }
}

13
src/modules/Elsa.Workflows.Runtime/Models/DispatcherChannel.cs Normal file
View file

@ -0,0 +1,13 @@
namespace Elsa.Workflows.Runtime.Models;
/// <summary>
/// Describes a workflow dispatcher channel.
/// </summary>
public class DispatcherChannel
{
/// <summary>
/// Gets or sets the name of the channel.
/// </summary>
public string Name { get; set; } = default!;
}

17
src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs Normal file
View file

@ -0,0 +1,17 @@
using Elsa.Workflows.Runtime.Models;
namespace Elsa.Workflows.Runtime.Options;
/// <summary>
/// Provides options to the workflow dispatcher.
/// </summary>
public class WorkflowDispatcherOptions
{
/// <summary>
/// A list of available channels to dispatch workflows to.
/// </summary>
/// <remarks>
/// Channels are used to dispatch workflows to different queues or endpoints.
/// </remarks>
public ICollection<DispatcherChannel> Channels { get; set; } = new List<DispatcherChannel>();
}

3
src/modules/Elsa.Workflows.Runtime/Requests/DispatchResumeWorkflowsResponse.cs
View file

@ -1,3 +0,0 @@
namespace Elsa.Workflows.Runtime.Requests;
public record DispatchResumeWorkflowsResponse;

3
src/modules/Elsa.Workflows.Runtime/Requests/DispatchTriggerWorkflowsResponse.cs
View file

@ -1,3 +0,0 @@
namespace Elsa.Workflows.Runtime.Requests;
public record DispatchTriggerWorkflowsResponse;

2
src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionRequest.cs
View file

@ -60,5 +60,5 @@ public class DispatchWorkflowDefinitionRequest
/// <summary>
/// The ID of the activity that triggered the workflow.
/// </summary>
public string? TriggerActivityId { get; init; }
public string? TriggerActivityId { get; set; }
}

3
src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionResponse.cs
View file

@ -1,3 +0,0 @@
namespace Elsa.Workflows.Runtime.Requests;
public record DispatchWorkflowDefinitionResponse();

3
src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceResponse.cs
View file

@ -1,3 +0,0 @@
namespace Elsa.Workflows.Runtime.Requests;
public record DispatchWorkflowInstanceResponse;

18
src/modules/Elsa.Workflows.Runtime/Responses/DispatchWorkflowDefinitionResponse.cs Normal file
View file

@ -0,0 +1,18 @@
namespace Elsa.Workflows.Runtime.Responses;
/// <summary>
/// Represents the response of a dispatch action for a workflow definition.
/// </summary>
public record DispatchWorkflowResponse(bool Succeeded, string? ErrorMessage)
{
/// <summary>
/// Creates a response indicating that the dispatch action was successful.
/// </summary>
/// <returns></returns>
public static DispatchWorkflowResponse Success() => new DispatchWorkflowResponse(true, default);
/// <summary>
/// Creates a response indicating that the specified channel does not exist.
/// </summary>
public static DispatchWorkflowResponse UnknownChannel() => new DispatchWorkflowResponse(false, "The specified channel does not exist.");
}

18
src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs
View file

@ -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
}
/// <inheritdoc />
public async Task<DispatchWorkflowDefinitionResponse> DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default)
public async Task<DispatchWorkflowResponse> 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();
}
/// <inheritdoc />
public async Task<DispatchWorkflowInstanceResponse> DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
public async Task<DispatchWorkflowResponse> 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();
}
/// <inheritdoc />
public async Task<DispatchTriggerWorkflowsResponse> DispatchAsync(DispatchTriggerWorkflowsRequest request, CancellationToken cancellationToken = default)
public async Task<DispatchWorkflowResponse> 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();
}
/// <inheritdoc />
public async Task<DispatchResumeWorkflowsResponse> DispatchAsync(DispatchResumeWorkflowsRequest request, CancellationToken cancellationToken = default)
public async Task<DispatchWorkflowResponse> 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();
}
}

2
src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs
View file

@ -142,6 +142,6 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker
return;
}
await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken);
await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken: cancellationToken);
}
}

4
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs
View file

@ -133,7 +133,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox
WorkflowInstanceId = workflowInstanceId,
ActivityInstanceId = activityInstanceId,
Input = input
}, cancellationToken);
}, cancellationToken: cancellationToken);
return new DeliverWorkflowInboxMessageResult(new List<WorkflowExecutionResult>());
}
@ -153,7 +153,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox
WorkflowInstanceId = workflowInstanceId,
ActivityInstanceId = activityInstanceId,
Input = input
}, cancellationToken);
}, cancellationToken: cancellationToken);
}
private async Task<IEnumerable<WorkflowExecutionResult>> ResumeWorkflowsSynchronouslyAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default)

2
src/modules/Elsa.Workflows.Runtime/Services/TaskReporter.cs
View file

@ -31,6 +31,6 @@ public class TaskReporter : ITaskReporter
var activityTypeName = ActivityTypeNameHelper.GenerateTypeName<RunTask>();
var request = new DispatchResumeWorkflowsRequest(activityTypeName, bookmarkPayload, Input: input);
await _workflowDispatcher.DispatchAsync(request, cancellationToken);
await _workflowDispatcher.DispatchAsync(request, cancellationToken: cancellationToken);
}
}

63
src/modules/Elsa.Workflows.Runtime/Services/ValidatingWorkflowDispatcher.cs Normal file
View file

@ -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;
/// <summary>
/// Validates the workflow request before dispatching it to the workflow dispatcher.
/// </summary>
/// <param name="decoratedService">The workflow dispatcher to decorate.</param>
public class ValidatingWorkflowDispatcher(IWorkflowDispatcher decoratedService, IOptions<WorkflowDispatcherOptions> dispatcherOptions) : IWorkflowDispatcher
{
private IWorkflowDispatcher DecoratedService { get; set; } = decoratedService;
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default)
{
if (!ValidateChannel(options?.Channel))
return DispatchWorkflowResponse.UnknownChannel();
return await DecoratedService.DispatchAsync(request, options, cancellationToken);
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default)
{
if (!ValidateChannel(options?.Channel))
return DispatchWorkflowResponse.UnknownChannel();
return await DecoratedService.DispatchAsync(request, options, cancellationToken);
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default)
{
if (!ValidateChannel(options?.Channel))
return DispatchWorkflowResponse.UnknownChannel();
return await DecoratedService.DispatchAsync(request, options, cancellationToken);
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> 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);
}
}

31
src/modules/Elsa.Workflows.Runtime/UIHints/DispatcherChannelOptionsProvider.cs Normal file
View file

@ -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;
/// <summary>
/// Provides options for activities that dispatch workflows to channels.
/// </summary>
public class DispatcherChannelOptionsProvider : DropDownOptionsProviderBase
{
private readonly WorkflowDispatcherOptions _options;
/// <inheritdoc />
public DispatcherChannelOptionsProvider(IOptions<WorkflowDispatcherOptions> options)
{
_options = options.Value;
}
/// <inheritdoc />
protected override ValueTask<ICollection<SelectListItem>> GetItemsAsync(PropertyInfo propertyInfo, object? context, CancellationToken cancellationToken)
{
var channelNames = _options.Channels.Select(x => x.Name).ToList();
var selectListItems = new List<SelectListItem> { new("Default", "") };
selectListItems.AddRange(channelNames.Select(x => new SelectListItem(x, x)));
return new(selectListItems);
}
}