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

Workflow refresh feature (#5713)

Browse source 
* Refactor workflow definition events handling and dispatch

Removed IDistributedWorkflowDefinitionEventsDispatcher interface and related classes in favor of direct message publishing through MassTransit. Introduced new workflow definition refresh endpoints and services for better modularity and responsiveness. Adjusted caching and notification systems to support the new architecture.

* Add comment

* Added workflow definition refresh component test

* Fixed component tests issue

* Fixed id

* Fixed test

* Rename IsRaisedFromConsumer to IsConsumerExecutionContext

Updated the naming for clarity and consistency across the system. This change affects AmbientConsumerScope, related consumers, and condition checks to improve readability and understanding of the consumer context.

---------

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>
This commit is contained in:
MariusVuscanNx 2024-07-05 19:35:22 +03:00 committed by GitHub
parent 1574b491ad
commit 6110eb561f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 433 additions and 173 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -49,7 +49,7 @@ const bool useMemoryStores = false;
const bool useCaching = true;
const bool useReadOnlyMode = false;
const DistributedCachingTransport distributedCachingTransport = DistributedCachingTransport.MassTransit;
const MassTransitBroker useMassTransitBroker = MassTransitBroker.Memory;
const MassTransitBroker useMassTransitBroker = MassTransitBroker.RabbitMq;
var builder = WebApplication.CreateBuilder(args);
var services = builder.Services;

43
src/modules/Elsa.MassTransit/Consumers/WorkflowDefinitionEventsConsumer.cs
View file

@ -1,27 +1,28 @@
using Elsa.MassTransit.Messages;
using Elsa.MassTransit.Services;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Management.Contracts;
using JetBrains.Annotations;
using MassTransit;
namespace Elsa.MassTransit.Consumers;
/// <summary>
/// Consumes messages related to workflow definition changes.
/// </summary>
[PublicAPI]
public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistryUpdater workflowDefinitionActivityRegistryUpdater) :
IConsumer<WorkflowDefinitionDeleted>,
IConsumer<WorkflowDefinitionPublished>,
IConsumer<WorkflowDefinitionRetracted>,
IConsumer<WorkflowDefinitionVersionRetracted>,
IConsumer<WorkflowDefinitionsDeleted>,
IConsumer<WorkflowDefinitionVersionDeleted>,
IConsumer<WorkflowDefinitionVersionsDeleted>,
IConsumer<WorkflowDefinitionVersionsUpdated>
public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistryUpdater workflowDefinitionActivityRegistryUpdater, INotificationSender notificationSender) :
global::MassTransit.IConsumer<WorkflowDefinitionDeleted>,
global::MassTransit.IConsumer<WorkflowDefinitionPublished>,
global::MassTransit.IConsumer<WorkflowDefinitionRetracted>,
global::MassTransit.IConsumer<WorkflowDefinitionVersionRetracted>,
global::MassTransit.IConsumer<WorkflowDefinitionsDeleted>,
global::MassTransit.IConsumer<WorkflowDefinitionVersionDeleted>,
global::MassTransit.IConsumer<WorkflowDefinitionVersionsDeleted>,
global::MassTransit.IConsumer<WorkflowDefinitionVersionsUpdated>,
global::MassTransit.IConsumer<WorkflowDefinitionsRefreshed>
{
/// <inheritdoc />
public Task Consume(ConsumeContext<WorkflowDefinitionDeleted> context)
{
{
workflowDefinitionActivityRegistryUpdater.RemoveDefinitionFromRegistry(context.Message.Id);
return Task.CompletedTask;
}
@ -48,9 +49,7 @@ public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistr
public Task Consume(ConsumeContext<WorkflowDefinitionsDeleted> context)
{
foreach (var id in context.Message.Ids)
{
workflowDefinitionActivityRegistryUpdater.RemoveDefinitionFromRegistry(id);
}
return Task.CompletedTask;
}
@ -66,9 +65,7 @@ public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistr
public Task Consume(ConsumeContext<WorkflowDefinitionVersionsDeleted> context)
{
foreach (var id in context.Message.Ids)
{
workflowDefinitionActivityRegistryUpdater.RemoveDefinitionVersionFromRegistry(id);
}
return Task.CompletedTask;
}
@ -77,14 +74,22 @@ public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistr
public async Task Consume(ConsumeContext<WorkflowDefinitionVersionsUpdated> context)
{
foreach (WorkflowDefinitionVersionUpdate definitionUpdate in context.Message.WorkflowDefinitionVersionUpdates)
{
await UpdateDefinition(definitionUpdate.Id, definitionUpdate.UsableAsActivity);
}
}
/// <inheritdoc />
public async Task Consume(ConsumeContext<WorkflowDefinitionsRefreshed> context)
{
var message = context.Message;
var notification = new Elsa.Workflows.Runtime.Notifications.WorkflowDefinitionsRefreshed(message.WorkflowDefinitionIds);
AmbientConsumerScope.IsConsumerExecutionContext = true;
await notificationSender.SendAsync(notification, context.CancellationToken);
AmbientConsumerScope.IsConsumerExecutionContext = false;
}
private Task UpdateDefinition(string id, bool usableAsActivity)
{
// Once a workflow has been published it should remain in the activity registry unless no longer being marked as an activity.
// Once a workflow has been published, it should remain in the activity registry unless no longer being marked as an activity.
if (usableAsActivity)
return workflowDefinitionActivityRegistryUpdater.AddToRegistry(id);

49
src/modules/Elsa.MassTransit/Contracts/IDistributedWorkflowDefinitionEventsDispatcher.cs
View file

@ -1,49 +0,0 @@
using Elsa.MassTransit.Messages;
namespace Elsa.MassTransit.Contracts;
/// <summary>
/// Dispatches workflow definition related notifications to refresh the activity registry.
/// </summary>
public interface IDistributedWorkflowDefinitionEventsDispatcher
{
/// <summary>
/// Dispatches a workflow definition publication event.
/// </summary>
Task DispatchAsync(WorkflowDefinitionPublished request, CancellationToken cancellationToken);
/// <summary>
/// Dispatches a workflow definition retracted event.
/// </summary>
Task DispatchAsync(WorkflowDefinitionRetracted request, CancellationToken cancellationToken);
/// <summary>
/// Dispatches a workflow definition version retracted event.
/// </summary>
Task DispatchAsync(WorkflowDefinitionVersionRetracted request, CancellationToken cancellationToken);
/// <summary>
/// Dispatches a workflow definition deleted event.
/// </summary>
Task DispatchAsync(WorkflowDefinitionDeleted request, CancellationToken cancellationToken);
/// <summary>
/// Dispatches a workflow definitions deleted event.
/// </summary>
Task DispatchAsync(WorkflowDefinitionsDeleted request, CancellationToken cancellationToken);
/// <summary>
/// Dispatches a workflow definition version deleted event.
/// </summary>
Task DispatchAsync(WorkflowDefinitionVersionDeleted request, CancellationToken cancellationToken);
/// <summary>
/// Dispatches a workflow definition versions deleted event.
/// </summary>
Task DispatchAsync(WorkflowDefinitionVersionsDeleted request, CancellationToken cancellationToken);
/// <summary>
/// Dispatches a workflow definition versions updated event.
/// </summary>
Task DispatchAsync(WorkflowDefinitionVersionsUpdated request, CancellationToken cancellationToken);
}

6
src/modules/Elsa.MassTransit/Features/MassTransitWorkflowManagementFeature.cs
View file

@ -1,11 +1,8 @@
using System.Diagnostics.CodeAnalysis;
using Elsa.Extensions;
using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.MassTransit.Consumers;
using Elsa.MassTransit.Contracts;
using Elsa.MassTransit.Services;
using Elsa.Workflows.Management.Features;
using Microsoft.Extensions.DependencyInjection;
@ -19,7 +16,6 @@ namespace Elsa.MassTransit.Features;
public class MassTransitWorkflowManagementFeature(IModule module) : FeatureBase(module)
{
/// <inheritdoc />
[RequiresUnreferencedCode("The assembly containing the specified marker type will be scanned for activity types.")]
public override void Configure()
{
Module.AddMassTransitConsumer<WorkflowDefinitionEventsConsumer>("elsa-workflow-definition-updates", true, true);
@ -28,8 +24,6 @@ public class MassTransitWorkflowManagementFeature(IModule module) : FeatureBase(
/// <inheritdoc />
public override void Apply()
{
Services.AddScoped<IDistributedWorkflowDefinitionEventsDispatcher, MassTransitDistributedEventsDispatcher>();
Services.AddNotificationHandlersFrom(GetType());
}
}

70
src/modules/Elsa.MassTransit/Handlers/DistributedWorkflowDefinitionNotificationsHandler.cs
View file

@ -1,12 +1,15 @@
using Elsa.MassTransit.Contracts;
using Elsa.MassTransit.Services;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Management.Notifications;
using Elsa.Workflows.Runtime.Notifications;
using MassTransit;
using Distributed = Elsa.MassTransit.Messages;
namespace Elsa.MassTransit.Handlers;
/// Represents a handler for distributed workflow definition notifications.
public class DistributedWorkflowDefinitionNotificationsHandler(IDistributedWorkflowDefinitionEventsDispatcher distributedEventsDispatcher) :
public class DistributedWorkflowDefinitionNotificationsHandler(IBus bus) :
INotificationHandler<WorkflowDefinitionPublished>,
INotificationHandler<WorkflowDefinitionRetracted>,
INotificationHandler<WorkflowDefinitionVersionRetracted>,
@ -14,44 +17,67 @@ public class DistributedWorkflowDefinitionNotificationsHandler(IDistributedWorkf
INotificationHandler<WorkflowDefinitionsDeleted>,
INotificationHandler<WorkflowDefinitionVersionDeleted>,
INotificationHandler<WorkflowDefinitionVersionsDeleted>,
INotificationHandler<WorkflowDefinitionVersionsUpdated>
INotificationHandler<WorkflowDefinitionVersionsUpdated>,
INotificationHandler<WorkflowDefinitionsRefreshed>
{
/// <inheritdoc />
public Task HandleAsync(WorkflowDefinitionPublished notification, CancellationToken cancellationToken) =>
distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionPublished(notification.WorkflowDefinition.Id,
notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken);
public Task HandleAsync(WorkflowDefinitionPublished notification, CancellationToken cancellationToken)
{
return bus.Publish(new Distributed.WorkflowDefinitionPublished(notification.WorkflowDefinition.Id, notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken);
}
/// <inheritdoc />
public Task HandleAsync(WorkflowDefinitionRetracted notification, CancellationToken cancellationToken) =>
distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionRetracted(notification.WorkflowDefinition.Id,
notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken);
public Task HandleAsync(WorkflowDefinitionRetracted notification, CancellationToken cancellationToken)
{
return bus.Publish(new Distributed.WorkflowDefinitionRetracted(notification.WorkflowDefinition.Id, notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken);
}
/// <inheritdoc />
public Task HandleAsync(WorkflowDefinitionVersionRetracted notification, CancellationToken cancellationToken) =>
distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionVersionRetracted(notification.WorkflowDefinition.Id,
notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken);
public Task HandleAsync(WorkflowDefinitionVersionRetracted notification, CancellationToken cancellationToken)
{
return bus.Publish(new Distributed.WorkflowDefinitionVersionRetracted(notification.WorkflowDefinition.Id, notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken);
}
/// <inheritdoc />
public Task HandleAsync(WorkflowDefinitionDeleted notification, CancellationToken cancellationToken) =>
distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionDeleted(notification.DefinitionId), cancellationToken);
public Task HandleAsync(WorkflowDefinitionDeleted notification, CancellationToken cancellationToken)
{
return bus.Publish(new Distributed.WorkflowDefinitionDeleted(notification.DefinitionId), cancellationToken);
}
/// <inheritdoc />
public Task HandleAsync(WorkflowDefinitionsDeleted notification, CancellationToken cancellationToken) =>
distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionsDeleted(notification.DefinitionIds), cancellationToken);
public Task HandleAsync(WorkflowDefinitionsDeleted notification, CancellationToken cancellationToken)
{
return bus.Publish(new Distributed.WorkflowDefinitionsDeleted(notification.DefinitionIds), cancellationToken);
}
/// <inheritdoc />
public Task HandleAsync(WorkflowDefinitionVersionDeleted notification, CancellationToken cancellationToken) =>
distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionVersionDeleted(notification.WorkflowDefinition.Id), cancellationToken);
public Task HandleAsync(WorkflowDefinitionVersionDeleted notification, CancellationToken cancellationToken)
{
return bus.Publish(new Distributed.WorkflowDefinitionVersionDeleted(notification.WorkflowDefinition.Id), cancellationToken);
}
/// <inheritdoc />
public Task HandleAsync(WorkflowDefinitionVersionsDeleted notification, CancellationToken cancellationToken) =>
distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionVersionsDeleted(notification.Ids), cancellationToken);
public Task HandleAsync(WorkflowDefinitionVersionsDeleted notification, CancellationToken cancellationToken)
{
return bus.Publish(new Distributed.WorkflowDefinitionVersionsDeleted(notification.Ids), cancellationToken);
}
/// <inheritdoc />
public async Task HandleAsync(WorkflowDefinitionVersionsUpdated notification, CancellationToken cancellationToken)
{
var updates = notification.WorkflowDefinitions.Select(x =>
new Distributed.WorkflowDefinitionVersionUpdate(x.Id, x.DefinitionId, x.IsPublished, x.Options.UsableAsActivity.GetValueOrDefault()));
await distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionVersionsUpdated(updates), cancellationToken);
var updates = notification.WorkflowDefinitions.Select(x => new Distributed.WorkflowDefinitionVersionUpdate(x.Id, x.DefinitionId, x.IsPublished, x.Options.UsableAsActivity.GetValueOrDefault()));
await bus.Publish(new Distributed.WorkflowDefinitionVersionsUpdated(updates), cancellationToken);
}
/// <inheritdoc />
public Task HandleAsync(WorkflowDefinitionsRefreshed notification, CancellationToken cancellationToken)
{
// Prevent re-entrance.
if (AmbientConsumerScope.IsConsumerExecutionContext)
return Task.CompletedTask;
var definitionIds = notification.WorkflowDefinitionIds;
var message = new Distributed.WorkflowDefinitionsRefreshed(definitionIds);
return bus.Publish(message, cancellationToken);
}
}

4
src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionVersionsUpdated.cs
View file

@ -1,12 +1,8 @@
namespace Elsa.MassTransit.Messages;
/// <summary>
/// Represents a distributed message that multiple workflow definition versions have been deleted.
/// </summary>
public class WorkflowDefinitionVersionsUpdated(IEnumerable<WorkflowDefinitionVersionUpdate> workflowDefinitionVersionUpdates)
{
/// <summary>
/// Represents a collection of updates to workflow definition versions.
/// </summary>
public IEnumerable<WorkflowDefinitionVersionUpdate> WorkflowDefinitionVersionUpdates { get; set; } = workflowDefinitionVersionUpdates;
}

8
src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionsRefreshed.cs Normal file
View file

@ -0,0 +1,8 @@
namespace Elsa.MassTransit.Messages;
/// Represents a message that indicates that the specified workflow definitions have been refreshed.
public class WorkflowDefinitionsRefreshed(ICollection<string> workflowDefinitionIds)
{
/// The workflow definition IDs that have been refreshed.
public ICollection<string> WorkflowDefinitionIds { get; set; } = workflowDefinitionIds;
}

14
src/modules/Elsa.MassTransit/Services/AmbientConsumerScope.cs Normal file
View file

@ -0,0 +1,14 @@
namespace Elsa.MassTransit.Services;
/// Holds ambient state about a consumer.
public static class AmbientConsumerScope
{
private static readonly AsyncLocal<bool> IsRaisedFromConsumerState = new();
/// Gets or sets a value that indicates if the current code is initiated from a consumer.
public static bool IsConsumerExecutionContext
{
get => IsRaisedFromConsumerState.Value;
set => IsRaisedFromConsumerState.Value = value;
}
}

59
src/modules/Elsa.MassTransit/Services/MassTransitDistributedEventsDispatcher.cs
View file

@ -1,59 +0,0 @@
using Elsa.MassTransit.Contracts;
using Elsa.MassTransit.Messages;
using MassTransit;
namespace Elsa.MassTransit.Services;
/// <summary>
/// Dispatches workflow definition related notifications via MassTransit.
/// </summary>
public class MassTransitDistributedEventsDispatcher(IBus bus) : IDistributedWorkflowDefinitionEventsDispatcher
{
/// <inheritdoc />
public Task DispatchAsync(WorkflowDefinitionPublished request, CancellationToken cancellationToken)
{
return bus.Publish(request, cancellationToken);
}
/// <inheritdoc />
public Task DispatchAsync(WorkflowDefinitionRetracted request, CancellationToken cancellationToken)
{
return bus.Publish(request, cancellationToken);
}
/// <inheritdoc />
public Task DispatchAsync(WorkflowDefinitionVersionRetracted request, CancellationToken cancellationToken)
{
return bus.Publish(request, cancellationToken);
}
/// <inheritdoc />
public Task DispatchAsync(WorkflowDefinitionDeleted request, CancellationToken cancellationToken)
{
return bus.Publish(request, cancellationToken);
}
/// <inheritdoc />
public Task DispatchAsync(WorkflowDefinitionsDeleted request, CancellationToken cancellationToken)
{
return bus.Publish(request, cancellationToken);
}
/// <inheritdoc />
public Task DispatchAsync(WorkflowDefinitionVersionDeleted request, CancellationToken cancellationToken)
{
return bus.Publish(request, cancellationToken);
}
/// <inheritdoc />
public Task DispatchAsync(WorkflowDefinitionVersionsDeleted request, CancellationToken cancellationToken)
{
return bus.Publish(request, cancellationToken);
}
/// <inheritdoc />
public Task DispatchAsync(WorkflowDefinitionVersionsUpdated request, CancellationToken cancellationToken)
{
return bus.Publish(request, cancellationToken);
}
}

30
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs Normal file
View file

@ -0,0 +1,30 @@
using Elsa.Abstractions;
using Elsa.Workflows.Runtime.Contracts;
using Elsa.Workflows.Runtime.Requests;
using JetBrains.Annotations;
namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Refresh;
[PublicAPI]
internal class Refresh(IWorkflowDefinitionsRefresher workflowDefinitionsRefresher) : ElsaEndpoint<Request>
{
private const int BatchSize = 10;
public override void Configure()
{
Post("/actions/workflow-definitions/refresh");
ConfigurePermissions("actions:workflow-definitions:refresh");
}
public override async Task HandleAsync(Request request, CancellationToken cancellationToken)
{
await RefreshWorkflowDefinitionsAsync(request.DefinitionIds, cancellationToken);
await SendOkAsync(cancellationToken);
}
private async Task RefreshWorkflowDefinitionsAsync(ICollection<string>? definitionIds, CancellationToken cancellationToken)
{
var request = new RefreshWorkflowDefinitionsRequest(definitionIds, BatchSize);
await workflowDefinitionsRefresher.RefreshWorkflowDefinitionsAsync(request, cancellationToken);
}
}

6
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Models.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Refresh;
public class Request
{
public ICollection<string>? DefinitionIds { get; set; }
}

11
src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionRefresher.cs Normal file
View file

@ -0,0 +1,11 @@
using Elsa.Workflows.Runtime.Requests;
using Elsa.Workflows.Runtime.Responses;
namespace Elsa.Workflows.Runtime.Contracts;
/// Refreshes all workflows by re-indexing their triggers.
public interface IWorkflowDefinitionsRefresher
{
/// Refreshes all workflows by re-indexing their triggers.
Task<RefreshWorkflowDefinitionsResponse> RefreshWorkflowDefinitionsAsync(RefreshWorkflowDefinitionsRequest request, CancellationToken cancellationToken = default);
}

11
src/modules/Elsa.Workflows.Runtime/Features/CachingWorkflowRuntimeFeature.cs
View file

@ -3,14 +3,13 @@ using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.Workflows.Runtime.Contracts;
using Elsa.Workflows.Runtime.Handlers;
using Elsa.Workflows.Runtime.Stores;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Workflows.Runtime.Features;
/// <summary>
/// Installs and configures workflow runtime caching features.
/// </summary>
[DependsOn(typeof(MemoryCacheFeature))]
public class CachingWorkflowRuntimeFeature : FeatureBase
{
@ -22,7 +21,11 @@ public class CachingWorkflowRuntimeFeature : FeatureBase
/// <inheritdoc />
public override void Apply()
{
// Decorators.
Services.Decorate<ITriggerStore, CachingTriggerStore>();
Services
// Decorators.
.Decorate<ITriggerStore, CachingTriggerStore>()
// Handlers.
.AddNotificationHandler<InvalidateTriggersCache>();
}
}

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

@ -220,6 +220,7 @@ public class WorkflowRuntimeFeature : FeatureBase
.AddScoped<IActivityExecutionMapper, DefaultActivityExecutionMapper>()
.AddScoped<IWorkflowDefinitionStorePopulator, DefaultWorkflowDefinitionStorePopulator>()
.AddScoped<IRegistriesPopulator, DefaultRegistriesPopulator>()
.AddScoped<IWorkflowDefinitionsRefresher, WorkflowDefinitionsRefresher>()
.AddScoped<IWorkflowRegistry, DefaultWorkflowRegistry>()
.AddScoped<ITaskReporter, TaskReporter>()
.AddScoped<SynchronousTaskDispatcher>()

25
src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateTriggersCache.cs Normal file
View file

@ -0,0 +1,25 @@
using Elsa.Caching;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Runtime.Notifications;
using Elsa.Workflows.Runtime.Stores;
using JetBrains.Annotations;
namespace Elsa.Workflows.Runtime.Handlers;
/// <summary>
/// A notification handler that invalidates triggers cache when workflow definitions are refreshed.
/// </summary>
/// <remarks>
/// The class implements the <c>INotificationHandler</c> interface and is responsible for handling <c>WorkflowDefinitionsRefreshed</c> notifications.
/// When a <c>WorkflowDefinitionsRefreshed</c> notification is received, the <c>HandleAsync</c> method is called to invalidate the triggers cache.
/// The cache is invalidated by calling the <c>TriggerTokenAsync</c> method of the <c>ICacheManager</c> passed to the class constructor.
/// </remarks>
[UsedImplicitly]
public class InvalidateTriggersCache(ICacheManager cacheManager) : INotificationHandler<WorkflowDefinitionsRefreshed>
{
/// <inheritdoc />
public Task HandleAsync(WorkflowDefinitionsRefreshed notification, CancellationToken cancellationToken)
{
return cacheManager.TriggerTokenAsync(CachingTriggerStore.CacheInvalidationTokenKey, cancellationToken).AsTask();
}
}

6
src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowDefinitionsRefreshed.cs Normal file
View file

@ -0,0 +1,6 @@
using Elsa.Mediator.Contracts;
namespace Elsa.Workflows.Runtime.Notifications;
/// Published when workflow definitions have been refreshed.
public record WorkflowDefinitionsRefreshed(ICollection<string> WorkflowDefinitionIds) : INotification;

2
src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowTriggersIndexed.cs
View file

@ -2,8 +2,6 @@ using Elsa.Mediator.Contracts;
namespace Elsa.Workflows.Runtime.Notifications;
/// <summary>
/// Published when the specified workflow's triggers have been indexed.
/// </summary>
/// <param name="IndexedWorkflowTriggers">Contains the workflow that was indexed and the resulting triggers.</param>
public record WorkflowTriggersIndexed(IndexedWorkflowTriggers IndexedWorkflowTriggers) : INotification;

25
src/modules/Elsa.Workflows.Runtime/Requests/RefreshWorkflowDefinitionsRequest.cs Normal file
View file

@ -0,0 +1,25 @@
namespace Elsa.Workflows.Runtime.Requests;
/// Represents a request to refresh workflow definitions by re-indexing their triggers.
public class RefreshWorkflowDefinitionsRequest
{
/// Initializes a new instance of <see cref="RefreshWorkflowDefinitionsRequest"/>.
public RefreshWorkflowDefinitionsRequest()
{
}
/// Initializes a new instance of <see cref="RefreshWorkflowDefinitionsRequest"/>.
public RefreshWorkflowDefinitionsRequest(IEnumerable<string>? definitionIds, int batchSize)
{
BatchSize = batchSize;
DefinitionIds = definitionIds?.ToList();
}
/// Gets or sets the batch size for refreshing workflow definitions.
/// The batch size determines the number of definitions to be processed at a time during the refresh process.
public int BatchSize { get; set; } = 10;
/// Gets or sets the collection of definition IDs for workflow definitions to be refreshed.
/// If not specified, all workflows will be refreshed.
public ICollection<string>? DefinitionIds { get; set; }
}

6
src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs Normal file
View file

@ -0,0 +1,6 @@
using Elsa.Workflows.Management.Entities;
namespace Elsa.Workflows.Runtime.Responses;
/// Represents a response to a request to refresh workflow definitions.
public record RefreshWorkflowDefinitionsResponse(ICollection<WorkflowDefinition> WorkflowDefinitions);

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

@ -86,7 +86,7 @@ public class TriggerIndexer : ITriggerIndexer
// Get current triggers
var currentTriggers = await GetCurrentTriggersAsync(workflow.Identity.DefinitionId, cancellationToken).ToList();
// Collect new triggers **if workflow is published**.
// Collect new triggers **if the workflow is published**.
var newTriggers = workflow.Publication.IsPublished
? await GetTriggersInternalAsync(workflow, cancellationToken).ToListAsync(cancellationToken)
: new List<StoredTrigger>(0);

55
src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionRefresher.cs Normal file
View file

@ -0,0 +1,55 @@
using Elsa.Common.Entities;
using Elsa.Common.Models;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Management.Contracts;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Filters;
using Elsa.Workflows.Runtime.Contracts;
using Elsa.Workflows.Runtime.Notifications;
using Elsa.Workflows.Runtime.Requests;
using Elsa.Workflows.Runtime.Responses;
namespace Elsa.Workflows.Runtime.Services;
/// <inheritdoc />
public class WorkflowDefinitionsRefresher(IWorkflowDefinitionStore store, ITriggerIndexer triggerIndexer, INotificationSender notificationSender) : IWorkflowDefinitionsRefresher
{
/// <inheritdoc />
public async Task<RefreshWorkflowDefinitionsResponse> RefreshWorkflowDefinitionsAsync(RefreshWorkflowDefinitionsRequest request, CancellationToken cancellationToken)
{
var filter = new WorkflowDefinitionFilter
{
DefinitionIds = request.DefinitionIds,
VersionOptions = VersionOptions.Published
};
var currentPage = 0;
var processedWorkflowDefinitions = new List<WorkflowDefinition>();
var batchSize = request.BatchSize;
var order = new WorkflowDefinitionOrder<string>(x => x.Id, OrderDirection.Ascending);
while (!cancellationToken.IsCancellationRequested)
{
var pageArgs = PageArgs.FromPage(currentPage, batchSize);
var definitions = await store.FindManyAsync(filter, order, pageArgs, cancellationToken);
if (definitions.Items.Count == 0)
break;
await IndexWorkflowTriggersAsync(definitions.Items, cancellationToken);
processedWorkflowDefinitions.AddRange(definitions.Items);
currentPage++;
}
var processedWorkflowDefinitionIds = processedWorkflowDefinitions.Select(x => x.Id).ToList();
var notification = new WorkflowDefinitionsRefreshed(processedWorkflowDefinitionIds);
await notificationSender.SendAsync(notification, cancellationToken);
return new RefreshWorkflowDefinitionsResponse(processedWorkflowDefinitions);
}
private async Task IndexWorkflowTriggersAsync(IEnumerable<WorkflowDefinition> definitions, CancellationToken cancellationToken)
{
foreach (var definition in definitions)
await triggerIndexer.IndexTriggersAsync(definition, cancellationToken);
}
}

3
src/modules/Elsa.Workflows.Runtime/Stores/CachingTriggerStore.cs
View file

@ -12,7 +12,8 @@ namespace Elsa.Workflows.Runtime.Stores;
/// </summary>
public class CachingTriggerStore(ITriggerStore decoratedStore, ICacheManager cacheManager, IHasher hasher) : ITriggerStore
{
private static readonly string CacheInvalidationTokenKey = typeof(CachingTriggerStore).FullName!;
/// The token key to evict all cached triggers.
public static readonly string CacheInvalidationTokenKey = typeof(CachingTriggerStore).FullName!;
/// <inheritdoc />
public async ValueTask SaveAsync(StoredTrigger record, CancellationToken cancellationToken = default)

13
test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj
View file

@ -1,4 +1,4 @@
<Project Sdk="Microsoft.NET.Sdk">
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
@ -10,13 +10,13 @@
</PropertyGroup>
<ItemGroup>
<Using Include="Xunit"/>
<Using Include="Xunit" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\src\bundles\Elsa.Server.Web\Elsa.Server.Web.csproj"/>
<ProjectReference Include="..\..\..\src\clients\Elsa.Api.Client\Elsa.Api.Client.csproj"/>
<ProjectReference Include="..\..\..\src\common\Elsa.Testing.Shared\Elsa.Testing.Shared.csproj"/>
<ProjectReference Include="..\..\..\src\bundles\Elsa.Server.Web\Elsa.Server.Web.csproj" />
<ProjectReference Include="..\..\..\src\clients\Elsa.Api.Client\Elsa.Api.Client.csproj" />
<ProjectReference Include="..\..\..\src\common\Elsa.Testing.Shared\Elsa.Testing.Shared.csproj" />
<ProjectReference Include="..\..\..\src\modules\Elsa.Caching.Distributed.MassTransit\Elsa.Caching.Distributed.MassTransit.csproj" />
</ItemGroup>
@ -72,6 +72,9 @@
<None Update="Scenarios\WorkflowJsonStructures\sub.json">
<CopyToOutputDirectory>Always</CopyToOutputDirectory>
</None>
<None Update="Scenarios\WorkflowDefinitionRefresh\dynamic-endpoint-http-workflow.json">
<CopyToOutputDirectory>Always</CopyToOutputDirectory>
</None>
</ItemGroup>
</Project>

9
test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs
View file

@ -6,6 +6,7 @@ using Elsa.Extensions;
using Elsa.Identity.Providers;
using Elsa.MassTransit.Extensions;
using Elsa.Workflows.ComponentTests.Consumers;
using Elsa.Workflows.ComponentTests.Helpers.Services;
using Elsa.Workflows.ComponentTests.Services;
using FluentStorage;
using Hangfire.Annotations;
@ -80,6 +81,14 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl
runtime.UseMassTransitDispatcher();
runtime.UseCache();
});
elsa.UseJavaScript(options =>
{
options.AllowClrAccess = true;
options.ConfigureEngine(engine =>
{
engine.SetValue("getStaticValue", () => StaticValueHolder.Value);
});
});
};
}

6
test/component/Elsa.Workflows.ComponentTests/Helpers/Services/StaticValueHolder.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Workflows.ComponentTests.Helpers.Services;
public static class StaticValueHolder
{
public static object? Value { get; set; }
}

36
test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/DynamicEndpointTests.cs Normal file
View file

@ -0,0 +1,36 @@
using Elsa.Workflows.ComponentTests.Helpers.Services;
using Elsa.Workflows.Runtime.Contracts;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Workflows.ComponentTests.Scenarios.WorkflowDefinitionRefresh;
public class DynamicEndpointTests : AppComponentTest
{
private readonly IWorkflowDefinitionsRefresher _workflowDefinitionsRefresher;
public DynamicEndpointTests(App app) : base(app)
{
StaticValueHolder.Value = "first-value";
_workflowDefinitionsRefresher = Scope.ServiceProvider.GetRequiredService<IWorkflowDefinitionsRefresher>();
}
[Fact]
public async Task HelloWorldWorkflow_ShouldRespondWithHelloWorld()
{
var client = WorkflowServer.CreateHttpWorkflowClient();
var firstResponse = await client.GetStringAsync("first-value");
StaticValueHolder.Value = "second-value";
await _workflowDefinitionsRefresher.RefreshWorkflowDefinitionsAsync(new Runtime.Requests.RefreshWorkflowDefinitionsRequest
{
BatchSize = 10,
DefinitionIds = ["f69f061159adc3ae"]
}, CancellationToken.None);
var secondResponse = await client.GetStringAsync("second-value");
Assert.Equal("", firstResponse);
Assert.Equal("", secondResponse);
}
}

104
test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/dynamic-endpoint-http-workflow.json Normal file
View file

@ -0,0 +1,104 @@
{
"id": "69cb6deaa8e20599",
"definitionId": "f69f061159adc3ae",
"name": "Workflow 1",
"createdAt": "2024-07-04T13:41:33.1905134+00:00",
"version": 1,
"toolVersion": "3.3.0.0",
"variables": [],
"inputs": [],
"outputs": [],
"outcomes": [],
"customProperties": {
"Elsa:WorkflowContextProviderTypes": []
},
"isReadonly": false,
"isSystem": false,
"isLatest": true,
"isPublished": true,
"options": {
"autoUpdateConsumingWorkflows": false
},
"root": {
"type": "Elsa.Flowchart",
"version": 1,
"id": "1bac532ee993dfdc",
"nodeId": "Workflow1:1bac532ee993dfdc",
"metadata": {},
"customProperties": {
"source": "FlowchartJsonConverter.cs:45",
"notFoundConnections": [],
"canStartWorkflow": false,
"runAsynchronously": false
},
"activities": [
{
"path": {
"typeName": "String",
"expression": {
"type": "JavaScript",
"value": "return getStaticValue() ?? \u0022first-value\u0022;"
}
},
"supportedMethods": {
"typeName": "String[]",
"expression": {
"type": "Literal",
"value": "[\u0022GET\u0022]"
}
},
"authorize": {
"typeName": "Boolean",
"expression": {
"type": "Literal",
"value": false
}
},
"policy": {
"typeName": "String",
"expression": {
"type": "Literal"
}
},
"requestTimeout": null,
"requestSizeLimit": null,
"fileSizeLimit": null,
"allowedFileExtensions": null,
"blockedFileExtensions": null,
"allowedMimeTypes": null,
"exposeRequestTooLargeOutcome": false,
"exposeFileTooLargeOutcome": false,
"exposeInvalidFileExtensionOutcome": false,
"exposeInvalidFileMimeTypeOutcome": false,
"parsedContent": null,
"files": null,
"routeData": null,
"queryStringData": null,
"headers": null,
"result": null,
"id": "25d3a6465094a796",
"nodeId": "Workflow1:1bac532ee993dfdc:25d3a6465094a796",
"name": "HttpEndpoint1",
"type": "Elsa.HttpEndpoint",
"version": 1,
"customProperties": {
"canStartWorkflow": true,
"runAsynchronously": false
},
"metadata": {
"designer": {
"position": {
"x": -260,
"y": 460
},
"size": {
"width": 176.390625,
"height": 50
}
}
}
}
],
"connections": []
}
}