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

Enhances Mediator with Tenant Context Propagation (#6738)

Browse source 
* Update package versions in Directory.Packages.props

Upgraded multiple package dependencies to latest versions, ensuring compatibility, security, and access to the newest features.

* Refactor mediator pipeline to support tenant context propagation

- Introduced `TenantPropagatingMiddleware` to handle tenant context propagation during command execution.
- Added `SetupMediatorPipelines` hosted service for configuring mediator pipelines.
- Enhanced `CommandPipeline` and builder to allow middleware insertion, removal, and reordering.
- Updated `CommandContext` and related components to support headers for tenant context handling.
- Improved logging and refactored `BackgroundWorkflowDispatcher` to include tenant headers during command dispatch.

* Fix typos in XML documentation and improve middleware extension clarity

- Corrected duplicated slashes in XML doc comments in `ICommandSender.cs`.
- Refined phrasing in `MiddlewareExtensions.cs` to clarify method parameters and improve readability.

* Update src/common/Elsa.Mediator/Middleware/Command/Components/CommandLoggingMiddleware.cs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Sipke Schoorstra 2025-06-13 14:26:51 +02:00 committed by GitHub
parent 623b75ed2c
commit c694a18c13
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
29 changed files with 375 additions and 220 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
Directory.Packages.props
View file

@ -126,47 +126,47 @@
<PackageVersion Include="xunit.runner.visualstudio" Version="3.0.2"/>
<PackageVersion Include="Yarp.ReverseProxy" Version="2.3.0"/>
<PackageVersion Include="AspNetCore.Authentication.ApiKey" Version="8.0.1"/>
<PackageVersion Include="Microsoft.AspNetCore.Authorization" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Components" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly.DevServer" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly.Server" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.DataProtection.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Data.Sqlite" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Data.Sqlite.Core" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Design" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Relational" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Sqlite" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.SqlServer" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Caching.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Configuration" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Http" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Http.Polly" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Logging" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Options" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="9.0.3"/>
<PackageVersion Include="MySql.Data" Version="9.2.0"/>
<PackageVersion Include="Microsoft.AspNetCore.Authorization" Version="9.0.6"/>
<PackageVersion Include="Microsoft.AspNetCore.Components" Version="9.0.6"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly" Version="9.0.6"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly.DevServer" Version="9.0.6"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly.Server" Version="9.0.6"/>
<PackageVersion Include="Microsoft.AspNetCore.DataProtection.Abstractions" Version="9.0.6"/>
<PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Data.Sqlite" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Data.Sqlite.Core" Version="9.0.6"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore" Version="9.0.6"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Design" Version="9.0.6"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Relational" Version="9.0.6"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Sqlite" Version="9.0.6"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.SqlServer" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Caching.Abstractions" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Configuration" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Http" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Http.Polly" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Logging" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Options" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="9.0.6"/>
<PackageVersion Include="MySql.Data" Version="9.3.0"/>
<PackageVersion Include="Npgsql" Version="9.0.3"/>
<PackageVersion Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="9.0.4"/>
<PackageVersion Include="Oracle.EntityFrameworkCore" Version="9.23.60"/>
<PackageVersion Include="Polly" Version="8.5.2"/>
<PackageVersion Include="Oracle.EntityFrameworkCore" Version="9.23.80"/>
<PackageVersion Include="Polly" Version="8.6.0"/>
<PackageVersion Include="Pomelo.EntityFrameworkCore.MySql" Version="9.0.0-preview.3.efcore.9.0.0"/>
<PackageVersion Include="Refit" Version="8.0.0"/>
<PackageVersion Include="Refit.HttpClientFactory" Version="8.0.0"/>
<PackageVersion Include="System.Formats.Asn1" Version="9.0.3"/>
<PackageVersion Include="System.Text.Json" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="9.3.0"/>
<PackageVersion Include="Microsoft.Extensions.Resilience" Version="9.3.0"/>
<PackageVersion Include="System.Formats.Asn1" Version="9.0.6"/>
<PackageVersion Include="System.Text.Json" Version="9.0.6"/>
<PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="9.6.0"/>
<PackageVersion Include="Microsoft.Extensions.Resilience" Version="9.6.0"/>
</ItemGroup>
</Project>

4
src/apps/Elsa.Server.Web/Program.cs
View file

@ -89,7 +89,7 @@ const PersistenceProvider persistenceProvider = PersistenceProvider.EntityFramew
const bool useDbContextPooling = false;
const bool useHangfire = false;
const bool useQuartz = true;
const bool useMassTransit = true;
const bool useMassTransit = false;
const bool useZipCompression = false;
const bool runEFCoreMigrations = true;
const bool useMemoryStores = false;
@ -98,7 +98,7 @@ const bool useKafka = false;
const bool useReadOnlyMode = false;
const bool useSignalR = false; // Disabled until Elsa Studio sends authenticated requests.
const WorkflowRuntime workflowRuntime = WorkflowRuntime.Distributed;
const DistributedCachingTransport distributedCachingTransport = DistributedCachingTransport.MassTransit;
const DistributedCachingTransport distributedCachingTransport = DistributedCachingTransport.Memory;
const MassTransitBroker massTransitBroker = MassTransitBroker.Memory;
const bool useMultitenancy = false;
const bool useTenantsFromConfiguration = true;

2
src/apps/Elsa.Server.Web/appsettings.json
View file

@ -44,7 +44,7 @@
},
{
"Id": "tenant-1",
"Name": "Tenant 12 dd",
"Name": "Tenant 1",
"Configuration": {
"Http": {
"Prefix": "/tenant-1",

3
src/common/Elsa.Mediator/Channels/CommandsChannel.cs
View file

@ -1,9 +1,10 @@
using Elsa.Mediator.Abstractions;
using Elsa.Mediator.Contracts;
using Elsa.Mediator.Middleware.Command;
namespace Elsa.Mediator.Channels;
/// <inheritdoc cref="Elsa.Mediator.Contracts.ICommandsChannel" />
public class CommandsChannel : ChannelBase<ICommand>, ICommandsChannel
public class CommandsChannel : ChannelBase<CommandContext>, ICommandsChannel
{
}

2
src/common/Elsa.Mediator/CommandStrategies/BackgroundStrategy.cs
View file

@ -13,7 +13,7 @@ public class BackgroundStrategy : ICommandStrategy
public async Task<TResult> ExecuteAsync<TResult>(CommandStrategyContext context)
{
var commandsChannel = context.ServiceProvider.GetRequiredService<ICommandsChannel>();
await commandsChannel.Writer.WriteAsync(context.Command, context.CancellationToken);
await commandsChannel.Writer.WriteAsync(context.CommandContext, context.CancellationToken);
return default!;
}
}

3
src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs
View file

@ -12,7 +12,8 @@ public class DefaultStrategy : ICommandStrategy
/// <inheritdoc />
public async Task<TResult> ExecuteAsync<TResult>(CommandStrategyContext context)
{
var command = context.Command;
var commandContext = context.CommandContext;
var command = commandContext.Command;
var cancellationToken = context.CancellationToken;
var commandType = command.GetType();
var handleMethod = commandType.GetCommandHandlerMethod();

5
src/common/Elsa.Mediator/Contexts/CommandStrategyContext.cs
View file

@ -1,12 +1,13 @@
using Elsa.Mediator.Contracts;
using Elsa.Mediator.Middleware.Command;
namespace Elsa.Mediator.Contexts;
/// <summary>
/// Represents a context for executing a command.
/// </summary>
/// <param name="Command">The command to execute.</param>
/// <param name="CommandContext">The command context to execute.</param>
/// <param name="Handler">The command handler.</param>
/// <param name="ServiceProvider">The service provider to resolve services from.</param>
/// <param name="CancellationToken">The cancellation token.</param>
public record CommandStrategyContext(ICommand Command, ICommandHandler Handler, IServiceProvider ServiceProvider, CancellationToken CancellationToken = default);
public record CommandStrategyContext(CommandContext CommandContext, ICommandHandler Handler, IServiceProvider ServiceProvider, CancellationToken CancellationToken = default);

32
src/common/Elsa.Mediator/Contracts/ICommandSender.cs
View file

@ -6,13 +6,19 @@ namespace Elsa.Mediator.Contracts;
public interface ICommandSender
{
/// <summary>
/// Sends a command using he default strategy.
/// Sends a command using the default strategy.
/// </summary>
Task<T> SendAsync<T>(ICommand<T> command, CancellationToken cancellationToken = default);
/// <summary>
/// Sends a command using the default strategy.
/// </summary>
/// <param name="command">The command to send.</param>
/// <param name="headers">Any headers to pass along.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <typeparam name="T">The type of the result.</typeparam>
/// <returns>The result.</returns>
Task<T> SendAsync<T>(ICommand<T> command, CancellationToken cancellationToken = default);
Task<T> SendAsync<T>(ICommand<T> command, IDictionary<object, object> headers, CancellationToken cancellationToken = default);
/// <summary>
/// Sends a command using the specified strategy.
@ -24,6 +30,17 @@ public interface ICommandSender
/// <returns>The result.</returns>
Task<T> SendAsync<T>(ICommand<T> command, ICommandStrategy strategy, CancellationToken cancellationToken = default);
/// <summary>
/// Sends a command using the specified strategy.
/// </summary>
/// <param name="command">The command to send.</param>
/// <param name="strategy">The command strategy to use.</param>
/// <param name="headers">Any headers to pass along.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <typeparam name="T">The type of the result.</typeparam>
/// <returns>The result.</returns>
Task<T> SendAsync<T>(ICommand<T> command, ICommandStrategy strategy, IDictionary<object, object> headers, CancellationToken cancellationToken = default);
/// <summary>
/// Sends a command using the default strategy.
/// </summary>
@ -37,5 +54,14 @@ public interface ICommandSender
/// <param name="command">The command to send.</param>
/// <param name="strategy">The command strategy to use.</param>
/// <param name="cancellationToken">The cancellation token.</param>
Task SendAsync(ICommand command, ICommandStrategy? strategy, CancellationToken cancellationToken = default);
Task SendAsync(ICommand command, ICommandStrategy strategy, CancellationToken cancellationToken = default);
/// <summary>
/// Sends a command using the specified strategy.
/// </summary>
/// <param name="command">The command to send.</param>
/// <param name="strategy">The command strategy to use.</param>
/// <param name="headers">Any headers to pass along.</param>
/// <param name="cancellationToken">The cancellation token.</param>
Task SendAsync(ICommand command, ICommandStrategy strategy, IDictionary<object, object> headers, CancellationToken cancellationToken = default);
}

5
src/common/Elsa.Mediator/Contracts/ICommandsChannel.cs
View file

@ -1,4 +1,5 @@
using System.Threading.Channels;
using Elsa.Mediator.Middleware.Command;
namespace Elsa.Mediator.Contracts;
@ -10,10 +11,10 @@ public interface ICommandsChannel
/// <summary>
/// Gets the writer for the commands queue.
/// </summary>
ChannelWriter<ICommand> Writer { get; }
ChannelWriter<CommandContext> Writer { get; }
/// <summary>
/// Gets the reader for the commands queue.
/// </summary>
ChannelReader<ICommand> Reader { get; }
ChannelReader<CommandContext> Reader { get; }
}

8
src/common/Elsa.Mediator/Extensions/DependencyInjectionExtensions.cs
View file

@ -36,9 +36,9 @@ public static class DependencyInjectionExtensions
.AddScoped<IRequestSender>(sp => sp.GetRequiredService<IMediator>())
.AddScoped<ICommandSender>(sp => sp.GetRequiredService<IMediator>())
.AddScoped<INotificationSender>(sp => sp.GetRequiredService<IMediator>())
.AddScoped<IRequestPipeline, RequestPipeline>()
.AddScoped<ICommandPipeline, CommandPipeline>()
.AddScoped<INotificationPipeline, NotificationPipeline>()
.AddSingleton<IRequestPipeline, RequestPipeline>()
.AddSingleton<ICommandPipeline, CommandPipeline>()
.AddSingleton<INotificationPipeline, NotificationPipeline>()
;
}
@ -53,7 +53,7 @@ public static class DependencyInjectionExtensions
.AddSingleton<INotificationsChannel, NotificationsChannel>()
.AddSingleton<ICommandsChannel, CommandsChannel>()
.AddSingleton<IJobsChannel, JobsChannel>()
.AddSingleton<IJobQueue, JobQueue>()
.AddSingleton<IJobQueue, JobQueue>()
.AddHostedService<JobRunnerHostedService>()
.AddHostedService<BackgroundCommandSenderHostedService>()
.AddHostedService<BackgroundEventPublisherHostedService>();

21
src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs
View file

@ -1,5 +1,6 @@
using System.Threading.Channels;
using Elsa.Mediator.Contracts;
using Elsa.Mediator.Middleware.Command;
using Elsa.Mediator.Options;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
@ -16,7 +17,7 @@ public class BackgroundCommandSenderHostedService : BackgroundService
private readonly int _workerCount;
private readonly ICommandsChannel _commandsChannel;
private readonly IServiceScopeFactory _scopeFactory;
private readonly List<Channel<ICommand>> _outputs;
private readonly List<Channel<CommandContext>> _outputs;
private readonly ILogger _logger;
/// <inheritdoc />
@ -26,7 +27,7 @@ public class BackgroundCommandSenderHostedService : BackgroundService
_commandsChannel = commandsChannel;
_scopeFactory = scopeFactory;
_logger = logger;
_outputs = new List<Channel<ICommand>>(_workerCount);
_outputs = new(_workerCount);
}
/// <inheritdoc />
@ -36,34 +37,32 @@ public class BackgroundCommandSenderHostedService : BackgroundService
for (var i = 0; i < _workerCount; i++)
{
var output = Channel.CreateUnbounded<ICommand>();
var output = Channel.CreateUnbounded<CommandContext>();
_outputs.Add(output);
_ = ReadOutputAsync(output, cancellationToken);
}
await foreach (var command in _commandsChannel.Reader.ReadAllAsync(cancellationToken))
await foreach (var commandContext in _commandsChannel.Reader.ReadAllAsync(cancellationToken))
{
var output = _outputs[index];
await output.Writer.WriteAsync(command, cancellationToken);
await output.Writer.WriteAsync(commandContext, cancellationToken);
index = (index + 1) % _workerCount;
}
foreach (var output in _outputs)
{
foreach (var output in _outputs)
output.Writer.Complete();
}
}
private async Task ReadOutputAsync(Channel<ICommand> output, CancellationToken cancellationToken)
private async Task ReadOutputAsync(Channel<CommandContext> output, CancellationToken cancellationToken)
{
await foreach (var command in output.Reader.ReadAllAsync(cancellationToken))
await foreach (var commandContext in output.Reader.ReadAllAsync(cancellationToken))
{
try
{
using var scope = _scopeFactory.CreateScope();
var commandSender = scope.ServiceProvider.GetRequiredService<ICommandSender>();
await commandSender.SendAsync(command, CommandStrategy.Default, cancellationToken);
await commandSender.SendAsync(commandContext.Command, CommandStrategy.Default, commandContext.Headers, cancellationToken);
}
catch (Exception e)
{

14
src/common/Elsa.Mediator/Middleware/Command/CommandContext.cs
View file

@ -10,11 +10,13 @@ public class CommandContext
/// <summary>
/// Initializes a new instance of the <see cref="CommandContext"/> class.
/// </summary>
public CommandContext(ICommand command, ICommandStrategy commandStrategy, Type resultType, CancellationToken cancellationToken)
public CommandContext(ICommand command, ICommandStrategy commandStrategy, Type resultType, IDictionary<object, object> headers, IServiceProvider serviceProvider, CancellationToken cancellationToken)
{
Command = command;
CommandStrategy = commandStrategy;
ResultType = resultType;
Headers = headers;
ServiceProvider = serviceProvider;
CancellationToken = cancellationToken;
}
@ -28,6 +30,16 @@ public class CommandContext
/// </summary>
public ICommandStrategy CommandStrategy { get; }
/// <summary>
/// Gets or sets the headers associated with the command context.
/// </summary>
public IDictionary<object, object> Headers { get; }
/// <summary>
/// Gets the service provider used to resolve services for the command context.
/// </summary>
public IServiceProvider ServiceProvider { get; }
/// <summary>
/// Gets the result type.
/// </summary>

25
src/common/Elsa.Mediator/Middleware/Command/CommandPipeline.cs
View file

@ -5,28 +5,29 @@ namespace Elsa.Mediator.Middleware.Command;
/// <inheritdoc />
public class CommandPipeline : ICommandPipeline
{
private readonly IServiceProvider _serviceProvider;
private CommandMiddlewareDelegate? _pipeline;
private readonly CommandPipelineBuilder _builder;
private CommandMiddlewareDelegate _pipeline = null!;
/// <summary>
/// Constructor.
/// </summary>
public CommandPipeline(IServiceProvider serviceProvider) => _serviceProvider = serviceProvider;
/// <inheritdoc />
public CommandMiddlewareDelegate Pipeline => _pipeline ??= CreateDefaultPipeline();
public CommandPipeline(IServiceProvider serviceProvider)
{
_builder = new(serviceProvider);
Setup(x => x.UseCommandInvoker().UseCommandLogging());
}
/// <inheritdoc />
public CommandMiddlewareDelegate Setup(Action<ICommandPipelineBuilder>? setup = default)
public CommandMiddlewareDelegate Pipeline => _pipeline;
/// <inheritdoc />
public CommandMiddlewareDelegate Setup(Action<ICommandPipelineBuilder>? setup = null)
{
var builder = new CommandPipelineBuilder(_serviceProvider);
setup?.Invoke(builder);
_pipeline = builder.Build();
setup?.Invoke(_builder);
_pipeline = _builder.Build();
return _pipeline;
}
/// <inheritdoc />
public async Task InvokeAsync(CommandContext context) => await Pipeline(context);
private CommandMiddlewareDelegate CreateDefaultPipeline() => Setup(x => x.UseCommandInvoker().UseCommandLogging());
}

36
src/common/Elsa.Mediator/Middleware/Command/CommandPipelineBuilder.cs
View file

@ -33,19 +33,45 @@ public class CommandPipelineBuilder : ICommandPipelineBuilder
return this;
}
/// <inheritdoc />
public ICommandPipelineBuilder Use(int index, Func<CommandMiddlewareDelegate, CommandMiddlewareDelegate> middleware)
{
_components.Insert(index, middleware);
return this;
}
/// <inheritdoc />
public ICommandPipelineBuilder Remove(Func<CommandMiddlewareDelegate, CommandMiddlewareDelegate> middleware)
{
_components.Remove(middleware);
return this;
}
/// <inheritdoc />
public ICommandPipelineBuilder RemoveAt(int index)
{
_components.RemoveAt(index);
return this;
}
/// <inheritdoc />
public ICommandPipelineBuilder Clear()
{
_components.Clear();
return this;
}
/// <inheritdoc />
public CommandMiddlewareDelegate Build()
{
CommandMiddlewareDelegate pipeline = _ => new ValueTask();
CommandMiddlewareDelegate pipeline = _ => new();
for (int i = _components.Count - 1; i >= 0; i--)
{
for (var i = _components.Count - 1; i >= 0; i--)
pipeline = _components[i](pipeline);
}
return pipeline;
}
private T? GetProperty<T>(string key) => Properties.TryGetValue(key, out var value) ? (T?)value : default(T);
private T? GetProperty<T>(string key) => Properties.TryGetValue(key, out var value) ? (T?)value : default;
private void SetProperty<T>(string key, T value) => Properties[key] = value;
}

31
src/common/Elsa.Mediator/Middleware/Command/Components/CommandHandlerInvokerMiddleware.cs
View file

@ -1,28 +1,17 @@
using Elsa.Mediator.Contexts;
using Elsa.Mediator.Contracts;
using Elsa.Mediator.Middleware.Command.Contracts;
using JetBrains.Annotations;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Mediator.Middleware.Command.Components;
/// <summary>
/// A command middleware that invokes the command.
/// </summary>
public class CommandHandlerInvokerMiddleware : ICommandMiddleware
[UsedImplicitly]
public class CommandHandlerInvokerMiddleware(CommandMiddlewareDelegate next) : ICommandMiddleware
{
private readonly CommandMiddlewareDelegate _next;
private readonly IServiceProvider _serviceProvider;
private readonly IEnumerable<ICommandHandler> _commandHandlers;
/// <summary>
/// Constructor.
/// </summary>
public CommandHandlerInvokerMiddleware(CommandMiddlewareDelegate next, IEnumerable<ICommandHandler> commandHandlers, IServiceProvider serviceProvider)
{
_next = next;
_serviceProvider = serviceProvider;
_commandHandlers = commandHandlers.DistinctBy(x => x.GetType()).ToList();
}
/// <inheritdoc />
public async ValueTask InvokeAsync(CommandContext context)
{
@ -31,7 +20,9 @@ public class CommandHandlerInvokerMiddleware : ICommandMiddleware
var commandType = command.GetType();
var resultType = context.ResultType;
var handlerType = typeof(ICommandHandler<,>).MakeGenericType(commandType, resultType);
var handlers = _commandHandlers.Where(x => handlerType.IsInstanceOfType(x)).ToArray();
var serviceProvider = context.ServiceProvider;
var commandHandlers = serviceProvider.GetServices<ICommandHandler>();
var handlers = commandHandlers.DistinctBy(x => x.GetType()).Where(x => handlerType.IsInstanceOfType(x)).ToArray();
if (handlers.Length == 0)
throw new InvalidOperationException($"There is no handler to handle the {commandType.FullName} command");
@ -40,20 +31,20 @@ public class CommandHandlerInvokerMiddleware : ICommandMiddleware
throw new InvalidOperationException($"Multiple handlers were found to handle the {commandType.FullName} command");
var handler = handlers.First();
var strategyContext = new CommandStrategyContext(command, handler, _serviceProvider, context.CancellationToken);
var strategyContext = new CommandStrategyContext(context, handler, serviceProvider, context.CancellationToken);
var strategy = context.CommandStrategy;
var executeMethod = strategy.GetType().GetMethod(nameof(ICommandStrategy.ExecuteAsync))!;
var executeMethodWithReturnType = executeMethod.MakeGenericMethod(resultType);
// Execute command.
var task = executeMethodWithReturnType.Invoke(strategy, new object[] { strategyContext });
var task = executeMethodWithReturnType.Invoke(strategy, [strategyContext]);
// Get result of task.
// Get the result of the task.
var taskWithReturnType = typeof(Task<>).MakeGenericType(resultType);
var resultProperty = taskWithReturnType.GetProperty(nameof(Task<object>.Result))!;
context.Result = resultProperty.GetValue(task);
// Invoke next middleware.
await _next(context);
await next(context);
}
}

24
src/common/Elsa.Mediator/Middleware/Command/Components/CommandLoggingMiddleware.cs
View file

@ -1,5 +1,6 @@
using Elsa.Mediator.Middleware.Command.Contracts;
using Elsa.Mediator.Models;
using JetBrains.Annotations;
using Microsoft.Extensions.Logging;
namespace Elsa.Mediator.Middleware.Command.Components;
@ -7,31 +8,20 @@ namespace Elsa.Mediator.Middleware.Command.Components;
/// <summary>
/// A command middleware that logs the command being invoked.
/// </summary>
public class CommandLoggingMiddleware : ICommandMiddleware
[UsedImplicitly]
public class CommandLoggingMiddleware(CommandMiddlewareDelegate next, ILogger<CommandLoggingMiddleware> logger) : ICommandMiddleware
{
private readonly CommandMiddlewareDelegate _next;
private readonly ILogger<CommandLoggingMiddleware> _logger;
/// <summary>
/// Constructor.
/// </summary>
public CommandLoggingMiddleware(CommandMiddlewareDelegate next, ILogger<CommandLoggingMiddleware> logger)
{
_next = next;
_logger = logger;
}
/// <inheritdoc />
public async ValueTask InvokeAsync(CommandContext context)
{
var commandType = context.Command.GetType();
_logger.LogInformation("Invoking {CommandName}", commandType.Name);
logger.LogInformation("Invoking {CommandName}", commandType.Name);
await _next(context);
await next(context);
if (context.Result is null or Unit)
_logger.LogInformation("{CommandName} completed with no result", commandType.Name);
logger.LogInformation("{CommandName} completed with no result", commandType.Name);
else
_logger.LogInformation("{CommandName} completed wit result {CommandResult}", commandType.Name, context.Result);
logger.LogInformation("{CommandName} completed with result {CommandResult}", commandType.Name, context.Result);
}
}

2
src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipeline.cs
View file

@ -1,7 +1,7 @@
namespace Elsa.Mediator.Middleware.Command.Contracts;
/// <summary>
///
/// Represents a pipeline for processing commands. The pipeline is responsible for orchestrating the execution of registered middleware in sequence.
/// </summary>
public interface ICommandPipeline
{

24
src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipelineBuilder.cs
View file

@ -16,11 +16,29 @@ public interface ICommandPipelineBuilder
IServiceProvider ApplicationServices { get; }
/// <summary>
/// Adds a middleware component to the pipeline.
/// Appends a middleware component to the pipeline.
/// </summary>
/// <param name="middleware">The middleware component.</param>
/// <returns>The pipeline builder.</returns>
ICommandPipelineBuilder Use(Func<CommandMiddlewareDelegate, CommandMiddlewareDelegate> middleware);
/// <summary>
/// Adds a middleware component at the specified index.
/// </summary>
ICommandPipelineBuilder Use(int index, Func<CommandMiddlewareDelegate, CommandMiddlewareDelegate> middleware);
/// <summary>
/// Removes a middleware component from the pipeline.
/// </summary>
ICommandPipelineBuilder Remove(Func<CommandMiddlewareDelegate, CommandMiddlewareDelegate> middleware);
/// <summary>
/// Removes a middleware component at the specified index from the pipeline.
/// </summary>
ICommandPipelineBuilder RemoveAt(int index);
/// <summary>
/// Clears the pipeline.
/// </summary>
ICommandPipelineBuilder Clear();
/// <summary>
/// Builds the pipeline.

37
src/common/Elsa.Mediator/Middleware/Command/MiddlewareExtensions.cs
View file

@ -11,20 +11,35 @@ public static class MiddlewareExtensions
/// <summary>
/// Adds middleware to the pipeline.
/// </summary>
/// <param name="builder">The pipeline builder.</param>
/// <param name="args">Any arguments to pass to the middleware constructor.</param>
/// <typeparam name="TMiddleware">The middleware type.</typeparam>
/// <returns>The pipeline builder.</returns>
public static ICommandPipelineBuilder UseMiddleware<TMiddleware>(this ICommandPipelineBuilder builder, params object[] args) where TMiddleware : ICommandMiddleware
{
var middleware = typeof(TMiddleware);
return builder.Use(next => BuildMiddlewareDelegate<TMiddleware>(builder, next, args));
}
return builder.Use(next =>
/// <summary>
/// Inserts middleware at a specific index in the pipeline.
/// </summary>
public static ICommandPipelineBuilder UseMiddleware<TMiddleware>(this ICommandPipelineBuilder builder, int index, params object[] args) where TMiddleware : ICommandMiddleware
{
return builder.Use(index, next => BuildMiddlewareDelegate<TMiddleware>(builder, next, args));
}
/// <summary>
/// Builds a delegate for the middleware type.
/// </summary>
private static CommandMiddlewareDelegate BuildMiddlewareDelegate<TMiddleware>(
ICommandPipelineBuilder builder,
CommandMiddlewareDelegate next,
object[] args
) where TMiddleware : ICommandMiddleware
{
var middleware = typeof(TMiddleware);
var invokeMethod = MiddlewareHelpers.GetInvokeMethod(middleware);
var ctorParams = new[]
{
var invokeMethod = MiddlewareHelpers.GetInvokeMethod(middleware);
var ctorParams = new[] { next }.Concat(args).Select(x => x!).ToArray();
var instance = ActivatorUtilities.CreateInstance(builder.ApplicationServices, middleware, ctorParams);
return (CommandMiddlewareDelegate)invokeMethod.CreateDelegate(typeof(CommandMiddlewareDelegate), instance);
});
next
}.Concat(args).Select(x => x!).ToArray();
var instance = ActivatorUtilities.CreateInstance(builder.ApplicationServices, middleware, ctorParams);
return (CommandMiddlewareDelegate)invokeMethod.CreateDelegate(typeof(CommandMiddlewareDelegate), instance);
}
}

