From feb7da4d28f222f8db25df02974faff1d73a2ce0 Mon Sep 17 00:00:00 2001 From: MariusVuscanNx <96233009+MariusVuscanNx@users.noreply.github.com> Date: Fri, 12 Jul 2024 09:35:46 +0300 Subject: [PATCH] Implemented workflows reload endpoint (#5732) * Implemented workflows reload endpoint * Added componenent tests for reload * Adjusted cache handling * Made the refresh endpoint return refreshed and not found definitions * Update test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionReload/RemoveReloadWorkflowTests.cs Fix typo Co-authored-by: raymonddenhaan <155616759+raymonddenhaan@users.noreply.github.com> --------- Co-authored-by: Sipke Schoorstra Co-authored-by: raymonddenhaan <155616759+raymonddenhaan@users.noreply.github.com> --- .../Handlers/InvalidateHttpWorkflowsCache.cs | 24 ++-- .../WorkflowDefinitionEventsConsumer.cs | 17 ++- ...dWorkflowDefinitionNotificationsHandler.cs | 18 ++- .../Messages/WorkflowDefinitionsReloaded.cs | 8 ++ .../Services/AmbientConsumerScope.cs | 2 +- .../WorkflowDefinitions/Refresh/Endpoint.cs | 9 +- .../WorkflowDefinitions/Refresh/Models.cs | 10 +- .../WorkflowDefinitions/Reload/Endpoint.cs | 28 +++++ .../IWorkflowDefinitionStorePopulator.cs | 6 +- .../Contracts/IWorkflowDefinitionsReloader.cs | 8 ++ .../Features/WorkflowRuntimeFeature.cs | 1 + .../Handlers/InvalidateTriggersCache.cs | 10 +- .../Handlers/InvalidateWorkflowsCache.cs | 26 +++++ .../WorkflowDefinitionsReloaded.cs | 6 + .../RefreshWorkflowDefinitionsResponse.cs | 2 +- ...DefaultWorkflowDefinitionStorePopulator.cs | 13 ++- .../Services/WorkflowDefinitionRefresher.cs | 7 +- .../Services/WorkflowDefinitionsReloader.cs | 17 +++ .../Elsa.Workflows.ComponentTests.csproj | 6 + .../DynamicEndpointTests.cs | 20 ++-- .../RemoveReloadWorkflowTests.cs | 35 ++++++ .../http-workflow.json | 104 ++++++++++++++++++ 22 files changed, 336 insertions(+), 41 deletions(-) create mode 100644 src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionsReloaded.cs create mode 100644 src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Reload/Endpoint.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionsReloader.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateWorkflowsCache.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowDefinitionsReloaded.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionsReloader.cs create mode 100644 test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionReload/RemoveReloadWorkflowTests.cs create mode 100644 test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionReload/http-workflow.json diff --git a/src/modules/Elsa.Http/Handlers/InvalidateHttpWorkflowsCache.cs b/src/modules/Elsa.Http/Handlers/InvalidateHttpWorkflowsCache.cs index 714ee1a36..bb87afe99 100644 --- a/src/modules/Elsa.Http/Handlers/InvalidateHttpWorkflowsCache.cs +++ b/src/modules/Elsa.Http/Handlers/InvalidateHttpWorkflowsCache.cs @@ -15,17 +15,18 @@ namespace Elsa.Http.Handlers; /// A handler that invalidates the HTTP workflows cache when a workflow definition is published, retracted, or deleted or when triggers are indexed. /// [UsedImplicitly] -public class InvalidateHttpWorkflowsCache(IHttpWorkflowsCacheManager httpWorkflowsCacheManager, - ITriggerStore triggerStore, - IHttpWorkflowsCacheManager cacheManager) : +public class InvalidateHttpWorkflowsCache( + IHttpWorkflowsCacheManager httpWorkflowsCacheManager, + ITriggerStore triggerStore) : INotificationHandler, INotificationHandler, INotificationHandler, - INotificationHandler, + INotificationHandler, INotificationHandler, INotificationHandler, INotificationHandler, - INotificationHandler + INotificationHandler, + INotificationHandler { /// public Task HandleAsync(WorkflowDefinitionPublished notification, CancellationToken cancellationToken) @@ -91,6 +92,15 @@ public class InvalidateHttpWorkflowsCache(IHttpWorkflowsCacheManager httpWorkflo await InvalidateCacheAsync(notification.IndexedWorkflowTriggers.Workflow.Identity.DefinitionId); } + /// + public async Task HandleAsync(WorkflowDefinitionsReloaded notification, CancellationToken cancellationToken) + { + foreach (var workflowDefinitionId in notification.WorkflowDefinitionIds) + { + await InvalidateCacheAsync(workflowDefinitionId); + } + } + private async Task InvalidateCacheAsync(string workflowDefinitionId) { await httpWorkflowsCacheManager.EvictWorkflowAsync(workflowDefinitionId); @@ -103,7 +113,7 @@ public class InvalidateHttpWorkflowsCache(IHttpWorkflowsCacheManager httpWorkflo WorkflowDefinitionVersionId = workflowDefinitionVersionId }; var triggers = await triggerStore.FindManyAsync(filter, cancellationToken); - + await InvalidateTriggerCacheAsync(triggers, cancellationToken); } @@ -113,7 +123,7 @@ public class InvalidateHttpWorkflowsCache(IHttpWorkflowsCacheManager httpWorkflo { if (trigger?.Payload is HttpEndpointBookmarkPayload httpPayload) { - var hash = cacheManager.ComputeBookmarkHash(httpPayload.Path, httpPayload.Method); + var hash = httpWorkflowsCacheManager.ComputeBookmarkHash(httpPayload.Path, httpPayload.Method); await httpWorkflowsCacheManager.EvictTriggerAsync(hash, cancellationToken); } } diff --git a/src/modules/Elsa.MassTransit/Consumers/WorkflowDefinitionEventsConsumer.cs b/src/modules/Elsa.MassTransit/Consumers/WorkflowDefinitionEventsConsumer.cs index 6fed5a9a5..d43d11172 100644 --- a/src/modules/Elsa.MassTransit/Consumers/WorkflowDefinitionEventsConsumer.cs +++ b/src/modules/Elsa.MassTransit/Consumers/WorkflowDefinitionEventsConsumer.cs @@ -18,7 +18,8 @@ public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistr global::MassTransit.IConsumer, global::MassTransit.IConsumer, global::MassTransit.IConsumer, - global::MassTransit.IConsumer + global::MassTransit.IConsumer, + global::MassTransit.IConsumer { /// public Task Consume(ConsumeContext context) @@ -82,9 +83,19 @@ public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistr { var message = context.Message; var notification = new Elsa.Workflows.Runtime.Notifications.WorkflowDefinitionsRefreshed(message.WorkflowDefinitionIds); - AmbientConsumerScope.IsConsumerExecutionContext = true; + AmbientConsumerScope.IsWorkflowDefinitionEventsConsumer = true; await notificationSender.SendAsync(notification, context.CancellationToken); - AmbientConsumerScope.IsConsumerExecutionContext = false; + AmbientConsumerScope.IsWorkflowDefinitionEventsConsumer = false; + } + + /// + public async Task Consume(ConsumeContext context) + { + var message = context.Message; + var notification = new Elsa.Workflows.Runtime.Notifications.WorkflowDefinitionsReloaded(message.WorkflowDefinitionIds); + AmbientConsumerScope.IsWorkflowDefinitionEventsConsumer = true; + await notificationSender.SendAsync(notification, context.CancellationToken); + AmbientConsumerScope.IsWorkflowDefinitionEventsConsumer = false; } private Task UpdateDefinition(string id, bool usableAsActivity) diff --git a/src/modules/Elsa.MassTransit/Handlers/DistributedWorkflowDefinitionNotificationsHandler.cs b/src/modules/Elsa.MassTransit/Handlers/DistributedWorkflowDefinitionNotificationsHandler.cs index 12fdef392..991eb0914 100644 --- a/src/modules/Elsa.MassTransit/Handlers/DistributedWorkflowDefinitionNotificationsHandler.cs +++ b/src/modules/Elsa.MassTransit/Handlers/DistributedWorkflowDefinitionNotificationsHandler.cs @@ -1,4 +1,3 @@ -using Elsa.MassTransit.Contracts; using Elsa.MassTransit.Services; using Elsa.Mediator.Contracts; using Elsa.Workflows.Management.Notifications; @@ -18,7 +17,8 @@ public class DistributedWorkflowDefinitionNotificationsHandler(IBus bus) : INotificationHandler, INotificationHandler, INotificationHandler, - INotificationHandler + INotificationHandler, + INotificationHandler { /// public Task HandleAsync(WorkflowDefinitionPublished notification, CancellationToken cancellationToken) @@ -73,11 +73,23 @@ public class DistributedWorkflowDefinitionNotificationsHandler(IBus bus) : public Task HandleAsync(WorkflowDefinitionsRefreshed notification, CancellationToken cancellationToken) { // Prevent re-entrance. - if (AmbientConsumerScope.IsConsumerExecutionContext) + if (AmbientConsumerScope.IsWorkflowDefinitionEventsConsumer) return Task.CompletedTask; var definitionIds = notification.WorkflowDefinitionIds; var message = new Distributed.WorkflowDefinitionsRefreshed(definitionIds); return bus.Publish(message, cancellationToken); } + + /// + public Task HandleAsync(WorkflowDefinitionsReloaded notification, CancellationToken cancellationToken) + { + // Prevent re-entrance. + if (AmbientConsumerScope.IsWorkflowDefinitionEventsConsumer) + return Task.CompletedTask; + + var definitionIds = notification.WorkflowDefinitionIds; + var message = new Distributed.WorkflowDefinitionsReloaded(definitionIds); + return bus.Publish(message, cancellationToken); + } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionsReloaded.cs b/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionsReloaded.cs new file mode 100644 index 000000000..bba459390 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionsReloaded.cs @@ -0,0 +1,8 @@ +namespace Elsa.MassTransit.Messages; + +/// Represents a message that indicates that the specified workflow definitions have been reloaded. +public class WorkflowDefinitionsReloaded(ICollection workflowDefinitionIds) +{ + /// The workflow definition IDs that have been reloaded. + public ICollection WorkflowDefinitionIds { get; set; } = workflowDefinitionIds; +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Services/AmbientConsumerScope.cs b/src/modules/Elsa.MassTransit/Services/AmbientConsumerScope.cs index 2236cd969..55c585edc 100644 --- a/src/modules/Elsa.MassTransit/Services/AmbientConsumerScope.cs +++ b/src/modules/Elsa.MassTransit/Services/AmbientConsumerScope.cs @@ -6,7 +6,7 @@ public static class AmbientConsumerScope private static readonly AsyncLocal IsRaisedFromConsumerState = new(); /// Gets or sets a value that indicates if the current code is initiated from a consumer. - public static bool IsConsumerExecutionContext + public static bool IsWorkflowDefinitionEventsConsumer { get => IsRaisedFromConsumerState.Value; set => IsRaisedFromConsumerState.Value = value; diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs index 752102397..8acc4c5a9 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs @@ -1,6 +1,7 @@ using Elsa.Abstractions; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; using JetBrains.Annotations; namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Refresh; @@ -18,13 +19,13 @@ internal class Refresh(IWorkflowDefinitionsRefresher workflowDefinitionsRefreshe public override async Task HandleAsync(Request request, CancellationToken cancellationToken) { - await RefreshWorkflowDefinitionsAsync(request.DefinitionIds, cancellationToken); - await SendOkAsync(cancellationToken); + var result = await RefreshWorkflowDefinitionsAsync(request.DefinitionIds, cancellationToken); + await SendOkAsync(new Response(result.Refreshed, result.NotFound), cancellationToken); } - private async Task RefreshWorkflowDefinitionsAsync(ICollection? definitionIds, CancellationToken cancellationToken) + private async Task RefreshWorkflowDefinitionsAsync(ICollection? definitionIds, CancellationToken cancellationToken) { var request = new RefreshWorkflowDefinitionsRequest(definitionIds, BatchSize); - await workflowDefinitionsRefresher.RefreshWorkflowDefinitionsAsync(request, cancellationToken); + return await workflowDefinitionsRefresher.RefreshWorkflowDefinitionsAsync(request, cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Models.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Models.cs index c5970467b..f52d3e9e6 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Models.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Models.cs @@ -1,6 +1,14 @@ +using System.Text.Json.Serialization; + namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Refresh; -public class Request +internal class Request { public ICollection? DefinitionIds { get; set; } +} + +internal class Response(ICollection refreshed, ICollection notFound) +{ + [JsonPropertyName("refreshed")] public ICollection Refreshed { get; } = refreshed; + [JsonPropertyName("notFound")] public ICollection NotFound { get; } = notFound; } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Reload/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Reload/Endpoint.cs new file mode 100644 index 000000000..71d14f949 --- /dev/null +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Reload/Endpoint.cs @@ -0,0 +1,28 @@ +using Elsa.Abstractions; +using Elsa.Workflows.Runtime.Contracts; +using JetBrains.Annotations; + +namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Reload; + +[PublicAPI] +internal class Reload(IWorkflowDefinitionsReloader workflowDefinitionsReloader) : ElsaEndpointWithoutRequest +{ + private const int BatchSize = 10; + + public override void Configure() + { + Post("/actions/workflow-definitions/reload"); + ConfigurePermissions("actions:workflow-definitions:reload"); + } + + public override async Task HandleAsync(CancellationToken cancellationToken) + { + await ReloadWorkflowDefinitionsAsync(cancellationToken); + await SendOkAsync(cancellationToken); + } + + private async Task ReloadWorkflowDefinitionsAsync(CancellationToken cancellationToken) + { + await workflowDefinitionsReloader.ReloadWorkflowDefinitionsAsync(cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionStorePopulator.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionStorePopulator.cs index d9326097f..aeb5c198b 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionStorePopulator.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionStorePopulator.cs @@ -12,14 +12,14 @@ public interface IWorkflowDefinitionStorePopulator /// Populates the with workflow definitions provided from implementations. /// /// The cancellation token. - Task PopulateStoreAsync(CancellationToken cancellationToken = default); - + Task> PopulateStoreAsync(CancellationToken cancellationToken = default); + /// /// Populates the with workflow definitions provided from implementations. /// /// Whether to index triggers. /// The cancellation token. - Task PopulateStoreAsync(bool indexTriggers, CancellationToken cancellationToken = default); + Task> PopulateStoreAsync(bool indexTriggers, CancellationToken cancellationToken = default); /// /// Adds a workflow definition to the store. diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionsReloader.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionsReloader.cs new file mode 100644 index 000000000..e2cc2987e --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionsReloader.cs @@ -0,0 +1,8 @@ +namespace Elsa.Workflows.Runtime.Contracts; + +/// Reloads all workflows by re-invoking the populator. +public interface IWorkflowDefinitionsReloader +{ + /// Reloads all workflows by re-invoking the populator. + Task ReloadWorkflowDefinitionsAsync(CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index 314979966..75a41922e 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -221,6 +221,7 @@ public class WorkflowRuntimeFeature : FeatureBase .AddScoped() .AddScoped() .AddScoped() + .AddScoped() .AddScoped() .AddScoped() .AddScoped() diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateTriggersCache.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateTriggersCache.cs index e0347168a..d1581b49f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateTriggersCache.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateTriggersCache.cs @@ -15,11 +15,19 @@ namespace Elsa.Workflows.Runtime.Handlers; /// The cache is invalidated by calling the TriggerTokenAsync method of the ICacheManager passed to the class constructor. /// [UsedImplicitly] -public class InvalidateTriggersCache(ICacheManager cacheManager) : INotificationHandler +public class InvalidateTriggersCache(ICacheManager cacheManager) : + INotificationHandler, + INotificationHandler { /// public Task HandleAsync(WorkflowDefinitionsRefreshed notification, CancellationToken cancellationToken) { return cacheManager.TriggerTokenAsync(CachingTriggerStore.CacheInvalidationTokenKey, cancellationToken).AsTask(); } + + /// + public async Task HandleAsync(WorkflowDefinitionsReloaded notification, CancellationToken cancellationToken) + { + await cacheManager.TriggerTokenAsync(CachingTriggerStore.CacheInvalidationTokenKey, cancellationToken); + } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateWorkflowsCache.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateWorkflowsCache.cs new file mode 100644 index 000000000..fb6dc6d45 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateWorkflowsCache.cs @@ -0,0 +1,26 @@ +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Management.Contracts; +using Elsa.Workflows.Runtime.Notifications; +using JetBrains.Annotations; + +namespace Elsa.Workflows.Runtime.Handlers; + +/// +/// A notification handler that invalidates workflows cache when workflow definitions are reloaded. +/// +/// +/// The class implements the INotificationHandler interface and is responsible for handling WorkflowDefinitionsReloaded notifications. +/// When a WorkflowDefinitionsReloaded notification is received, the HandleAsync method is called to invalidate the http definition cache. +/// +[UsedImplicitly] +public class InvalidateWorkflowsCache(IWorkflowDefinitionCacheManager workflowDefinitionCacheManager) : INotificationHandler +{ + /// + public async Task HandleAsync(WorkflowDefinitionsReloaded notification, CancellationToken cancellationToken) + { + foreach (var workflowDefinitionId in notification.WorkflowDefinitionIds) + { + await workflowDefinitionCacheManager.EvictWorkflowDefinitionAsync(workflowDefinitionId, cancellationToken); + } + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowDefinitionsReloaded.cs b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowDefinitionsReloaded.cs new file mode 100644 index 000000000..882fa7a51 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowDefinitionsReloaded.cs @@ -0,0 +1,6 @@ +using Elsa.Mediator.Contracts; + +namespace Elsa.Workflows.Runtime.Notifications; + +/// Published when workflow definitions have been reloaded. +public record WorkflowDefinitionsReloaded(ICollection WorkflowDefinitionIds) : INotification; diff --git a/src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs b/src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs index 36cfc5360..ce380a1ec 100644 --- a/src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs +++ b/src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs @@ -3,4 +3,4 @@ using Elsa.Workflows.Management.Entities; namespace Elsa.Workflows.Runtime.Responses; /// Represents a response to a request to refresh workflow definitions. -public record RefreshWorkflowDefinitionsResponse(ICollection WorkflowDefinitions); \ No newline at end of file +public record RefreshWorkflowDefinitionsResponse(ICollection Refreshed, ICollection NotFound); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs index aa0208a7b..cba487b5a 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs @@ -49,21 +49,26 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP } /// - public Task PopulateStoreAsync(CancellationToken cancellationToken = default) + public Task> PopulateStoreAsync(CancellationToken cancellationToken = default) { return PopulateStoreAsync(true, cancellationToken); } /// - public async Task PopulateStoreAsync(bool indexTriggers, CancellationToken cancellationToken = default) + public async Task> PopulateStoreAsync(bool indexTriggers, CancellationToken cancellationToken = default) { var providers = _workflowDefinitionProviders(); + var workflowDefinitionIds = new List(); foreach (var provider in providers) { var results = await provider.GetWorkflowsAsync(cancellationToken).AsTask().ToList(); + workflowDefinitionIds.AddRange(results.Select(w => w.Workflow.Id)); + foreach (var result in results) await AddAsync(result, indexTriggers, cancellationToken); } + + return workflowDefinitionIds; } /// @@ -131,8 +136,8 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP if (existingDefinitionVersion != null) { workflowDefinitionsToSave.Add(existingDefinitionVersion); - - if(existingDefinitionVersion.Id != workflow.Identity.Id) + + if (existingDefinitionVersion.Id != workflow.Identity.Id) { // It's possible that the imported workflow definition has a different ID than the existing one in the store. // In a future update, we might store this discrepancy in a "troubleshooting" table and provide tooling for managing these, and other, discrepancies. diff --git a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionRefresher.cs b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionRefresher.cs index 39eb938e1..8c3f64fa1 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionRefresher.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionRefresher.cs @@ -39,12 +39,15 @@ public class WorkflowDefinitionsRefresher(IWorkflowDefinitionStore store, ITrigg await IndexWorkflowTriggersAsync(definitions.Items, cancellationToken); processedWorkflowDefinitions.AddRange(definitions.Items); currentPage++; + + if (definitions.Items.Count < batchSize) + break; } - var processedWorkflowDefinitionIds = processedWorkflowDefinitions.Select(x => x.Id).ToList(); + var processedWorkflowDefinitionIds = processedWorkflowDefinitions.Select(x => x.DefinitionId).ToList(); var notification = new WorkflowDefinitionsRefreshed(processedWorkflowDefinitionIds); await notificationSender.SendAsync(notification, cancellationToken); - return new RefreshWorkflowDefinitionsResponse(processedWorkflowDefinitions); + return new RefreshWorkflowDefinitionsResponse(processedWorkflowDefinitionIds, request.DefinitionIds?.Except(processedWorkflowDefinitionIds)?.ToList() ?? []); } private async Task IndexWorkflowTriggersAsync(IEnumerable definitions, CancellationToken cancellationToken) diff --git a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionsReloader.cs b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionsReloader.cs new file mode 100644 index 000000000..3afadafd2 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionsReloader.cs @@ -0,0 +1,17 @@ +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Notifications; + +namespace Elsa.Workflows.Runtime.Services; + +/// +public class WorkflowDefinitionsReloader(IWorkflowDefinitionStorePopulator workflowDefinitionStorePopulator, INotificationSender notificationSender) : IWorkflowDefinitionsReloader +{ + /// + public async Task ReloadWorkflowDefinitionsAsync(CancellationToken cancellationToken) + { + var definitionIds = await workflowDefinitionStorePopulator.PopulateStoreAsync(true, cancellationToken); + var notification = new WorkflowDefinitionsReloaded(definitionIds); + await notificationSender.SendAsync(notification, cancellationToken); + } +} \ No newline at end of file diff --git a/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj b/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj index 7d745b355..c6588808f 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj +++ b/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj @@ -21,6 +21,9 @@ + + Always + Always @@ -75,6 +78,9 @@ Always + + Always + diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/DynamicEndpointTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/DynamicEndpointTests.cs index f3b8e3df2..2be751964 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/DynamicEndpointTests.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/DynamicEndpointTests.cs @@ -1,4 +1,5 @@ -using Elsa.Workflows.ComponentTests.Helpers.Services; +using System.Net; +using Elsa.Workflows.ComponentTests.Helpers.Services; using Elsa.Workflows.Runtime.Contracts; using Microsoft.Extensions.DependencyInjection; @@ -15,22 +16,19 @@ public class DynamicEndpointTests : AppComponentTest } [Fact] - public async Task HelloWorldWorkflow_ShouldRespondWithHelloWorld() + public async Task ChangingEndpointValueThenRefresh_WorkflowShouldRespondToTheNewValue() { var client = WorkflowServer.CreateHttpWorkflowClient(); - var firstResponse = await client.GetStringAsync("first-value"); + var firstResponse = await client.SendAsync(new HttpRequestMessage(HttpMethod.Get, "first-value")); StaticValueHolder.Value = "second-value"; - await _workflowDefinitionsRefresher.RefreshWorkflowDefinitionsAsync(new Runtime.Requests.RefreshWorkflowDefinitionsRequest - { - BatchSize = 10, - DefinitionIds = ["f69f061159adc3ae"] - }, CancellationToken.None); + var _ = await _workflowDefinitionsRefresher.RefreshWorkflowDefinitionsAsync( + new Runtime.Requests.RefreshWorkflowDefinitionsRequest() { DefinitionIds = ["f69f061159adc3ae"] }, CancellationToken.None); - var secondResponse = await client.GetStringAsync("second-value"); + var secondResponse = await client.SendAsync(new HttpRequestMessage(HttpMethod.Get, "second-value")); - Assert.Equal("", firstResponse); - Assert.Equal("", secondResponse); + Assert.Equal(HttpStatusCode.OK, firstResponse.StatusCode); + Assert.Equal(HttpStatusCode.OK, secondResponse.StatusCode); } } \ No newline at end of file diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionReload/RemoveReloadWorkflowTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionReload/RemoveReloadWorkflowTests.cs new file mode 100644 index 000000000..390c58d14 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionReload/RemoveReloadWorkflowTests.cs @@ -0,0 +1,35 @@ +using System.Net; +using Elsa.Workflows.Management.Contracts; +using Elsa.Workflows.Runtime.Contracts; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Workflows.ComponentTests.Scenarios.WorkflowDefinitionReload; + +public class RemoveReloadWorkflowTests : AppComponentTest +{ + private readonly IWorkflowDefinitionManager _workflowDefinitionManager; + private readonly IWorkflowDefinitionsReloader _workflowDefinitionsReloader; + + public RemoveReloadWorkflowTests(App app) : base(app) + { + _workflowDefinitionManager = Scope.ServiceProvider.GetRequiredService(); + _workflowDefinitionsReloader = Scope.ServiceProvider.GetRequiredService(); + } + + [Fact] + public async Task RemovingTheWorkflowThenReload_WorkflowShouldBeReachableAgain() + { + var client = WorkflowServer.CreateHttpWorkflowClient(); + + var result = await _workflowDefinitionManager.DeleteByDefinitionIdAsync("f68b09bc-2013-4617-b82f-d76b6819a624", CancellationToken.None); + + var firstResponse = await client.SendAsync(new HttpRequestMessage(HttpMethod.Get, "reload-test")); + + await _workflowDefinitionsReloader.ReloadWorkflowDefinitionsAsync(CancellationToken.None); + + var secondResponse = await client.SendAsync(new HttpRequestMessage(HttpMethod.Get, "reload-test")); + + Assert.Equal(HttpStatusCode.NotFound, firstResponse.StatusCode); + Assert.Equal(HttpStatusCode.OK, secondResponse.StatusCode); + } +} \ No newline at end of file diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionReload/http-workflow.json b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionReload/http-workflow.json new file mode 100644 index 000000000..c885f29b6 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionReload/http-workflow.json @@ -0,0 +1,104 @@ +{ + "id": "c2e19917-d03e-41ed-9d19-ef5c1ddc9871", + "definitionId": "f68b09bc-2013-4617-b82f-d76b6819a624", + "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": "f8019bae-c506-4467-812b-e403e7aad8ff", + "nodeId": "Workflow1:f8019bae-c506-4467-812b-e403e7aad8ff", + "metadata": {}, + "customProperties": { + "source": "FlowchartJsonConverter.cs:45", + "notFoundConnections": [], + "canStartWorkflow": false, + "runAsynchronously": false + }, + "activities": [ + { + "path": { + "typeName": "String", + "expression": { + "type": "Literal", + "value": "reload-test" + } + }, + "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": "cc196a5e-04f7-4794-b25c-1a0987632d8a", + "nodeId": "Workflow1:f8019bae-c506-4467-812b-e403e7aad8ff:cc196a5e-04f7-4794-b25c-1a0987632d8a", + "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": [] + } +} \ No newline at end of file