diff --git a/Elsa.sln b/Elsa.sln
index d23fb3914..642fbb80c 100644
--- a/Elsa.sln
+++ b/Elsa.sln
@@ -222,6 +222,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "adr", "adr", "{0A04B1FD-06C
doc\adr\0004-token-centric-flowchart-execution-model.md = doc\adr\0004-token-centric-flowchart-execution-model.md
doc\adr\graph.dot = doc\adr\graph.dot
doc\adr\toc.md = doc\adr\toc.md
+ doc\adr\0005-activity-execution-snapshots.md = doc\adr\0004-activity-execution-snapshots.md
+ doc\adr\0006-tenant-deleted-event.md = doc\adr\0005-tenant-deleted-event.md
EndProjectSection
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "bounty", "bounty", "{9B80A705-2E31-4012-964A-83963DCDB384}"
diff --git a/doc/adr/0005-tenant-deleted-event.md b/doc/adr/0005-tenant-deleted-event.md
new file mode 100644
index 000000000..d0d2fa9f1
--- /dev/null
+++ b/doc/adr/0005-tenant-deleted-event.md
@@ -0,0 +1,23 @@
+# 5. Tenant Deleted Event
+
+Date: 2025-08-05
+
+## Status
+
+Accepted
+
+## Context
+
+As outlined in issue [#6661](https://github.com/elsa-workflows/elsa-core/issues/6661), there is a need to differentiate between **Tenant Deactivating** and **Tenant Deleting** events.
+
+Currently, the `TenantDeactivated` event is used to unregister timer-based triggers. However, this causes the triggers to be unregistered when the application host shuts down, which is not the desired behavior. Instead, we want the triggers to remain registered until the tenant is explicitly deleted.
+
+## Decision
+
+To address this, we will introduce a new event called `TenantDeleted`. This event will be raised when a tenant is deleted and will be responsible for unregistering timer-based triggers. This ensures that the triggers remain active until the tenant is explicitly deleted.
+
+## Consequences
+
+- Timer-based triggers will no longer be unregistered during tenant deactivation. Instead, they will remain active until the tenant is deleted.
+- The `TenantDeactivated` event will continue to be used for deactivating tenants without affecting the registration of timer-based triggers.
+- The new `TenantDeleted` event will specifically handle the cleanup of resources associated with a tenant when it is deleted, ensuring a clear separation of responsibilities.
diff --git a/doc/adr/graph.dot b/doc/adr/graph.dot
index 61e1e6f7c..4818df2a2 100644
--- a/doc/adr/graph.dot
+++ b/doc/adr/graph.dot
@@ -1,12 +1,14 @@
digraph {
-node [shape = plaintext];
+node [shape=plaintext];
subgraph {
-_1 [label = "1. Record architecture decisions"; URL = "0001-record-architecture-decisions.html"];
-_2 [label = "2. Fault Propagation from Child to Parent Activities"; URL ="0002-fault-propagation-from-child-to-parent-activities.html"];
-_1 -> _2 [style= "dotted", weight = 1];
-_3 [label = "3. Direct Bookmark Management in WorkflowExecutionContext"; URL = "0003-direct-bookmark-management-in-workflowexecutioncontext.html"];
-_2 -> _3 [style = "dotted", weight = 1];
-_4 [label ="4. Activity Execution Snapshots"; URL = "0004-activity-execution-snapshots.html"];
-_3 -> _4 [style = "dotted", weight = 1];
+_1 [label="1. Record architecture decisions"; URL="0001-record-architecture-decisions.html"];
+_2 [label="2. Fault Propagation from Child to Parent Activities"; URL="0002-fault-propagation-from-child-to-parent-activities.html"];
+_1 -> _2 [style="dotted", weight=1];
+_3 [label="3. Direct Bookmark Management in WorkflowExecutionContext"; URL="0003-direct-bookmark-management-in-workflowexecutioncontext.html"];
+_2 -> _3 [style="dotted", weight=1];
+_4 [label="4. Activity Execution Snapshots"; URL="0004-activity-execution-snapshots.html"];
+_3 -> _4 [style="dotted", weight=1];
+_5 [label="5. Tenant Deleted Event"; URL="0005-tenant-deleted-event.html"];
+_4 -> _5 [style="dotted", weight=1];
}
}
\ No newline at end of file
diff --git a/doc/adr/toc.md b/doc/adr/toc.md
index 7eecf8af6..26b15683d 100644
--- a/doc/adr/toc.md
+++ b/doc/adr/toc.md
@@ -3,4 +3,5 @@
* [1. Record architecture decisions](0001-record-architecture-decisions.md)
* [2. Fault Propagation from Child to Parent Activities](0002-fault-propagation-from-child-to-parent-activities.md)
* [3. Direct Bookmark Management in WorkflowExecutionContext](0003-direct-bookmark-management-in-workflowexecutioncontext.md)
-* [4. Activity Execution Snapshots](0004-activity-execution-snapshots.md)
\ No newline at end of file
+* [4. Activity Execution Snapshots](0004-activity-execution-snapshots.md)
+* [5. Tenant Deleted Event](0005-tenant-deleted-event.md)
\ No newline at end of file
diff --git a/src/apps/Elsa.Server.Web/appsettings.json b/src/apps/Elsa.Server.Web/appsettings.json
index 2f9e6b48c..b0bd725e6 100644
--- a/src/apps/Elsa.Server.Web/appsettings.json
+++ b/src/apps/Elsa.Server.Web/appsettings.json
@@ -29,7 +29,7 @@
"Name": "Tenant 1",
"Configuration": {
"Http": {
- "Prefix": "/tenant-1",
+ "Prefix": "tenant-1",
"Host": "localhost:5001"
},
"ConnectionStrings": {
diff --git a/src/common/Elsa.Mediator/Channels/NotificationsChannel.cs b/src/common/Elsa.Mediator/Channels/NotificationsChannel.cs
index 5480c3a60..676673630 100644
--- a/src/common/Elsa.Mediator/Channels/NotificationsChannel.cs
+++ b/src/common/Elsa.Mediator/Channels/NotificationsChannel.cs
@@ -1,9 +1,10 @@
using Elsa.Mediator.Abstractions;
using Elsa.Mediator.Contracts;
+using Elsa.Mediator.Middleware.Notification;
namespace Elsa.Mediator.Channels;
///
-public class NotificationsChannel : ChannelBase, INotificationsChannel
+public class NotificationsChannel : ChannelBase, INotificationsChannel
{
}
\ 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 a1e26aa79..12bf6e2a0 100644
--- a/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs
+++ b/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs
@@ -14,11 +14,10 @@ public class DefaultStrategy : ICommandStrategy
{
var commandContext = context.CommandContext;
var command = commandContext.Command;
- var cancellationToken = context.CancellationToken;
var commandType = command.GetType();
var handleMethod = commandType.GetCommandHandlerMethod();
var handler = context.Handler;
- return await handler.InvokeAsync(handleMethod, command, cancellationToken);
+ return await handler.InvokeAsync(handleMethod, commandContext);
}
}
\ No newline at end of file
diff --git a/src/common/Elsa.Mediator/Contexts/NotificationStrategyContext.cs b/src/common/Elsa.Mediator/Contexts/NotificationStrategyContext.cs
index b07339f82..7872e1a83 100644
--- a/src/common/Elsa.Mediator/Contexts/NotificationStrategyContext.cs
+++ b/src/common/Elsa.Mediator/Contexts/NotificationStrategyContext.cs
@@ -1,4 +1,5 @@
using Elsa.Mediator.Contracts;
+using Elsa.Mediator.Middleware.Notification;
using Microsoft.Extensions.Logging;
namespace Elsa.Mediator.Contexts;
@@ -6,9 +7,9 @@ namespace Elsa.Mediator.Contexts;
///
/// Represents a context for publishing events.
///
-/// The notification to publish.
+/// The notification to publish.
/// The handlers to publish the notification to.
/// The logger.
/// The service provider to resolve services from.
/// The cancellation token.
-public record NotificationStrategyContext(INotification Notification, INotificationHandler[] Handlers, ILogger Logger, IServiceProvider ServiceProvider, CancellationToken CancellationToken = default);
\ No newline at end of file
+public record NotificationStrategyContext(NotificationContext NotificationContext, INotificationHandler[] Handlers, ILogger Logger, IServiceProvider ServiceProvider, CancellationToken CancellationToken = default);
\ No newline at end of file
diff --git a/src/common/Elsa.Mediator/Contracts/INotificationsChannel.cs b/src/common/Elsa.Mediator/Contracts/INotificationsChannel.cs
index 7ee0240f7..59f25449a 100644
--- a/src/common/Elsa.Mediator/Contracts/INotificationsChannel.cs
+++ b/src/common/Elsa.Mediator/Contracts/INotificationsChannel.cs
@@ -1,4 +1,5 @@
using System.Threading.Channels;
+using Elsa.Mediator.Middleware.Notification;
namespace Elsa.Mediator.Contracts;
@@ -8,12 +9,12 @@ namespace Elsa.Mediator.Contracts;
public interface INotificationsChannel
{
///
- /// Gets the writer for the notifications queue.
+ /// Gets the writer for the notification queue.
///
- ChannelWriter Writer { get; }
+ ChannelWriter Writer { get; }
///
- /// Gets the reader for the notifications queue.
+ /// Gets the reader for the notification queue.
///
- ChannelReader Reader { get; }
+ ChannelReader Reader { get; }
}
\ No newline at end of file
diff --git a/src/common/Elsa.Mediator/Extensions/HandlerExtensions.cs b/src/common/Elsa.Mediator/Extensions/HandlerExtensions.cs
index cc5b55fd7..5ac381cb0 100644
--- a/src/common/Elsa.Mediator/Extensions/HandlerExtensions.cs
+++ b/src/common/Elsa.Mediator/Extensions/HandlerExtensions.cs
@@ -1,6 +1,8 @@
using System.Diagnostics.CodeAnalysis;
using System.Reflection;
using Elsa.Mediator.Contracts;
+using Elsa.Mediator.Middleware.Command;
+using Elsa.Mediator.Middleware.Notification;
using Elsa.Mediator.Models;
namespace Elsa.Mediator.Extensions;
@@ -40,10 +42,12 @@ public static class HandlerExtensions
///
/// The handler to invoke.
/// The handle method.
- /// The notification to handle.
+ /// The notification to handle.
/// The cancellation token.
- public static Task InvokeAsync(this INotificationHandler handler, MethodBase handleMethod, INotification notification, CancellationToken cancellationToken)
+ public static Task InvokeAsync(this INotificationHandler handler, MethodBase handleMethod, NotificationContext notificationContext)
{
+ var notification = notificationContext.Notification;
+ var cancellationToken = notificationContext.CancellationToken;
return (Task)handleMethod.Invoke(handler, [notification, cancellationToken])!;
}
@@ -52,10 +56,11 @@ public static class HandlerExtensions
///
/// The handler to invoke.
/// The handle method.
- /// The command to handle.
- /// The cancellation token.
- public static Task InvokeAsync(this ICommandHandler handler, MethodBase handleMethod, ICommand command, CancellationToken cancellationToken)
+ /// The command to handle.
+ public static Task InvokeAsync(this ICommandHandler handler, MethodBase handleMethod, CommandContext commandContext)
{
+ var command = commandContext.Command;
+ var cancellationToken = commandContext.CancellationToken;
var task = (Task)handleMethod.Invoke(handler, [command, cancellationToken])!;
return task;
}
diff --git a/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs b/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs
index 419231537..f0606a1f5 100644
--- a/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs
+++ b/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs
@@ -24,48 +24,66 @@ public class BackgroundCommandSenderHostedService : BackgroundService
public BackgroundCommandSenderHostedService(IOptions options, ICommandsChannel commandsChannel, IServiceScopeFactory scopeFactory, ILogger logger)
{
_workerCount = options.Value.CommandWorkerCount;
- _commandsChannel = commandsChannel;
+ _commandsChannel = commandsChannel; // The shared input channel for all commands
_scopeFactory = scopeFactory;
_logger = logger;
- _outputs = new(_workerCount);
+ _outputs = new(_workerCount); // Prepare a list to hold worker-specific channels
}
///
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
{
- var index = 0;
+ var index = 0; // Used for round-robin distribution of work
+ // Set up worker channels and start background tasks for each worker
for (var i = 0; i < _workerCount; i++)
{
var output = Channel.CreateUnbounded();
_outputs.Add(output);
+ // Start a background task that processes commands from this worker's channel
_ = ReadOutputAsync(output, cancellationToken);
}
+ // Main dispatcher loop: read from the input channel and distribute to worker channels
await foreach (var commandContext in _commandsChannel.Reader.ReadAllAsync(cancellationToken))
{
var output = _outputs[index];
await output.Writer.WriteAsync(commandContext, cancellationToken);
+ // Round-robin distribution - move to next worker
index = (index + 1) % _workerCount;
}
- foreach (var output in _outputs)
+ // If the input channel is completed, complete all worker channels
+ foreach (var output in _outputs)
output.Writer.Complete();
}
private async Task ReadOutputAsync(Channel output, CancellationToken cancellationToken)
{
+ // Worker task: process commands from the worker's channel
await foreach (var commandContext in output.Reader.ReadAllAsync(cancellationToken))
{
try
{
+ // Create a fresh scope for each command to ensure proper service lifetime
using var scope = _scopeFactory.CreateScope();
var commandSender = scope.ServiceProvider.GetRequiredService();
- await commandSender.SendAsync(commandContext.Command, CommandStrategy.Default, commandContext.Headers, cancellationToken);
+ // Link the service cancellation token with the command's token to ensure proper cancellation
+ using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
+ cancellationToken,
+ commandContext.CancellationToken);
+
+ // Process the command using the command sender service with the linked token
+ await commandSender.SendAsync(
+ commandContext.Command,
+ CommandStrategy.Default,
+ commandContext.Headers,
+ linkedTokenSource.Token);
}
catch (Exception e)
{
+ // Log errors but continue processing other commands
_logger.LogError(e, "An unhandled exception occurred while processing the queue");
}
}
diff --git a/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs b/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs
index 3a10e8403..00c29bba1 100644
--- a/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs
+++ b/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs
@@ -1,5 +1,6 @@
using System.Threading.Channels;
using Elsa.Mediator.Contracts;
+using Elsa.Mediator.Middleware.Notification;
using Elsa.Mediator.Options;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
@@ -16,7 +17,7 @@ public class BackgroundEventPublisherHostedService : BackgroundService
private readonly int _workerCount;
private readonly INotificationsChannel _notificationsChannel;
private readonly IServiceScopeFactory _scopeFactory;
- private readonly List> _outputs;
+ private readonly List> _outputs;
private readonly ILogger _logger;
///
@@ -26,46 +27,62 @@ public class BackgroundEventPublisherHostedService : BackgroundService
_notificationsChannel = notificationsChannel;
_scopeFactory = scopeFactory;
_logger = logger;
- _outputs = new List>(_workerCount);
+ _outputs = new(_workerCount);
}
///
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
{
+ // Index to round-robin distribute notifications across worker channels
var index = 0;
using var scope = _scopeFactory.CreateScope();
var notificationSender = scope.ServiceProvider.GetRequiredService();
+ // Create multiple output channels and start worker tasks for parallel processing
for (var i = 0; i < _workerCount; i++)
{
- var output = Channel.CreateUnbounded();
+ var output = Channel.CreateUnbounded();
_outputs.Add(output);
+ // Start a background task to process notifications from this output channel
_ = ReadOutputAsync(output, notificationSender, cancellationToken);
}
var channelReader = _notificationsChannel.Reader;
+ // Continuously read notifications from the input channel and distribute them to worker channels
+ // using round-robin distribution for load balancing
await foreach (var notification in channelReader.ReadAllAsync(cancellationToken))
{
var output = _outputs[index];
await output.Writer.WriteAsync(notification, cancellationToken);
+ // Move to the next worker in a circular fashion
index = (index + 1) % _workerCount;
}
+ // When the input channel is completed, complete all output channels
foreach (var output in _outputs)
{
output.Writer.Complete();
}
}
- private async Task ReadOutputAsync(Channel output, INotificationSender notificationSender, CancellationToken cancellationToken)
+ ///
+ /// Processes notifications from an output channel asynchronously.
+ ///
+ /// The channel to read notifications from
+ /// The service used to send notifications
+ /// Cancellation token from the hosted service
+ private async Task ReadOutputAsync(Channel output, INotificationSender notificationSender, CancellationToken cancellationToken)
{
- await foreach (var notification in output.Reader.ReadAllAsync(cancellationToken))
+ await foreach (var notificationContext in output.Reader.ReadAllAsync(cancellationToken))
{
try
{
- await notificationSender.SendAsync(notification, NotificationStrategy.Sequential, cancellationToken);
+ var notification = notificationContext.Notification;
+ // Link the cancellation tokens so that cancellation can happen from either source
+ using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, notificationContext.CancellationToken);
+ await notificationSender.SendAsync(notification, NotificationStrategy.Sequential, linkedTokenSource.Token);
}
catch (OperationCanceledException e)
{
diff --git a/src/common/Elsa.Mediator/HostedServices/JobRunnerHostedService.cs b/src/common/Elsa.Mediator/HostedServices/JobRunnerHostedService.cs
index 4eeec9061..14af18b4b 100644
--- a/src/common/Elsa.Mediator/HostedServices/JobRunnerHostedService.cs
+++ b/src/common/Elsa.Mediator/HostedServices/JobRunnerHostedService.cs
@@ -17,7 +17,7 @@ public class JobRunnerHostedService : BackgroundService
///
public JobRunnerHostedService(IOptions options, IJobsChannel jobsChannel, ILogger logger)
- {
+ {
_workerCount = options.Value.JobWorkerCount;
_jobsChannel = jobsChannel;
_logger = logger;
@@ -26,21 +26,31 @@ public class JobRunnerHostedService : BackgroundService
///
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
- var workers = new Task[_workerCount];
-
- for (var i = 0; i < _workerCount; i++)
+ // Create an array of worker tasks to process jobs in parallel
+ var workers = new Task[_workerCount];
+
+ // Start multiple workers (tasks) to process jobs concurrently
+ for (var i = 0; i < _workerCount; i++)
workers[i] = ProcessJobsAsync(stoppingToken);
+ // Wait for all worker tasks to complete (typically when the application is shutting down)
await Task.WhenAll(workers);
}
+ ///
+ /// Continuously processes jobs from the job channel until cancellation is requested.
+ ///
+ /// Cancellation token from the hosted service
private async Task ProcessJobsAsync(CancellationToken stoppingToken)
{
+ // Process all jobs from the channel until it's completed or cancellation is requested
await foreach (var jobItem in _jobsChannel.Reader.ReadAllAsync(stoppingToken))
{
try
{
- await jobItem.Action(jobItem.CancellationTokenSource.Token);
+ // Link the cancellation tokens so that cancellation can happen from either source
+ using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken, jobItem.CancellationTokenSource.Token);
+ await jobItem.Action(linkedTokenSource.Token);
_logger.LogInformation("Worker {CurrentTaskId} processed job {JobId}", Task.CurrentId, jobItem.JobId);
}
catch (OperationCanceledException)
@@ -53,6 +63,7 @@ public class JobRunnerHostedService : BackgroundService
}
finally
{
+ // Notify that the job has completed (whether successfully or not)
jobItem.OnJobCompleted(jobItem.JobId);
}
}
diff --git a/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs b/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs
index 874c64f17..e7e22125c 100644
--- a/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs
+++ b/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs
@@ -24,7 +24,7 @@ public class NotificationHandlerInvokerMiddleware(
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);
+ var strategyContext = new NotificationStrategyContext(context, handlers, logger, serviceProvider, context.CancellationToken);
await context.NotificationStrategy.PublishAsync(strategyContext);
diff --git a/src/common/Elsa.Mediator/PublishingStrategies/BackgroundProcessingStrategy.cs b/src/common/Elsa.Mediator/PublishingStrategies/BackgroundProcessingStrategy.cs
index b228c24a2..e960050f5 100644
--- a/src/common/Elsa.Mediator/PublishingStrategies/BackgroundProcessingStrategy.cs
+++ b/src/common/Elsa.Mediator/PublishingStrategies/BackgroundProcessingStrategy.cs
@@ -14,6 +14,6 @@ public class BackgroundProcessingStrategy : IEventPublishingStrategy
{
var notificationsChannel = context.ServiceProvider.GetRequiredService();
- await notificationsChannel.Writer.WriteAsync(context.Notification, context.CancellationToken);
+ await notificationsChannel.Writer.WriteAsync(context.NotificationContext, context.CancellationToken);
}
}
\ No newline at end of file
diff --git a/src/common/Elsa.Mediator/PublishingStrategies/ParallelProcessingStrategy.cs b/src/common/Elsa.Mediator/PublishingStrategies/ParallelProcessingStrategy.cs
index fd16c8436..8eec00643 100644
--- a/src/common/Elsa.Mediator/PublishingStrategies/ParallelProcessingStrategy.cs
+++ b/src/common/Elsa.Mediator/PublishingStrategies/ParallelProcessingStrategy.cs
@@ -12,11 +12,11 @@ public class ParallelProcessingStrategy : IEventPublishingStrategy
///
public async Task PublishAsync(NotificationStrategyContext context)
{
- var notification = context.Notification;
- var cancellationToken = context.CancellationToken;
+ var notificationContext = context.NotificationContext;
+ var notification = notificationContext.Notification;
var notificationType = notification.GetType();
var handleMethod = notificationType.GetNotificationHandlerMethod();
- var tasks = context.Handlers.Select(handler => handler.InvokeAsync(handleMethod, notification, cancellationToken)).ToList();
+ var tasks = context.Handlers.Select(handler => handler.InvokeAsync(handleMethod, context.NotificationContext)).ToList();
await Task.WhenAll(tasks);
}
diff --git a/src/common/Elsa.Mediator/PublishingStrategies/SequentialProcessingStrategy.cs b/src/common/Elsa.Mediator/PublishingStrategies/SequentialProcessingStrategy.cs
index ee7510445..494b9de2b 100644
--- a/src/common/Elsa.Mediator/PublishingStrategies/SequentialProcessingStrategy.cs
+++ b/src/common/Elsa.Mediator/PublishingStrategies/SequentialProcessingStrategy.cs
@@ -12,12 +12,12 @@ public class SequentialProcessingStrategy : IEventPublishingStrategy
///
public async Task PublishAsync(NotificationStrategyContext context)
{
- var notification = context.Notification;
- var cancellationToken = context.CancellationToken;
+ var notificationContext = context.NotificationContext;
+ var notification = notificationContext.Notification;
var notificationType = notification.GetType();
var handleMethod = notificationType.GetNotificationHandlerMethod();
foreach (var handler in context.Handlers)
- await handler.InvokeAsync(handleMethod, notification, cancellationToken);
+ await handler.InvokeAsync(handleMethod, context.NotificationContext);
}
}
\ No newline at end of file
diff --git a/src/modules/Elsa.Common/Multitenancy/Contracts/ITenantDeletedEvent.cs b/src/modules/Elsa.Common/Multitenancy/Contracts/ITenantDeletedEvent.cs
new file mode 100644
index 000000000..f651cdf70
--- /dev/null
+++ b/src/modules/Elsa.Common/Multitenancy/Contracts/ITenantDeletedEvent.cs
@@ -0,0 +1,6 @@
+namespace Elsa.Common.Multitenancy;
+
+public interface ITenantDeletedEvent
+{
+ Task TenantDeletedAsync(TenantDeletedEventArgs args);
+}
\ No newline at end of file
diff --git a/src/modules/Elsa.Common/Multitenancy/EventArgs/TenantDeletedEventArgs.cs b/src/modules/Elsa.Common/Multitenancy/EventArgs/TenantDeletedEventArgs.cs
new file mode 100644
index 000000000..07f6f1136
--- /dev/null
+++ b/src/modules/Elsa.Common/Multitenancy/EventArgs/TenantDeletedEventArgs.cs
@@ -0,0 +1,3 @@
+namespace Elsa.Common.Multitenancy;
+
+public record TenantDeletedEventArgs(Tenant Tenant, TenantScope TenantScope, CancellationToken CancellationToken) : TenantEventArgs(Tenant, TenantScope, CancellationToken);
\ No newline at end of file
diff --git a/src/modules/Elsa.Common/Multitenancy/Implementations/DefaultTenantService.cs b/src/modules/Elsa.Common/Multitenancy/Implementations/DefaultTenantService.cs
index 89298e5ac..6a1f9cb98 100644
--- a/src/modules/Elsa.Common/Multitenancy/Implementations/DefaultTenantService.cs
+++ b/src/modules/Elsa.Common/Multitenancy/Implementations/DefaultTenantService.cs
@@ -64,7 +64,7 @@ public class DefaultTenantService(IServiceScopeFactory scopeFactory, ITenantScop
var tenants = dictionary.Values.ToArray();
foreach (var tenant in tenants)
- await UnregisterTenantAsync(tenant, cancellationToken);
+ await UnregisterTenantAsync(tenant, false, cancellationToken);
}
public async Task RefreshAsync(CancellationToken cancellationToken = default)
@@ -85,7 +85,7 @@ public class DefaultTenantService(IServiceScopeFactory scopeFactory, ITenantScop
foreach (var removedTenantId in removedTenantIds)
{
var removedTenant = currentTenants[removedTenantId];
- await UnregisterTenantAsync(removedTenant, cancellationToken);
+ await UnregisterTenantAsync(removedTenant, true, cancellationToken);
}
foreach (var addedTenantId in addedTenantIds)
@@ -137,14 +137,19 @@ public class DefaultTenantService(IServiceScopeFactory scopeFactory, ITenantScop
await tenantEvents.TenantActivatedAsync(new(tenant, scope, cancellationToken));
}
- private async Task UnregisterTenantAsync(Tenant tenant, CancellationToken cancellationToken = default)
+ private async Task UnregisterTenantAsync(Tenant tenant, bool isDeleted, CancellationToken cancellationToken = default)
{
if (_tenantScopesDictionary!.Remove(tenant, out var scope))
{
_tenantsDictionary!.Remove(tenant.Id.EmptyIfNull(), out _);
using (tenantAccessor.PushContext(tenant))
+ {
await tenantEvents.TenantDeactivatedAsync(new(tenant, scope, cancellationToken));
+
+ if (isDeleted)
+ await tenantEvents.TenantDeletedAsync(new(tenant, scope, cancellationToken));
+ }
}
}
}
\ No newline at end of file
diff --git a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantEventsManager.cs b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantEventsManager.cs
index 1c8e85a85..c8b97639a 100644
--- a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantEventsManager.cs
+++ b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantEventsManager.cs
@@ -2,34 +2,54 @@ using Microsoft.Extensions.Logging;
namespace Elsa.Common.Multitenancy;
-public class TenantEventsManager(IEnumerable tenantActivatedEvents, IEnumerable tenantDeactivatedEvents, ILogger logger)
+public class TenantEventsManager(
+ IEnumerable tenantActivatedEvents,
+ IEnumerable tenantDeactivatedEvents,
+ IEnumerable tenantDeletedEvents,
+ ILogger logger)
{
public async Task TenantActivatedAsync(TenantActivatedEventArgs args)
{
- foreach (var tenantActivatedEvent in tenantActivatedEvents)
- {
- try
- {
- await tenantActivatedEvent.TenantActivatedAsync(args);
- }
- catch (Exception e)
- {
- logger.LogError(e, "Error occurred while processing tenant activated event.");
- }
- }
+ await ExecuteEventHandlersAsync(
+ tenantActivatedEvents,
+ (handler, eventArgs) => handler.TenantActivatedAsync(eventArgs),
+ args,
+ "activated");
}
public async Task TenantDeactivatedAsync(TenantDeactivatedEventArgs args)
{
- foreach (var tenantDeactivatedEvent in tenantDeactivatedEvents)
+ await ExecuteEventHandlersAsync(
+ tenantDeactivatedEvents,
+ (handler, eventArgs) => handler.TenantDeactivatedAsync(eventArgs),
+ args,
+ "deactivated");
+ }
+
+ public async Task TenantDeletedAsync(TenantDeletedEventArgs args)
+ {
+ await ExecuteEventHandlersAsync(
+ tenantDeletedEvents,
+ (handler, eventArgs) => handler.TenantDeletedAsync(eventArgs),
+ args,
+ "deleted");
+ }
+
+ private async Task ExecuteEventHandlersAsync(
+ IEnumerable handlers,
+ Func handlerAction,
+ TArgs args,
+ string eventType)
+ {
+ foreach (var handler in handlers)
{
try
{
- await tenantDeactivatedEvent.TenantDeactivatedAsync(args);
+ await handlerAction(handler, args);
}
catch (Exception e)
{
- logger.LogError(e, "Error occurred while processing tenant deactivated event.");
+ logger.LogError(e, "Error occurred while processing tenant {EventType} event.", eventType);
}
}
}
diff --git a/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs b/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs
index 41eb4b5eb..9dd4ce1d4 100644
--- a/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs
+++ b/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs
@@ -41,7 +41,7 @@ public class SchedulingFeature : FeatureBase
Services
.AddSingleton()
.AddSingleton(sp => sp.GetRequiredService())
- .AddSingleton(sp => sp.GetRequiredService())
+ .AddSingleton(sp => sp.GetRequiredService())
.AddSingleton()
.AddSingleton()
.AddSingleton(CronParser)
diff --git a/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs b/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs
index bdeecad26..46456cb9b 100644
--- a/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs
+++ b/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs
@@ -9,7 +9,7 @@ using Timer = Elsa.Scheduling.Activities.Timer;
namespace Elsa.Scheduling.Handlers;
-public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeactivatedEvent
+public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeletedEvent
{
private static readonly string[] ActivityTypeNames =
[
@@ -31,7 +31,7 @@ public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeactivatedEv
await bookmarkScheduler.ScheduleAsync(bookmarks, args.CancellationToken);
}
- public async Task TenantDeactivatedAsync(TenantDeactivatedEventArgs args)
+ public async Task TenantDeletedAsync(TenantDeletedEventArgs args)
{
var serviceProvider = args.TenantScope.ServiceProvider;
var cancellationToken = args.CancellationToken;
diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs
index faa250d04..7e5f87c59 100644
--- a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs
+++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs
@@ -63,9 +63,6 @@ public partial class ActivityExecutionContext
// Update the completed at timestamp.
CompletedAt = WorkflowExecutionContext.SystemClock.UtcNow;
-
- var mediator = GetRequiredService();
- await mediator.SendAsync(new Notifications.ActivityCompleted(this), CancellationToken);
}
///
diff --git a/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs b/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs
index 6731bb8a6..24ffb4939 100644
--- a/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs
+++ b/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs
@@ -1,5 +1,6 @@
using System.Reflection;
using Elsa.Extensions;
+using Elsa.Mediator.Contracts;
using Elsa.Workflows.Activities;
using Elsa.Workflows.CommitStates;
using Elsa.Workflows.Pipelines.ActivityExecution;
@@ -59,10 +60,14 @@ public class DefaultActivityInvokerMiddleware(ActivityMiddlewareDelegate next, I
if (ShouldCommit(context, ActivityLifetimeEvent.ActivityExecuting))
await context.WorkflowExecutionContext.CommitAsync();
+ var previousActivityStatus = context.Status;
context.TransitionTo(ActivityStatus.Running);
// Execute activity.
await ExecuteActivityAsync(context);
+
+ var currentActivityStatus = context.Status;
+ var activityDidComplete = previousActivityStatus != ActivityStatus.Completed && currentActivityStatus == ActivityStatus.Completed;
// Reset execute delegate.
workflowExecutionContext.ExecuteDelegate = null;
@@ -81,6 +86,13 @@ public class DefaultActivityInvokerMiddleware(ActivityMiddlewareDelegate next, I
// Invoke next middleware.
await next(context);
+
+ // If the activity completed, send a notification.
+ if (activityDidComplete)
+ {
+ var mediator = context.GetRequiredService();
+ await mediator.SendAsync(new Notifications.ActivityCompleted(context), context.CancellationToken);
+ }
// Conditionally commit the workflow state.
if (ShouldCommit(context, ActivityLifetimeEvent.ActivityExecuted))