36
src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs
View file

@ -1,33 +1,19 @@
using Elsa.Mediator.Contexts;
using Elsa.Mediator.Contracts;
using Elsa.Mediator.Middleware.Notification.Contracts;
using JetBrains.Annotations;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Elsa.Mediator.Middleware.Notification.Components;
/// <inheritdoc />
public class NotificationHandlerInvokerMiddleware : INotificationMiddleware
[UsedImplicitly]
public class NotificationHandlerInvokerMiddleware(
NotificationMiddlewareDelegate next,
ILogger<NotificationHandlerInvokerMiddleware> logger)
: INotificationMiddleware
{
private readonly NotificationMiddlewareDelegate _next;
private readonly ILogger<NotificationHandlerInvokerMiddleware> _logger;
private readonly IServiceProvider _serviceProvider;
private readonly IEnumerable<INotificationHandler> _notificationHandlers;
/// <summary>
/// Initializes a new instance of the <see cref="NotificationHandlerInvokerMiddleware"/> class.
/// </summary>
public NotificationHandlerInvokerMiddleware(
NotificationMiddlewareDelegate next,
ILogger<NotificationHandlerInvokerMiddleware> logger,
IServiceProvider serviceProvider,
IEnumerable<INotificationHandler> notificationHandlers)
{
_next = next;
_logger = logger;
_serviceProvider = serviceProvider;
_notificationHandlers = notificationHandlers;
}
/// <inheritdoc />
public async ValueTask InvokeAsync(NotificationContext context)
{
@ -35,12 +21,14 @@ public class NotificationHandlerInvokerMiddleware : INotificationMiddleware
var notification = context.Notification;
var notificationType = notification.GetType();
var handlerType = typeof(INotificationHandler<>).MakeGenericType(notificationType);
var handlers = _notificationHandlers.Where(x => handlerType.IsInstanceOfType(x)).DistinctBy(x => x.GetType()).ToArray();
var strategyContext = new NotificationStrategyContext(notification, handlers, _logger, _serviceProvider, context.CancellationToken);
var serviceProvider = context.ServiceProvider;
var notificationHandlers = serviceProvider.GetServices<INotificationHandler>();
var handlers = notificationHandlers.Where(x => handlerType.IsInstanceOfType(x)).DistinctBy(x => x.GetType()).ToArray();
var strategyContext = new NotificationStrategyContext(notification, handlers, logger, serviceProvider, context.CancellationToken);
await context.NotificationStrategy.PublishAsync(strategyContext);
// Invoke next middleware.
await _next(context);
await next(context);
}
}

