diff --git a/Directory.Packages.props b/Directory.Packages.props index 3758682dd..0f9791acc 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -126,47 +126,47 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + - - + + - - - - + + + + \ No newline at end of file diff --git a/src/apps/Elsa.Server.Web/Program.cs b/src/apps/Elsa.Server.Web/Program.cs index 1f2617642..aee6fb0c9 100644 --- a/src/apps/Elsa.Server.Web/Program.cs +++ b/src/apps/Elsa.Server.Web/Program.cs @@ -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; diff --git a/src/apps/Elsa.Server.Web/appsettings.json b/src/apps/Elsa.Server.Web/appsettings.json index 91735a81b..a617ec190 100644 --- a/src/apps/Elsa.Server.Web/appsettings.json +++ b/src/apps/Elsa.Server.Web/appsettings.json @@ -44,7 +44,7 @@ }, { "Id": "tenant-1", - "Name": "Tenant 12 dd", + "Name": "Tenant 1", "Configuration": { "Http": { "Prefix": "/tenant-1", diff --git a/src/common/Elsa.Mediator/Channels/CommandsChannel.cs b/src/common/Elsa.Mediator/Channels/CommandsChannel.cs index 43a8b98b0..f80bbdbaa 100644 --- a/src/common/Elsa.Mediator/Channels/CommandsChannel.cs +++ b/src/common/Elsa.Mediator/Channels/CommandsChannel.cs @@ -1,9 +1,10 @@ using Elsa.Mediator.Abstractions; using Elsa.Mediator.Contracts; +using Elsa.Mediator.Middleware.Command; namespace Elsa.Mediator.Channels; /// -public class CommandsChannel : ChannelBase, ICommandsChannel +public class CommandsChannel : ChannelBase, ICommandsChannel { } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/CommandStrategies/BackgroundStrategy.cs b/src/common/Elsa.Mediator/CommandStrategies/BackgroundStrategy.cs index e99a0a1b0..8d4210e05 100644 --- a/src/common/Elsa.Mediator/CommandStrategies/BackgroundStrategy.cs +++ b/src/common/Elsa.Mediator/CommandStrategies/BackgroundStrategy.cs @@ -13,7 +13,7 @@ public class BackgroundStrategy : ICommandStrategy public async Task ExecuteAsync(CommandStrategyContext context) { var commandsChannel = context.ServiceProvider.GetRequiredService(); - await commandsChannel.Writer.WriteAsync(context.Command, context.CancellationToken); + await commandsChannel.Writer.WriteAsync(context.CommandContext, context.CancellationToken); return default!; } } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs b/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs index 9b204332f..a1e26aa79 100644 --- a/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs +++ b/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs @@ -12,7 +12,8 @@ public class DefaultStrategy : ICommandStrategy /// public async Task ExecuteAsync(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(); diff --git a/src/common/Elsa.Mediator/Contexts/CommandStrategyContext.cs b/src/common/Elsa.Mediator/Contexts/CommandStrategyContext.cs index d0e395063..9c4c9a640 100644 --- a/src/common/Elsa.Mediator/Contexts/CommandStrategyContext.cs +++ b/src/common/Elsa.Mediator/Contexts/CommandStrategyContext.cs @@ -1,12 +1,13 @@ using Elsa.Mediator.Contracts; +using Elsa.Mediator.Middleware.Command; namespace Elsa.Mediator.Contexts; /// /// Represents a context for executing a command. /// -/// The command to execute. +/// The command context to execute. /// The command handler. /// The service provider to resolve services from. /// The cancellation token. -public record CommandStrategyContext(ICommand Command, ICommandHandler Handler, IServiceProvider ServiceProvider, CancellationToken CancellationToken = default); \ No newline at end of file +public record CommandStrategyContext(CommandContext CommandContext, ICommandHandler Handler, IServiceProvider ServiceProvider, CancellationToken CancellationToken = default); \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Contracts/ICommandSender.cs b/src/common/Elsa.Mediator/Contracts/ICommandSender.cs index fb7a44b34..5373afb3b 100644 --- a/src/common/Elsa.Mediator/Contracts/ICommandSender.cs +++ b/src/common/Elsa.Mediator/Contracts/ICommandSender.cs @@ -6,13 +6,19 @@ namespace Elsa.Mediator.Contracts; public interface ICommandSender { /// - /// Sends a command using he default strategy. + /// Sends a command using the default strategy. + /// + Task SendAsync(ICommand command, CancellationToken cancellationToken = default); + + /// + /// Sends a command using the default strategy. /// /// The command to send. + /// Any headers to pass along. /// The cancellation token. /// The type of the result. /// The result. - Task SendAsync(ICommand command, CancellationToken cancellationToken = default); + Task SendAsync(ICommand command, IDictionary headers, CancellationToken cancellationToken = default); /// /// Sends a command using the specified strategy. @@ -24,6 +30,17 @@ public interface ICommandSender /// The result. Task SendAsync(ICommand command, ICommandStrategy strategy, CancellationToken cancellationToken = default); + /// + /// Sends a command using the specified strategy. + /// + /// The command to send. + /// The command strategy to use. + /// Any headers to pass along. + /// The cancellation token. + /// The type of the result. + /// The result. + Task SendAsync(ICommand command, ICommandStrategy strategy, IDictionary headers, CancellationToken cancellationToken = default); + /// /// Sends a command using the default strategy. /// @@ -37,5 +54,14 @@ public interface ICommandSender /// The command to send. /// The command strategy to use. /// The cancellation token. - Task SendAsync(ICommand command, ICommandStrategy? strategy, CancellationToken cancellationToken = default); + Task SendAsync(ICommand command, ICommandStrategy strategy, CancellationToken cancellationToken = default); + + /// + /// Sends a command using the specified strategy. + /// + /// The command to send. + /// The command strategy to use. + /// Any headers to pass along. + /// The cancellation token. + Task SendAsync(ICommand command, ICommandStrategy strategy, IDictionary headers, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Contracts/ICommandsChannel.cs b/src/common/Elsa.Mediator/Contracts/ICommandsChannel.cs index 1ce9b9b14..280df942d 100644 --- a/src/common/Elsa.Mediator/Contracts/ICommandsChannel.cs +++ b/src/common/Elsa.Mediator/Contracts/ICommandsChannel.cs @@ -1,4 +1,5 @@ using System.Threading.Channels; +using Elsa.Mediator.Middleware.Command; namespace Elsa.Mediator.Contracts; @@ -10,10 +11,10 @@ public interface ICommandsChannel /// /// Gets the writer for the commands queue. /// - ChannelWriter Writer { get; } + ChannelWriter Writer { get; } /// /// Gets the reader for the commands queue. /// - ChannelReader Reader { get; } + ChannelReader Reader { get; } } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Extensions/DependencyInjectionExtensions.cs b/src/common/Elsa.Mediator/Extensions/DependencyInjectionExtensions.cs index 2ac94c073..68ea123d8 100644 --- a/src/common/Elsa.Mediator/Extensions/DependencyInjectionExtensions.cs +++ b/src/common/Elsa.Mediator/Extensions/DependencyInjectionExtensions.cs @@ -36,9 +36,9 @@ public static class DependencyInjectionExtensions .AddScoped(sp => sp.GetRequiredService()) .AddScoped(sp => sp.GetRequiredService()) .AddScoped(sp => sp.GetRequiredService()) - .AddScoped() - .AddScoped() - .AddScoped() + .AddSingleton() + .AddSingleton() + .AddSingleton() ; } @@ -53,7 +53,7 @@ public static class DependencyInjectionExtensions .AddSingleton() .AddSingleton() .AddSingleton() - .AddSingleton() + .AddSingleton() .AddHostedService() .AddHostedService() .AddHostedService(); diff --git a/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs b/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs index 8b4a0db0c..47770bab5 100644 --- a/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs +++ b/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs @@ -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> _outputs; + private readonly List> _outputs; private readonly ILogger _logger; /// @@ -26,7 +27,7 @@ public class BackgroundCommandSenderHostedService : BackgroundService _commandsChannel = commandsChannel; _scopeFactory = scopeFactory; _logger = logger; - _outputs = new List>(_workerCount); + _outputs = new(_workerCount); } /// @@ -36,34 +37,32 @@ public class BackgroundCommandSenderHostedService : BackgroundService for (var i = 0; i < _workerCount; i++) { - var output = Channel.CreateUnbounded(); + var output = Channel.CreateUnbounded(); _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 output, CancellationToken cancellationToken) + private async Task ReadOutputAsync(Channel 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(); - await commandSender.SendAsync(command, CommandStrategy.Default, cancellationToken); + await commandSender.SendAsync(commandContext.Command, CommandStrategy.Default, commandContext.Headers, cancellationToken); } catch (Exception e) { diff --git a/src/common/Elsa.Mediator/Middleware/Command/CommandContext.cs b/src/common/Elsa.Mediator/Middleware/Command/CommandContext.cs index b7e61b836..f04384942 100644 --- a/src/common/Elsa.Mediator/Middleware/Command/CommandContext.cs +++ b/src/common/Elsa.Mediator/Middleware/Command/CommandContext.cs @@ -10,11 +10,13 @@ public class CommandContext /// /// Initializes a new instance of the class. /// - public CommandContext(ICommand command, ICommandStrategy commandStrategy, Type resultType, CancellationToken cancellationToken) + public CommandContext(ICommand command, ICommandStrategy commandStrategy, Type resultType, IDictionary headers, IServiceProvider serviceProvider, CancellationToken cancellationToken) { Command = command; CommandStrategy = commandStrategy; ResultType = resultType; + Headers = headers; + ServiceProvider = serviceProvider; CancellationToken = cancellationToken; } @@ -28,6 +30,16 @@ public class CommandContext /// public ICommandStrategy CommandStrategy { get; } + /// + /// Gets or sets the headers associated with the command context. + /// + public IDictionary Headers { get; } + + /// + /// Gets the service provider used to resolve services for the command context. + /// + public IServiceProvider ServiceProvider { get; } + /// /// Gets the result type. /// diff --git a/src/common/Elsa.Mediator/Middleware/Command/CommandPipeline.cs b/src/common/Elsa.Mediator/Middleware/Command/CommandPipeline.cs index f0e060c9b..3af4df116 100644 --- a/src/common/Elsa.Mediator/Middleware/Command/CommandPipeline.cs +++ b/src/common/Elsa.Mediator/Middleware/Command/CommandPipeline.cs @@ -5,28 +5,29 @@ namespace Elsa.Mediator.Middleware.Command; /// public class CommandPipeline : ICommandPipeline { - private readonly IServiceProvider _serviceProvider; - private CommandMiddlewareDelegate? _pipeline; + private readonly CommandPipelineBuilder _builder; + private CommandMiddlewareDelegate _pipeline = null!; /// /// Constructor. /// - public CommandPipeline(IServiceProvider serviceProvider) => _serviceProvider = serviceProvider; - - /// - public CommandMiddlewareDelegate Pipeline => _pipeline ??= CreateDefaultPipeline(); + public CommandPipeline(IServiceProvider serviceProvider) + { + _builder = new(serviceProvider); + Setup(x => x.UseCommandInvoker().UseCommandLogging()); + } /// - public CommandMiddlewareDelegate Setup(Action? setup = default) + public CommandMiddlewareDelegate Pipeline => _pipeline; + + /// + public CommandMiddlewareDelegate Setup(Action? setup = null) { - var builder = new CommandPipelineBuilder(_serviceProvider); - setup?.Invoke(builder); - _pipeline = builder.Build(); + setup?.Invoke(_builder); + _pipeline = _builder.Build(); return _pipeline; } /// public async Task InvokeAsync(CommandContext context) => await Pipeline(context); - - private CommandMiddlewareDelegate CreateDefaultPipeline() => Setup(x => x.UseCommandInvoker().UseCommandLogging()); } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Middleware/Command/CommandPipelineBuilder.cs b/src/common/Elsa.Mediator/Middleware/Command/CommandPipelineBuilder.cs index 8b98d49a3..94012e682 100644 --- a/src/common/Elsa.Mediator/Middleware/Command/CommandPipelineBuilder.cs +++ b/src/common/Elsa.Mediator/Middleware/Command/CommandPipelineBuilder.cs @@ -33,19 +33,45 @@ public class CommandPipelineBuilder : ICommandPipelineBuilder return this; } + /// + public ICommandPipelineBuilder Use(int index, Func middleware) + { + _components.Insert(index, middleware); + return this; + } + + /// + public ICommandPipelineBuilder Remove(Func middleware) + { + _components.Remove(middleware); + return this; + } + + /// + public ICommandPipelineBuilder RemoveAt(int index) + { + _components.RemoveAt(index); + return this; + } + + /// + public ICommandPipelineBuilder Clear() + { + _components.Clear(); + return this; + } + /// 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(string key) => Properties.TryGetValue(key, out var value) ? (T?)value : default(T); + private T? GetProperty(string key) => Properties.TryGetValue(key, out var value) ? (T?)value : default; private void SetProperty(string key, T value) => Properties[key] = value; } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Middleware/Command/Components/CommandHandlerInvokerMiddleware.cs b/src/common/Elsa.Mediator/Middleware/Command/Components/CommandHandlerInvokerMiddleware.cs index 7ad8487eb..47cceb0d9 100644 --- a/src/common/Elsa.Mediator/Middleware/Command/Components/CommandHandlerInvokerMiddleware.cs +++ b/src/common/Elsa.Mediator/Middleware/Command/Components/CommandHandlerInvokerMiddleware.cs @@ -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; /// /// A command middleware that invokes the command. /// -public class CommandHandlerInvokerMiddleware : ICommandMiddleware +[UsedImplicitly] +public class CommandHandlerInvokerMiddleware(CommandMiddlewareDelegate next) : ICommandMiddleware { - private readonly CommandMiddlewareDelegate _next; - private readonly IServiceProvider _serviceProvider; - private readonly IEnumerable _commandHandlers; - - /// - /// Constructor. - /// - public CommandHandlerInvokerMiddleware(CommandMiddlewareDelegate next, IEnumerable commandHandlers, IServiceProvider serviceProvider) - { - _next = next; - _serviceProvider = serviceProvider; - _commandHandlers = commandHandlers.DistinctBy(x => x.GetType()).ToList(); - } - /// 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(); + 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.Result))!; context.Result = resultProperty.GetValue(task); // Invoke next middleware. - await _next(context); + await next(context); } } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Middleware/Command/Components/CommandLoggingMiddleware.cs b/src/common/Elsa.Mediator/Middleware/Command/Components/CommandLoggingMiddleware.cs index a4a51b426..910f03f6c 100644 --- a/src/common/Elsa.Mediator/Middleware/Command/Components/CommandLoggingMiddleware.cs +++ b/src/common/Elsa.Mediator/Middleware/Command/Components/CommandLoggingMiddleware.cs @@ -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; /// /// A command middleware that logs the command being invoked. /// -public class CommandLoggingMiddleware : ICommandMiddleware +[UsedImplicitly] +public class CommandLoggingMiddleware(CommandMiddlewareDelegate next, ILogger logger) : ICommandMiddleware { - private readonly CommandMiddlewareDelegate _next; - private readonly ILogger _logger; - - /// - /// Constructor. - /// - public CommandLoggingMiddleware(CommandMiddlewareDelegate next, ILogger logger) - { - _next = next; - _logger = logger; - } - /// 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); } } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipeline.cs b/src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipeline.cs index a74048b18..b049f1c86 100644 --- a/src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipeline.cs +++ b/src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipeline.cs @@ -1,7 +1,7 @@ namespace Elsa.Mediator.Middleware.Command.Contracts; /// -/// +/// Represents a pipeline for processing commands. The pipeline is responsible for orchestrating the execution of registered middleware in sequence. /// public interface ICommandPipeline { diff --git a/src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipelineBuilder.cs b/src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipelineBuilder.cs index ca5c763cd..b6ac7c83d 100644 --- a/src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipelineBuilder.cs +++ b/src/common/Elsa.Mediator/Middleware/Command/Contracts/ICommandPipelineBuilder.cs @@ -16,11 +16,29 @@ public interface ICommandPipelineBuilder IServiceProvider ApplicationServices { get; } /// - /// Adds a middleware component to the pipeline. + /// Appends a middleware component to the pipeline. /// - /// The middleware component. - /// The pipeline builder. ICommandPipelineBuilder Use(Func middleware); + + /// + /// Adds a middleware component at the specified index. + /// + ICommandPipelineBuilder Use(int index, Func middleware); + + /// + /// Removes a middleware component from the pipeline. + /// + ICommandPipelineBuilder Remove(Func middleware); + + /// + /// Removes a middleware component at the specified index from the pipeline. + /// + ICommandPipelineBuilder RemoveAt(int index); + + /// + /// Clears the pipeline. + /// + ICommandPipelineBuilder Clear(); /// /// Builds the pipeline. diff --git a/src/common/Elsa.Mediator/Middleware/Command/MiddlewareExtensions.cs b/src/common/Elsa.Mediator/Middleware/Command/MiddlewareExtensions.cs index 7ab5e20f9..deac86de2 100644 --- a/src/common/Elsa.Mediator/Middleware/Command/MiddlewareExtensions.cs +++ b/src/common/Elsa.Mediator/Middleware/Command/MiddlewareExtensions.cs @@ -11,20 +11,35 @@ public static class MiddlewareExtensions /// /// Adds middleware to the pipeline. /// - /// The pipeline builder. - /// Any arguments to pass to the middleware constructor. - /// The middleware type. - /// The pipeline builder. public static ICommandPipelineBuilder UseMiddleware(this ICommandPipelineBuilder builder, params object[] args) where TMiddleware : ICommandMiddleware { - var middleware = typeof(TMiddleware); + return builder.Use(next => BuildMiddlewareDelegate(builder, next, args)); + } - return builder.Use(next => + /// + /// Inserts middleware at a specific index in the pipeline. + /// + public static ICommandPipelineBuilder UseMiddleware(this ICommandPipelineBuilder builder, int index, params object[] args) where TMiddleware : ICommandMiddleware + { + return builder.Use(index, next => BuildMiddlewareDelegate(builder, next, args)); + } + + /// + /// Builds a delegate for the middleware type. + /// + private static CommandMiddlewareDelegate BuildMiddlewareDelegate( + 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); } } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs b/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs index c555a7f5f..874c64f17 100644 --- a/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs +++ b/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs @@ -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; /// -public class NotificationHandlerInvokerMiddleware : INotificationMiddleware +[UsedImplicitly] +public class NotificationHandlerInvokerMiddleware( + NotificationMiddlewareDelegate next, + ILogger logger) + : INotificationMiddleware { - private readonly NotificationMiddlewareDelegate _next; - private readonly ILogger _logger; - private readonly IServiceProvider _serviceProvider; - private readonly IEnumerable _notificationHandlers; - - /// - /// Initializes a new instance of the class. - /// - public NotificationHandlerInvokerMiddleware( - NotificationMiddlewareDelegate next, - ILogger logger, - IServiceProvider serviceProvider, - IEnumerable notificationHandlers) - { - _next = next; - _logger = logger; - _serviceProvider = serviceProvider; - _notificationHandlers = notificationHandlers; - } - /// 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(); + 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); } } diff --git a/src/common/Elsa.Mediator/Middleware/Notification/MiddlewareExtensions.cs b/src/common/Elsa.Mediator/Middleware/Notification/MiddlewareExtensions.cs index c45fc9f27..5be932c94 100644 --- a/src/common/Elsa.Mediator/Middleware/Notification/MiddlewareExtensions.cs +++ b/src/common/Elsa.Mediator/Middleware/Notification/MiddlewareExtensions.cs @@ -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); }); diff --git a/src/common/Elsa.Mediator/Middleware/Notification/NotificationContext.cs b/src/common/Elsa.Mediator/Middleware/Notification/NotificationContext.cs index 60a6dd1e4..4f4f63048 100644 --- a/src/common/Elsa.Mediator/Middleware/Notification/NotificationContext.cs +++ b/src/common/Elsa.Mediator/Middleware/Notification/NotificationContext.cs @@ -12,11 +12,13 @@ public class NotificationContext /// /// The notification to publish. /// The publishing strategy to use. + /// The service provider to resolve services from. /// The cancellation token. - 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. /// public IEventPublishingStrategy NotificationStrategy { get; init; } - + + /// + /// Gets the service provider used for resolving dependencies within the notification context. + /// + public IServiceProvider ServiceProvider { get; } + /// /// Gets the cancellation token. /// diff --git a/src/common/Elsa.Mediator/Middleware/Request/RequestContext.cs b/src/common/Elsa.Mediator/Middleware/Request/RequestContext.cs index b6643a567..422c0e102 100644 --- a/src/common/Elsa.Mediator/Middleware/Request/RequestContext.cs +++ b/src/common/Elsa.Mediator/Middleware/Request/RequestContext.cs @@ -5,35 +5,27 @@ namespace Elsa.Mediator.Middleware.Request; /// /// Provides context to a request handler. /// -public class RequestContext +public class RequestContext(IRequest request, Type responseType, IServiceProvider serviceProvider, CancellationToken cancellationToken) { - /// - /// Initializes a new instance of the class. - /// - /// The request. - /// The response type. - /// The cancellation token. - public RequestContext(IRequest request, Type responseType, CancellationToken cancellationToken) - { - Request = request; - ResponseType = responseType; - CancellationToken = cancellationToken; - } - /// /// Gets the request. /// - public IRequest Request { get; init; } - + public IRequest Request { get; init; } = request; + /// /// Gets the response type. /// - public Type ResponseType { get; init; } - + public Type ResponseType { get; init; } = responseType; + + /// + /// Gets the service provider used for resolving dependencies within the request context. + /// + public IServiceProvider ServiceProvider { get; } = serviceProvider; + /// /// Gets the cancellation token. /// - public CancellationToken CancellationToken { get; init; } + public CancellationToken CancellationToken { get; init; } = cancellationToken; /// /// Gets the response the request handler. diff --git a/src/common/Elsa.Mediator/Services/DefaultMediator.cs b/src/common/Elsa.Mediator/Services/DefaultMediator.cs index 23a68162b..40587c3b7 100644 --- a/src/common/Elsa.Mediator/Services/DefaultMediator.cs +++ b/src/common/Elsa.Mediator/Services/DefaultMediator.cs @@ -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 options) + IOptions 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 SendAsync(IRequest 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; } - /// - public async Task SendAsync(ICommand command, CancellationToken cancellationToken = default) => await SendAsync(command, _defaultCommandStrategy, cancellationToken); - - /// - 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); - } - - /// - public async Task SendAsync(ICommand command, CancellationToken cancellationToken = default) => await SendAsync(command, _defaultCommandStrategy, cancellationToken); - - /// - public async Task SendAsync(ICommand command, ICommandStrategy? strategy, CancellationToken cancellationToken = default) + public async Task SendAsync(ICommand command, ICommandStrategy strategy, IDictionary 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!; } /// - 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); + + /// + public Task SendAsync(ICommand command, ICommandStrategy? strategy = null, CancellationToken cancellationToken = default) + { + return SendAsync(command, strategy, new Dictionary(), cancellationToken); + } + + public async Task SendAsync(ICommand command, ICommandStrategy? strategy, IDictionary headers, CancellationToken cancellationToken = default) + { + var resultType = typeof(Unit); + strategy ??= _defaultCommandStrategy; + var context = new CommandContext(command, strategy, resultType, headers, _serviceProvider, cancellationToken); + await _commandPipeline.InvokeAsync(context); + } + + /// + public Task SendAsync(ICommand command, CancellationToken cancellationToken = default) + { + return SendAsync(command, new Dictionary(), cancellationToken); + } + + public Task SendAsync(ICommand command, IDictionary headers, CancellationToken cancellationToken = default) + { + return SendAsync(command, _defaultCommandStrategy, headers, cancellationToken); + } + + /// + public Task SendAsync(ICommand command, ICommandStrategy strategy, CancellationToken cancellationToken = default) + { + return SendAsync(command, strategy, new Dictionary(), cancellationToken); + } + + /// + public async Task SendAsync(INotification notification, CancellationToken cancellationToken = default) + { + await SendAsync(notification, _defaultPublishingStrategy, cancellationToken); + } /// 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); } } \ No newline at end of file diff --git a/src/modules/Elsa.Tenants/Features/TenantsFeature.cs b/src/modules/Elsa.Tenants/Features/TenantsFeature.cs index ce4a18a9a..e19e248b7 100644 --- a/src/modules/Elsa.Tenants/Features/TenantsFeature.cs +++ b/src/modules/Elsa.Tenants/Features/TenantsFeature.cs @@ -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(feature => feature.UseTenantsProvider()); } + public override void ConfigureHostedServices() + { + Module.ConfigureHostedService(); + } + /// public override void Apply() { diff --git a/src/modules/Elsa.Tenants/Mediator/Middleware/TenantPropagatingMiddleware.cs b/src/modules/Elsa.Tenants/Mediator/Middleware/TenantPropagatingMiddleware.cs new file mode 100644 index 000000000..ad18be2a8 --- /dev/null +++ b/src/modules/Elsa.Tenants/Mediator/Middleware/TenantPropagatingMiddleware.cs @@ -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; + +/// +/// Middleware that ensures tenant context is propagated through the request pipeline. +/// +[UsedImplicitly] +public class TenantPropagatingMiddleware(CommandMiddlewareDelegate next, ITenantScopeFactory tenantScopeFactory, ITenantService tenantService) : ICommandMiddleware +{ + /// + 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); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Tenants/Mediator/Tasks/SetupMediatorPipelines.cs b/src/modules/Elsa.Tenants/Mediator/Tasks/SetupMediatorPipelines.cs new file mode 100644 index 000000000..33613f09a --- /dev/null +++ b/src/modules/Elsa.Tenants/Mediator/Tasks/SetupMediatorPipelines.cs @@ -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(0)); + return Task.CompletedTask; + } + + public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; +} \ No newline at end of file diff --git a/src/modules/Elsa.Tenants/Mediator/TenantHeaders.cs b/src/modules/Elsa.Tenants/Mediator/TenantHeaders.cs new file mode 100644 index 000000000..a6aa66334 --- /dev/null +++ b/src/modules/Elsa.Tenants/Mediator/TenantHeaders.cs @@ -0,0 +1,13 @@ +namespace Elsa.Tenants.Mediator; + +public static class TenantHeaders +{ + public static readonly object TenantIdKey = new(); + + public static IDictionary CreateHeaders(string? tenantId) + { + var headers = new Dictionary(); + if (tenantId != null) headers.Add(TenantIdKey, tenantId); + return headers; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs index 09e889628..d0279afcb 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs @@ -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; /// /// A simple implementation that queues the specified request for workflow execution on a non-durable background worker. /// -public class BackgroundWorkflowDispatcher : IWorkflowDispatcher +public class BackgroundWorkflowDispatcher(ICommandSender commandSender, ITenantAccessor tenantAccessor) : IWorkflowDispatcher { - private readonly ICommandSender _commandSender; - - /// - /// Constructor. - /// - public BackgroundWorkflowDispatcher(ICommandSender commandSender) - { - _commandSender = commandSender; - } - /// public async Task 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 CreateHeaders() + { + return TenantHeaders.CreateHeaders(tenantAccessor.Tenant?.Id); + } } \ No newline at end of file