2
src/common/Elsa.Mediator/Middleware/Notification/MiddlewareExtensions.cs
View file

@ -21,7 +21,7 @@ public static class MiddlewareExtensions
return builder.Use(next =>
{
var invokeMethod = MiddlewareHelpers.GetInvokeMethod(middleware);
var ctorParams = new[] { next }.Concat(args).Select(x => x!).ToArray();
var ctorParams = new[] { next }.Concat(args).Select(x => x).ToArray();
var instance = ActivatorUtilities.CreateInstance(builder.ApplicationServices, middleware, ctorParams);
return (NotificationMiddlewareDelegate)invokeMethod.CreateDelegate(typeof(NotificationMiddlewareDelegate), instance);
});

11
src/common/Elsa.Mediator/Middleware/Notification/NotificationContext.cs
View file

@ -12,11 +12,13 @@ public class NotificationContext
/// </summary>
/// <param name="notification">The notification to publish.</param>
/// <param name="notificationStrategy">The publishing strategy to use.</param>
/// <param name="serviceProvider">The service provider to resolve services from.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public NotificationContext(INotification notification, IEventPublishingStrategy notificationStrategy, CancellationToken cancellationToken = default)
public NotificationContext(INotification notification, IEventPublishingStrategy notificationStrategy, IServiceProvider serviceProvider, CancellationToken cancellationToken = default)
{
Notification = notification;
NotificationStrategy = notificationStrategy;
ServiceProvider = serviceProvider;
CancellationToken = cancellationToken;
}
@ -29,7 +31,12 @@ public class NotificationContext
/// Gets the publishing strategy to use.
/// </summary>
public IEventPublishingStrategy NotificationStrategy { get; init; }
/// <summary>
/// Gets the service provider used for resolving dependencies within the notification context.
/// </summary>
public IServiceProvider ServiceProvider { get; }
/// <summary>
/// Gets the cancellation token.
/// </summary>

30
src/common/Elsa.Mediator/Middleware/Request/RequestContext.cs
View file

@ -5,35 +5,27 @@ namespace Elsa.Mediator.Middleware.Request;
/// <summary>
/// Provides context to a request handler.
/// </summary>
public class RequestContext
public class RequestContext(IRequest request, Type responseType, IServiceProvider serviceProvider, CancellationToken cancellationToken)
{
/// <summary>
/// Initializes a new instance of the <see cref="RequestContext"/> class.
/// </summary>
/// <param name="request">The request.</param>
/// <param name="responseType">The response type.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public RequestContext(IRequest request, Type responseType, CancellationToken cancellationToken)
{
Request = request;
ResponseType = responseType;
CancellationToken = cancellationToken;
}
/// <summary>
/// Gets the request.
/// </summary>
public IRequest Request { get; init; }
public IRequest Request { get; init; } = request;
/// <summary>
/// Gets the response type.
/// </summary>
public Type ResponseType { get; init; }
public Type ResponseType { get; init; } = responseType;
/// <summary>
/// Gets the service provider used for resolving dependencies within the request context.
/// </summary>
public IServiceProvider ServiceProvider { get; } = serviceProvider;
/// <summary>
/// Gets the cancellation token.
/// </summary>
public CancellationToken CancellationToken { get; init; }
public CancellationToken CancellationToken { get; init; } = cancellationToken;
/// <summary>
/// Gets the response the request handler.

69
src/common/Elsa.Mediator/Services/DefaultMediator.cs
View file

@ -17,6 +17,7 @@ public class DefaultMediator : IMediator
private readonly IRequestPipeline _requestPipeline;
private readonly ICommandPipeline _commandPipeline;
private readonly INotificationPipeline _notificationPipeline;
private readonly IServiceProvider _serviceProvider;
private readonly IEventPublishingStrategy _defaultPublishingStrategy;
private readonly ICommandStrategy _defaultCommandStrategy;
@ -31,11 +32,13 @@ public class DefaultMediator : IMediator
IRequestPipeline requestPipeline,
ICommandPipeline commandPipeline,
INotificationPipeline notificationPipeline,
IOptions<MediatorOptions> options)
IOptions<MediatorOptions> options,
IServiceProvider serviceProvider)
{
_requestPipeline = requestPipeline;
_commandPipeline = commandPipeline;
_notificationPipeline = notificationPipeline;
_serviceProvider = serviceProvider;
_defaultPublishingStrategy = options.Value.DefaultPublishingStrategy;
_defaultCommandStrategy = options.Value.DefaultCommandStrategy;
}
@ -44,46 +47,66 @@ public class DefaultMediator : IMediator
public async Task<T> SendAsync<T>(IRequest<T> request, CancellationToken cancellationToken = default)
{
var responseType = typeof(T);
var context = new RequestContext(request, responseType, cancellationToken);
var context = new RequestContext(request, responseType, _serviceProvider, cancellationToken);
await _requestPipeline.ExecuteAsync(context);
return (T)context.Response;
}
/// <inheritdoc />
public async Task SendAsync(ICommand command, CancellationToken cancellationToken = default) => await SendAsync(command, _defaultCommandStrategy, cancellationToken);
/// <inheritdoc />
public async Task SendAsync(ICommand command, ICommandStrategy? strategy = null, CancellationToken cancellationToken = default)
{
var resultType = typeof(Unit);
strategy ??= _defaultCommandStrategy;
var context = new CommandContext(command, strategy, resultType, cancellationToken);
await _commandPipeline.InvokeAsync(context);
}
/// <inheritdoc />
public async Task<T> SendAsync<T>(ICommand<T> command, CancellationToken cancellationToken = default) => await SendAsync(command, _defaultCommandStrategy, cancellationToken);
/// <inheritdoc />
public async Task<T> SendAsync<T>(ICommand<T> command, ICommandStrategy? strategy, CancellationToken cancellationToken = default)
public async Task<T> SendAsync<T>(ICommand<T> command, ICommandStrategy strategy, IDictionary<object, object> headers, CancellationToken cancellationToken = default)
{
var resultType = typeof(T);
strategy ??= _defaultCommandStrategy;
var context = new CommandContext(command, strategy, resultType, cancellationToken);
var context = new CommandContext(command, strategy, resultType, headers, _serviceProvider, cancellationToken);
await _commandPipeline.InvokeAsync(context);
return (T)context.Result!;
}
/// <inheritdoc />
public async Task SendAsync(INotification notification, CancellationToken cancellationToken = default) => await SendAsync(notification, _defaultPublishingStrategy, cancellationToken);
public async Task SendAsync(ICommand command, CancellationToken cancellationToken = default) => await SendAsync(command, _defaultCommandStrategy, cancellationToken);
/// <inheritdoc />
public Task SendAsync(ICommand command, ICommandStrategy? strategy = null, CancellationToken cancellationToken = default)
{
return SendAsync(command, strategy, new Dictionary<object, object>(), cancellationToken);
}
public async Task SendAsync(ICommand command, ICommandStrategy? strategy, IDictionary<object, object> headers, CancellationToken cancellationToken = default)
{
var resultType = typeof(Unit);
strategy ??= _defaultCommandStrategy;
var context = new CommandContext(command, strategy, resultType, headers, _serviceProvider, cancellationToken);
await _commandPipeline.InvokeAsync(context);
}
/// <inheritdoc />
public Task<T> SendAsync<T>(ICommand<T> command, CancellationToken cancellationToken = default)
{
return SendAsync(command, new Dictionary<object, object>(), cancellationToken);
}
public Task<T> SendAsync<T>(ICommand<T> command, IDictionary<object, object> headers, CancellationToken cancellationToken = default)
{
return SendAsync(command, _defaultCommandStrategy, headers, cancellationToken);
}
/// <inheritdoc />
public Task<T> SendAsync<T>(ICommand<T> command, ICommandStrategy strategy, CancellationToken cancellationToken = default)
{
return SendAsync(command, strategy, new Dictionary<object, object>(), cancellationToken);
}
/// <inheritdoc />
public async Task SendAsync(INotification notification, CancellationToken cancellationToken = default)
{
await SendAsync(notification, _defaultPublishingStrategy, cancellationToken);
}
/// <inheritdoc />
public async Task SendAsync(INotification notification, IEventPublishingStrategy? strategy = null, CancellationToken cancellationToken = default)
{
strategy ??= _defaultPublishingStrategy;
var context = new NotificationContext(notification, strategy, cancellationToken);
var context = new NotificationContext(notification, strategy, _serviceProvider, cancellationToken);
await _notificationPipeline.ExecuteAsync(context);
}
}

6
src/modules/Elsa.Tenants/Features/TenantsFeature.cs
View file

@ -3,6 +3,7 @@ using Elsa.Common.Multitenancy;
using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.Tenants.Mediator.Tasks;
using Elsa.Tenants.Options;
using Elsa.Tenants.Providers;
using Microsoft.Extensions.DependencyInjection;
@ -45,6 +46,11 @@ public class TenantsFeature(IModule serviceConfiguration) : FeatureBase(serviceC
Module.Configure<MultitenancyFeature>(feature => feature.UseTenantsProvider<StoreTenantsProvider>());
}
public override void ConfigureHostedServices()
{
Module.ConfigureHostedService<SetupMediatorPipelines>();
}
/// <inheritdoc />
public override void Apply()
{

28
src/modules/Elsa.Tenants/Mediator/Middleware/TenantPropagatingMiddleware.cs Normal file
View file

@ -0,0 +1,28 @@
using Elsa.Common.Multitenancy;
using Elsa.Mediator.Middleware.Command;
using Elsa.Mediator.Middleware.Command.Contracts;
using JetBrains.Annotations;
namespace Elsa.Tenants.Mediator.Middleware;
/// <summary>
/// Middleware that ensures tenant context is propagated through the request pipeline.
/// </summary>
[UsedImplicitly]
public class TenantPropagatingMiddleware(CommandMiddlewareDelegate next, ITenantScopeFactory tenantScopeFactory, ITenantService tenantService) : ICommandMiddleware
{
/// <inheritdoc />
public async ValueTask InvokeAsync(CommandContext context)
{
if (context.Headers.TryGetValue(TenantHeaders.TenantIdKey, out var tenantIdVal))
{
var tenantId = (string)tenantIdVal;
var tenant = await tenantService.FindAsync(tenantId);
await using var tenantScope = tenantScopeFactory.CreateScope(tenant);
await next(context);
return;
}
await next(context);
}
}

19
src/modules/Elsa.Tenants/Mediator/Tasks/SetupMediatorPipelines.cs Normal file
View file

@ -0,0 +1,19 @@
using Elsa.Mediator.Middleware.Command;
using Elsa.Mediator.Middleware.Command.Contracts;
using Elsa.Tenants.Mediator.Middleware;
using JetBrains.Annotations;
using Microsoft.Extensions.Hosting;
namespace Elsa.Tenants.Mediator.Tasks;
[UsedImplicitly]
public class SetupMediatorPipelines(ICommandPipeline commandPipeline) : IHostedService
{
public Task StartAsync(CancellationToken cancellationToken)
{
commandPipeline.Setup(pipeline => pipeline.UseMiddleware<TenantPropagatingMiddleware>(0));
return Task.CompletedTask;
}
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}

13
src/modules/Elsa.Tenants/Mediator/TenantHeaders.cs Normal file
View file

@ -0,0 +1,13 @@
namespace Elsa.Tenants.Mediator;
public static class TenantHeaders
{
public static readonly object TenantIdKey = new();
public static IDictionary<object, object> CreateHeaders(string? tenantId)
{
var headers = new Dictionary<object, object>();
if (tenantId != null) headers.Add(TenantIdKey, tenantId);
return headers;
}
}

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

@ -1,5 +1,7 @@
using Elsa.Common.Multitenancy;
using Elsa.Mediator;
using Elsa.Mediator.Contracts;
using Elsa.Tenants.Mediator;
using Elsa.Workflows.Runtime.Commands;
using Elsa.Workflows.Runtime.Requests;
using Elsa.Workflows.Runtime.Responses;
@ -9,18 +11,8 @@ namespace Elsa.Workflows.Runtime;
/// <summary>
/// A simple implementation that queues the specified request for workflow execution on a non-durable background worker.
/// </summary>
public class BackgroundWorkflowDispatcher : IWorkflowDispatcher
public class BackgroundWorkflowDispatcher(ICommandSender commandSender, ITenantAccessor tenantAccessor) : IWorkflowDispatcher
{
private readonly ICommandSender _commandSender;
/// <summary>
/// Constructor.
/// </summary>
public BackgroundWorkflowDispatcher(ICommandSender commandSender)
{
_commandSender = commandSender;
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = null, CancellationToken cancellationToken = default)
{
@ -32,8 +24,8 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher
InstanceId = request.InstanceId,
TriggerActivityId = request.TriggerActivityId
};
await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken);
await commandSender.SendAsync(command, CommandStrategy.Background, CreateHeaders(), cancellationToken);
return DispatchWorkflowResponse.Success();
}
@ -47,7 +39,7 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher
Properties = request.Properties,
CorrelationId = request.CorrelationId};
await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken);
await commandSender.SendAsync(command, CommandStrategy.Background, CreateHeaders(), cancellationToken);
return DispatchWorkflowResponse.Success();
}
@ -62,7 +54,7 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher
Input = request.Input,
Properties = request.Properties
};
await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken);
await commandSender.SendAsync(command, CommandStrategy.Background, CreateHeaders(), cancellationToken);
return DispatchWorkflowResponse.Success();
}
@ -76,7 +68,12 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher
ActivityInstanceId = request.ActivityInstanceId,
Input = request.Input
};
await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken);
await commandSender.SendAsync(command, CommandStrategy.Background, CreateHeaders(), cancellationToken);
return DispatchWorkflowResponse.Success();
}
private IDictionary<object, object> CreateHeaders()
{
return TenantHeaders.CreateHeaders(tenantAccessor.Tenant?.Id);
}
}