From 7c313325296adcf15efde0a41bcef03e82ef635b Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Thu, 19 Sep 2024 16:18:56 +0200 Subject: [PATCH] Fix variable serialization (#5974) * Improve dispatched workflow input handling Addressed input handling in dispatch messages by adding `SerializedInput` property. Also removed initialization logic and moved input deserialization to a helper method, ensuring compatibility with both new and deprecated input property formats. * Update workflow Docker images and version tags Changed Docker image tags from v3-2-0-rc3 to v3-2-1-preview across multiple GitHub workflows. Updated the VERSION environment variable in packages.yml to reflect the new versioning scheme. These changes ensure consistency with the new preview release. * Update versioning to include 'preview' in package workflow Modified the workflow to append 'preview' to the version number for non-tagged builds. This ensures clearer differentiation between stable and non-stable versions in the CI pipeline. * Add WorkflowInstanceStorageDriver for workflow variable storage Introduced a new storage driver, WorkflowInstanceStorageDriver, to store workflow variables directly in the workflow state. Updated relevant classes and methods to incorporate this new storage driver, ensuring seamless read/write/delete operations and extending support for it throughout the codebase. * Refactor object conversion and update variable retrieval. Switched from JsonObject to JsonNode for object conversion and corrected a typo in the summary comment. Changed the return type of GetVariablesDictionary method and updated its implementation to use VariablesDictionary. * Rename 'input' to 'serializedInput' in DispatchWorkflowDefinition. This change clarifies that the input provided to the workflow should be serialized. It enhances the readability and accuracy of the code documentation, ensuring that developers understand the expected format of the input parameter. * Add priority and deprecation attributes to storage drivers Introduced a priority attribute to the `IStorageDriver` interface and implemented it in various storage drivers. Additionally, marked `WorkflowStorageDriver` as deprecated and reordered storage driver listing based on priority. * Switch MassTransit broker to in-memory and refactor converter Changed MassTransit broker from AzureServiceBus to in-memory for improved performance in development environment. Simplified PolymorphicObjectConverterFactory by removing redundant constructor and dependencies. Removed unused folder from the project file. --- .github/workflows/elsa-server-and-studio.yml | 2 +- .github/workflows/elsa-server.yml | 2 +- .github/workflows/elsa-studio.yml | 2 +- .github/workflows/packages.yml | 2 +- Elsa.sln.DotSettings | 1 + .../Models/StorageDriverDescriptor.cs | 2 +- .../Helpers/ObjectConverter.cs | 6 +- .../Features/AzureServiceBusFeature.cs | 8 +++ .../Features/RabbitMqServiceBusFeature.cs | 9 ++- .../DispatchWorkflowRequestConsumer.cs | 46 ++++++++------- .../Messages/DispatchResumeWorkflows.cs | 4 ++ .../DispatchTriggerWorkflowsRequest.cs | 4 ++ .../Messages/DispatchWorkflowDefinition.cs | 15 +++-- .../Messages/DispatchWorkflowInstance.cs | 4 ++ .../Services/MassTransitWorkflowDispatcher.cs | 10 +++- .../Endpoints/StorageDrivers/List/Endpoint.cs | 6 +- .../Endpoints/StorageDrivers/List/Models.cs | 2 +- .../Activities/ParallelForEachT.cs | 4 +- .../Contracts/IStorageDriver.cs | 5 ++ .../ExpressionExecutionContextExtensions.cs | 2 +- .../Extensions/VariableExtensions.cs | 8 +-- .../Features/WorkflowsFeature.cs | 1 + .../PolymorphicObjectConverterFactory.cs | 4 +- .../Services/MemoryStorageDriver.cs | 2 + .../Services/WorkflowInstanceStorageDriver.cs | 58 +++++++++++++++++++ .../Services/WorkflowStateExtractor.cs | 2 +- .../Services/WorkflowStorageDriver.cs | 4 ++ .../Activities/BulkDispatchWorkflows.cs | 2 +- .../DefaultBackgroundActivityInvoker.cs | 2 +- .../Variables/CountdownWorkflowTests.cs | 6 +- 30 files changed, 171 insertions(+), 54 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Core/Services/WorkflowInstanceStorageDriver.cs diff --git a/.github/workflows/elsa-server-and-studio.yml b/.github/workflows/elsa-server-and-studio.yml index 46417543a..8f58dbcc2 100644 --- a/.github/workflows/elsa-server-and-studio.yml +++ b/.github/workflows/elsa-server-and-studio.yml @@ -29,7 +29,7 @@ jobs: with: # list of Docker images to use as base name for tags images: | - elsaworkflows/elsa-server-and-studio-v3-2-0-rc3 + elsaworkflows/elsa-server-and-studio-v3-2-1-preview flavor: | latest=true # generate Docker tags based on the following events/attributes diff --git a/.github/workflows/elsa-server.yml b/.github/workflows/elsa-server.yml index 7c50ce0c8..82a92a31b 100644 --- a/.github/workflows/elsa-server.yml +++ b/.github/workflows/elsa-server.yml @@ -29,7 +29,7 @@ jobs: with: # list of Docker images to use as base name for tags images: | - elsaworkflows/elsa-server-v3-2-0-rc3 + elsaworkflows/elsa-server-v3-2-1-preview flavor: | latest=true # generate Docker tags based on the following events/attributes diff --git a/.github/workflows/elsa-studio.yml b/.github/workflows/elsa-studio.yml index 35c8eb51a..a8673a0dd 100644 --- a/.github/workflows/elsa-studio.yml +++ b/.github/workflows/elsa-studio.yml @@ -29,7 +29,7 @@ jobs: with: # list of Docker images to use as base name for tags images: | - elsaworkflows/elsa-studio-v3-2-0-rc3 + elsaworkflows/elsa-studio-v3-2-1-preview flavor: | latest=true # generate Docker tags based on the following events/attributes diff --git a/.github/workflows/packages.yml b/.github/workflows/packages.yml index 549e11d6e..c4461693c 100644 --- a/.github/workflows/packages.yml +++ b/.github/workflows/packages.yml @@ -62,7 +62,7 @@ jobs: TAG_NAME=${TAG_NAME#refs/tags/} # remove the refs/tags/ prefix echo "VERSION=${TAG_NAME}" >> $GITHUB_ENV else - echo "VERSION=3.2.0-rc6.${{github.run_number}}" >> $GITHUB_ENV + echo "VERSION=3.2.1-preview.${{github.run_number}}" >> $GITHUB_ENV fi - name: Set up JDK 17 uses: actions/setup-java@v2 diff --git a/Elsa.sln.DotSettings b/Elsa.sln.DotSettings index b291186ee..37f0ea98b 100644 --- a/Elsa.sln.DotSettings +++ b/Elsa.sln.DotSettings @@ -14,6 +14,7 @@ True True True + True True True True diff --git a/src/clients/Elsa.Api.Client/Resources/StorageDrivers/Models/StorageDriverDescriptor.cs b/src/clients/Elsa.Api.Client/Resources/StorageDrivers/Models/StorageDriverDescriptor.cs index e0bab6929..835893b38 100644 --- a/src/clients/Elsa.Api.Client/Resources/StorageDrivers/Models/StorageDriverDescriptor.cs +++ b/src/clients/Elsa.Api.Client/Resources/StorageDrivers/Models/StorageDriverDescriptor.cs @@ -5,4 +5,4 @@ namespace Elsa.Api.Client.Resources.StorageDrivers.Models; /// /// The type name of the storage driver. /// The display name of the storage driver. -public record StorageDriverDescriptor(string TypeName, string DisplayName); \ No newline at end of file +public record StorageDriverDescriptor(string TypeName, string DisplayName, double Priority = 0, bool Deprecated = false); \ No newline at end of file diff --git a/src/modules/Elsa.Expressions/Helpers/ObjectConverter.cs b/src/modules/Elsa.Expressions/Helpers/ObjectConverter.cs index a66d344e6..26a1a01bf 100644 --- a/src/modules/Elsa.Expressions/Helpers/ObjectConverter.cs +++ b/src/modules/Elsa.Expressions/Helpers/ObjectConverter.cs @@ -99,13 +99,13 @@ public static class ObjectConverter return jsonElement.Deserialize(targetType, serializerOptions); } - if (value is JsonObject jsonObject) + if (value is JsonNode jsonObject) { return underlyingTargetType switch { { } t when t == typeof(string) => jsonObject.ToString(), { } t when t != typeof(object) => jsonObject.Deserialize(targetType, serializerOptions), - _ => jsonObject, + _ => jsonObject }; } @@ -240,7 +240,7 @@ public static class ObjectConverter } /// - /// Returns true if the specified type is date-like type, false otherwise. + /// Returns true if the specified type is a date-like type, false otherwise. /// private static bool IsDateType(Type type) { diff --git a/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs b/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs index 0fc93e992..3ba3cc824 100644 --- a/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs +++ b/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs @@ -1,4 +1,5 @@ using Azure.Messaging.ServiceBus.Administration; +using Elsa.Common.Contracts; using Elsa.Extensions; using Elsa.Features.Abstractions; using Elsa.Features.Attributes; @@ -121,6 +122,13 @@ public class AzureServiceBusFeature : FeatureBase } configurator.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); + + configurator.ConfigureJsonSerializerOptions(serializerOptions => + { + var serializer = context.GetRequiredService(); + serializer.ApplyOptions(serializerOptions); + return serializerOptions; + }); }); }; }); diff --git a/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs b/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs index f8b443fdf..3359fffb0 100644 --- a/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs +++ b/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs @@ -1,10 +1,10 @@ +using Elsa.Common.Contracts; using Elsa.Extensions; using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; using Elsa.Hosting.Management.Contracts; using Elsa.Hosting.Management.Features; -using Elsa.MassTransit.Consumers; using Elsa.MassTransit.Extensions; using Elsa.MassTransit.Features; using Elsa.MassTransit.Options; @@ -88,6 +88,13 @@ public class RabbitMqServiceBusFeature : FeatureBase } configurator.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); + + configurator.ConfigureJsonSerializerOptions(serializerOptions => + { + var serializer = context.GetRequiredService(); + serializer.ApplyOptions(serializerOptions); + return serializerOptions; + }); }); }; }); diff --git a/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs b/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs index dbcd4c2db..511a5d5ee 100644 --- a/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs +++ b/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs @@ -1,5 +1,5 @@ using Elsa.MassTransit.Messages; -using Elsa.Workflows.Management.Contracts; +using Elsa.Workflows.Contracts; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Parameters; @@ -12,22 +12,12 @@ namespace Elsa.MassTransit.Consumers; /// A consumer of various dispatch message types to asynchronously execute workflows. /// [UsedImplicitly] -public class DispatchWorkflowRequestConsumer : +public class DispatchWorkflowRequestConsumer(IWorkflowRuntime workflowRuntime, IPayloadSerializer jsonSerializer) : IConsumer, IConsumer, IConsumer, IConsumer { - private readonly IWorkflowRuntime _workflowRuntime; - - /// - /// Initializes a new instance of the class. - /// - public DispatchWorkflowRequestConsumer(IWorkflowRuntime workflowRuntime, IWorkflowInstanceManager workflowInstanceManager) - { - _workflowRuntime = workflowRuntime; - } - /// public async Task Consume(ConsumeContext context) { @@ -42,6 +32,7 @@ public class DispatchWorkflowRequestConsumer : { var message = context.Message; var cancellationToken = context.CancellationToken; + var input = message.Input ?? DeserializeInput(message.SerializedInput); var options = new ResumeWorkflowRuntimeParams { @@ -51,12 +42,12 @@ public class DispatchWorkflowRequestConsumer : ActivityNodeId = message.ActivityNodeId, ActivityInstanceId = message.ActivityInstanceId, ActivityHash = message.ActivityHash, - Input = message.Input, + Input = input, Properties = message.Properties, CancellationTokens = cancellationToken }; - await _workflowRuntime.ResumeWorkflowAsync(message.InstanceId, options); + await workflowRuntime.ResumeWorkflowAsync(message.InstanceId, options); } /// @@ -64,16 +55,17 @@ public class DispatchWorkflowRequestConsumer : { var message = context.Message; var cancellationToken = context.CancellationToken; + var input = message.Input ?? DeserializeInput(message.SerializedInput); var options = new TriggerWorkflowsOptions { CorrelationId = message.CorrelationId, WorkflowInstanceId = message.WorkflowInstanceId, ActivityInstanceId = message.ActivityInstanceId, - Input = message.Input, + Input = input, Properties = message.Properties, CancellationTokens = cancellationToken }; - await _workflowRuntime.TriggerWorkflowsAsync(message.ActivityTypeName, message.BookmarkPayload, options); + await workflowRuntime.TriggerWorkflowsAsync(message.ActivityTypeName, message.BookmarkPayload, options); } /// @@ -81,29 +73,31 @@ public class DispatchWorkflowRequestConsumer : { var message = context.Message; var cancellationToken = context.CancellationToken; + var input = message.Input ?? DeserializeInput(message.SerializedInput); var options = new TriggerWorkflowsOptions { CorrelationId = message.CorrelationId, WorkflowInstanceId = message.WorkflowInstanceId, - Input = message.Input, + Input = input, Properties = message.Properties, CancellationTokens = cancellationToken }; - await _workflowRuntime.ResumeWorkflowsAsync(message.ActivityTypeName, message.BookmarkPayload, options); + await workflowRuntime.ResumeWorkflowsAsync(message.ActivityTypeName, message.BookmarkPayload, options); } - + private async Task DispatchNewWorkflowInstanceAsync(DispatchWorkflowDefinition message, CancellationToken cancellationToken) { if (string.IsNullOrWhiteSpace(message.DefinitionId)) throw new ArgumentException("The definition ID is required when dispatching a workflow definition."); if (message.VersionOptions == null) throw new ArgumentException("The version options are required when dispatching a workflow definition."); + var input = message.Input ?? DeserializeInput(message.SerializedInput); var options = new StartWorkflowRuntimeParams { ParentWorkflowInstanceId = message.ParentWorkflowInstanceId, CorrelationId = message.CorrelationId, - Input = message.Input, + Input = input, Properties = message.Properties, VersionOptions = message.VersionOptions.Value, TriggerActivityId = message.TriggerActivityId, @@ -111,7 +105,7 @@ public class DispatchWorkflowRequestConsumer : CancellationTokens = cancellationToken }; - await _workflowRuntime.TryStartWorkflowAsync(message.DefinitionId, options); + await workflowRuntime.TryStartWorkflowAsync(message.DefinitionId, options); } private async Task DispatchExistingWorkflowInstanceAsync(DispatchWorkflowDefinition message, CancellationToken cancellationToken) @@ -126,6 +120,14 @@ public class DispatchWorkflowRequestConsumer : CancellationTokens = cancellationToken }; - await _workflowRuntime.StartWorkflowAsync(message.InstanceId, options); + await workflowRuntime.StartWorkflowAsync(message.InstanceId, options); + } + + private IDictionary? DeserializeInput(string? json) + { + if (string.IsNullOrWhiteSpace(json)) + return null; + + return jsonSerializer.Deserialize>(json); } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Messages/DispatchResumeWorkflows.cs b/src/modules/Elsa.MassTransit/Messages/DispatchResumeWorkflows.cs index 9d059d277..1ec3b68de 100644 --- a/src/modules/Elsa.MassTransit/Messages/DispatchResumeWorkflows.cs +++ b/src/modules/Elsa.MassTransit/Messages/DispatchResumeWorkflows.cs @@ -14,6 +14,10 @@ public class DispatchResumeWorkflows(string activityTypeName, object bookmarkPay public string? CorrelationId { get; set; } public string? WorkflowInstanceId { get; set; } public string? ActivityInstanceId { get; set; } + + [Obsolete("This property is no longer used and will be removed in a future version. Use the SerializedInput property instead.")] public IDictionary? Input { get; set; } + + public string? SerializedInput { get; set; } public IDictionary? Properties { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Messages/DispatchTriggerWorkflowsRequest.cs b/src/modules/Elsa.MassTransit/Messages/DispatchTriggerWorkflowsRequest.cs index 36a140b13..71456ecca 100644 --- a/src/modules/Elsa.MassTransit/Messages/DispatchTriggerWorkflowsRequest.cs +++ b/src/modules/Elsa.MassTransit/Messages/DispatchTriggerWorkflowsRequest.cs @@ -14,6 +14,10 @@ public class DispatchTriggerWorkflows(string activityTypeName, object bookmarkPa public string? CorrelationId { get; set; } public string? WorkflowInstanceId { get; set; } public string? ActivityInstanceId { get; set; } + + [Obsolete("This property is no longer used and will be removed in a future version. Use the SerializedInput property instead.")] public IDictionary? Input { get; set; } + + public string? SerializedInput { get; set; } public IDictionary? Properties { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Messages/DispatchWorkflowDefinition.cs b/src/modules/Elsa.MassTransit/Messages/DispatchWorkflowDefinition.cs index 22f6348b0..a5fc979ac 100644 --- a/src/modules/Elsa.MassTransit/Messages/DispatchWorkflowDefinition.cs +++ b/src/modules/Elsa.MassTransit/Messages/DispatchWorkflowDefinition.cs @@ -1,4 +1,3 @@ -using System.Text.Json.Serialization; using Elsa.Common.Models; namespace Elsa.MassTransit.Messages; @@ -27,7 +26,7 @@ public record DispatchWorkflowDefinition /// The ID of the workflow definition to dispatch. /// The version options to use when dispatching the workflow definition. /// The ID of the parent workflow instance. - /// Any input to pass to the workflow. + /// Any input to pass to the workflow. /// Any properties to attach to the workflow. /// A correlation ID to associate the workflow with. /// The ID to use when creating an instance of the workflow to dispatch. @@ -36,7 +35,7 @@ public record DispatchWorkflowDefinition string? definitionId, VersionOptions? versionOptions, string? parentWorkflowInstanceId, - IDictionary? input, + string? serializedInput, IDictionary? properties, string? correlationId, string? instanceId, @@ -47,7 +46,7 @@ public record DispatchWorkflowDefinition DefinitionId = definitionId, VersionOptions = versionOptions, ParentWorkflowInstanceId = parentWorkflowInstanceId, - Input = input, + SerializedInput = serializedInput, Properties = properties, CorrelationId = correlationId, InstanceId = instanceId, @@ -64,8 +63,14 @@ public record DispatchWorkflowDefinition /// The ID of the parent workflow instance. public string? ParentWorkflowInstanceId { get; init; } + /// Deprecated. Use the property instead. + [Obsolete("This property is no longer used and will be removed in a future version. Use the SerializedInput property instead.")] + public IDictionary? Input { get; set; } + + /// /// Any input to pass to the workflow. - public IDictionary? Input { get; init; } + /// + public string? SerializedInput { get; set; } /// Any properties to attach to the workflow. public IDictionary? Properties { get; init; } diff --git a/src/modules/Elsa.MassTransit/Messages/DispatchWorkflowInstance.cs b/src/modules/Elsa.MassTransit/Messages/DispatchWorkflowInstance.cs index 212d70a89..04bc4935c 100644 --- a/src/modules/Elsa.MassTransit/Messages/DispatchWorkflowInstance.cs +++ b/src/modules/Elsa.MassTransit/Messages/DispatchWorkflowInstance.cs @@ -8,7 +8,11 @@ public class DispatchWorkflowInstance(string instanceId) public string? ActivityNodeId { get; set; } public string? ActivityInstanceId { get; set; } public string? ActivityHash { get; set; } + + [Obsolete("This property is no longer used and will be removed in a future version. Use the SerializedInput property instead.")] public IDictionary? Input { get; set; } + + public string? SerializedInput { get; set; } public IDictionary? Properties { get; set; } public string? CorrelationId { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs index a3870d3d3..edf665e58 100644 --- a/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs +++ b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs @@ -11,7 +11,6 @@ using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; using Elsa.Workflows.Runtime.Responses; using MassTransit; -using Medallion.Threading; using Microsoft.Extensions.Logging; namespace Elsa.MassTransit.Services; @@ -27,6 +26,7 @@ public class MassTransitWorkflowDispatcher( IBookmarkHasher bookmarkHasher, ITriggerStore triggerStore, IBookmarkStore bookmarkStore, + IPayloadSerializer jsonSerializer, ILogger logger) : IWorkflowDispatcher { @@ -57,6 +57,7 @@ public class MassTransitWorkflowDispatcher( public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) { var sendEndpoint = await GetSendEndpointAsync(options); + var serializedInput = SerializeInput(request.Input); await sendEndpoint.Send(new DispatchWorkflowInstance(request.InstanceId) { @@ -66,7 +67,7 @@ public class MassTransitWorkflowDispatcher( ActivityInstanceId = request.ActivityInstanceId, ActivityHash = request.ActivityHash, CorrelationId = request.CorrelationId, - Input = request.Input + SerializedInput = serializedInput, }, cancellationToken); return DispatchWorkflowResponse.Success(); } @@ -180,4 +181,9 @@ public class MassTransitWorkflowDispatcher( var sendEndpoint = await bus.GetSendEndpoint(new Uri($"queue:{endpointName}")); return sendEndpoint; } + + private string? SerializeInput(object? input) + { + return input != null ? jsonSerializer.Serialize(input) : null; + } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/StorageDrivers/List/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/StorageDrivers/List/Endpoint.cs index a38030ea5..207501e21 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/StorageDrivers/List/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/StorageDrivers/List/Endpoint.cs @@ -31,7 +31,7 @@ public class List : ElsaEndpointWithoutRequest public override Task ExecuteAsync(CancellationToken ct) { var drivers = _registry.List(); - var descriptors = drivers.Select(FromDriver).ToList(); + var descriptors = drivers.Select(FromDriver).OrderByDescending(x => x.Priority).ToList(); var response = new Response(descriptors); return Task.FromResult(response); @@ -40,7 +40,9 @@ public class List : ElsaEndpointWithoutRequest private static StorageDriverDescriptor FromDriver(IStorageDriver driver) { var type = driver.GetType(); + var deprecated = type.GetCustomAttribute() != null; var displayName = type.GetCustomAttribute()?.Name ?? type.GetCustomAttribute()?.DisplayName ?? type.Name.Replace("StorageDriver", ""); - return new StorageDriverDescriptor(type.GetSimpleAssemblyQualifiedName(), displayName); + var priority = driver.Priority; + return new StorageDriverDescriptor(type.GetSimpleAssemblyQualifiedName(), displayName, priority, deprecated); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/StorageDrivers/List/Models.cs b/src/modules/Elsa.Workflows.Api/Endpoints/StorageDrivers/List/Models.cs index 127a1ce68..c52464164 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/StorageDrivers/List/Models.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/StorageDrivers/List/Models.cs @@ -5,4 +5,4 @@ public class Response(ICollection items) public ICollection Items { get; set; } = items; } -public record StorageDriverDescriptor(string TypeName, string DisplayName); \ No newline at end of file +public record StorageDriverDescriptor(string TypeName, string DisplayName, double Priority = 0, bool Deprecated = false); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/ParallelForEachT.cs b/src/modules/Elsa.Workflows.Core/Activities/ParallelForEachT.cs index d5d8fcd63..3c3ab86da 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/ParallelForEachT.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/ParallelForEachT.cs @@ -53,10 +53,10 @@ public class ParallelForEach : Activity var currentValueVariable = new Variable("CurrentValue", item) { // TODO: This should be configurable, because this won't work for e.g. file streams and other non-serializable types. - StorageDriverType = typeof(WorkflowStorageDriver) + StorageDriverType = typeof(WorkflowInstanceStorageDriver) }; - var currentIndexVariable = new Variable("CurrentIndex", currentIndex++) { StorageDriverType = typeof(WorkflowStorageDriver) }; + var currentIndexVariable = new Variable("CurrentIndex", currentIndex++) { StorageDriverType = typeof(WorkflowInstanceStorageDriver) }; var variables = new List { currentValueVariable, currentIndexVariable }; // Schedule a body of work for each item. diff --git a/src/modules/Elsa.Workflows.Core/Contracts/IStorageDriver.cs b/src/modules/Elsa.Workflows.Core/Contracts/IStorageDriver.cs index d17ab4735..50ca0eec2 100644 --- a/src/modules/Elsa.Workflows.Core/Contracts/IStorageDriver.cs +++ b/src/modules/Elsa.Workflows.Core/Contracts/IStorageDriver.cs @@ -5,6 +5,11 @@ namespace Elsa.Workflows.Contracts; /// public interface IStorageDriver { + /// + /// The priority of the storage driver. Drivers with higher priority are used before drivers with lower priority. + /// + double Priority { get; } + /// /// Writes a value to the storage driver. /// diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ExpressionExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/ExpressionExecutionContextExtensions.cs index 5759d88d7..7bb7d2f97 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ExpressionExecutionContextExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ExpressionExecutionContextExtensions.cs @@ -141,7 +141,7 @@ public static class ExpressionExecutionContextExtensions var variable = new Variable(name, value) { - StorageDriverType = storageDriverType ?? typeof(WorkflowStorageDriver) + StorageDriverType = storageDriverType ?? typeof(WorkflowInstanceStorageDriver) }; // Find the first parent context that has a variable container. diff --git a/src/modules/Elsa.Workflows.Core/Extensions/VariableExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/VariableExtensions.cs index d40876542..85f242ceb 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/VariableExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/VariableExtensions.cs @@ -31,14 +31,14 @@ public static class VariableExtensions new ExpandoObjectConverterFactory()); /// - /// Configures the variable to use the . + /// Configures the variable to use the . /// - public static Variable WithWorkflowStorage(this Variable variable) => variable.WithStorage(); + public static Variable WithWorkflowStorage(this Variable variable) => variable.WithStorage(); /// - /// Configures the variable to use the . + /// Configures the variable to use the . /// - public static Variable WithWorkflowStorage(this Variable variable) => (Variable)variable.WithStorage(); + public static Variable WithWorkflowStorage(this Variable variable) => (Variable)variable.WithStorage(); /// /// Configures the variable to use the . diff --git a/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs b/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs index 1dc698cf0..1026cb7c2 100644 --- a/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs +++ b/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs @@ -171,6 +171,7 @@ public class WorkflowsFeature : FeatureBase // Storage drivers. .AddScoped() .AddStorageDriver() + .AddStorageDriver() .AddStorageDriver() // Serialization. diff --git a/src/modules/Elsa.Workflows.Core/Serialization/Converters/PolymorphicObjectConverterFactory.cs b/src/modules/Elsa.Workflows.Core/Serialization/Converters/PolymorphicObjectConverterFactory.cs index d50c29b88..448fc16be 100644 --- a/src/modules/Elsa.Workflows.Core/Serialization/Converters/PolymorphicObjectConverterFactory.cs +++ b/src/modules/Elsa.Workflows.Core/Serialization/Converters/PolymorphicObjectConverterFactory.cs @@ -2,6 +2,7 @@ using System.Dynamic; using System.Text.Json; using System.Text.Json.Serialization; using Elsa.Expressions.Contracts; +using Elsa.Expressions.Services; namespace Elsa.Workflows.Serialization.Converters; @@ -16,7 +17,8 @@ public class PolymorphicObjectConverterFactory(IWellKnownTypeRegistry wellKnownT var canConvert = typeToConvert.IsClass && typeToConvert == typeof(object) || typeToConvert == typeof(ExpandoObject) - || typeToConvert == typeof(Dictionary); + || typeToConvert == typeof(Dictionary) + || typeToConvert == typeof(IDictionary); return canConvert; } diff --git a/src/modules/Elsa.Workflows.Core/Services/MemoryStorageDriver.cs b/src/modules/Elsa.Workflows.Core/Services/MemoryStorageDriver.cs index 485e0715f..e04642ddf 100644 --- a/src/modules/Elsa.Workflows.Core/Services/MemoryStorageDriver.cs +++ b/src/modules/Elsa.Workflows.Core/Services/MemoryStorageDriver.cs @@ -11,6 +11,8 @@ public class MemoryStorageDriver : IStorageDriver { private readonly IDictionary _dictionary = new Dictionary(); + public double Priority => 0; + /// public ValueTask WriteAsync(string id, object value, StorageDriverContext context) { diff --git a/src/modules/Elsa.Workflows.Core/Services/WorkflowInstanceStorageDriver.cs b/src/modules/Elsa.Workflows.Core/Services/WorkflowInstanceStorageDriver.cs new file mode 100644 index 000000000..b6e7fd7ae --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Services/WorkflowInstanceStorageDriver.cs @@ -0,0 +1,58 @@ +using System.ComponentModel.DataAnnotations; +using System.Text.Json; +using System.Text.Json.Nodes; +using Elsa.Extensions; +using Elsa.Workflows.Contracts; +using JetBrains.Annotations; + +namespace Elsa.Workflows.Services; + +/// A storage driver that stores objects in the workflow state itself. +[Display(Name = "Workflow Instance")] +[UsedImplicitly] +public class WorkflowInstanceStorageDriver : IStorageDriver +{ + /// The key used to store the variables in the workflow state. + public const string VariablesDictionaryStateKey = "Variables"; + + /// + public double Priority => 1; + + /// + public ValueTask WriteAsync(string id, object value, StorageDriverContext context) + { + UpdateVariablesDictionary(context, dictionary => + { + var node = JsonSerializer.SerializeToNode(value); + dictionary[id] = node; + }); + return ValueTask.CompletedTask; + } + + /// + public ValueTask ReadAsync(string id, StorageDriverContext context) + { + var dictionary = GetVariablesDictionary(context); + var node = dictionary.GetValueOrDefault(id); + return new(node); + } + + /// + public ValueTask DeleteAsync(string id, StorageDriverContext context) + { + UpdateVariablesDictionary(context, dictionary => dictionary.Remove(id)); + return ValueTask.CompletedTask; + } + + private VariablesDictionary GetVariablesDictionary(StorageDriverContext context) => context.ExecutionContext.Properties.GetOrAdd(VariablesDictionaryStateKey, () => new VariablesDictionary()); + private void SetVariablesDictionary(StorageDriverContext context, VariablesDictionary dictionary) => context.ExecutionContext.Properties[VariablesDictionaryStateKey] = dictionary; + + private void UpdateVariablesDictionary(StorageDriverContext context, Action update) + { + var dictionary = GetVariablesDictionary(context); + update(dictionary); + SetVariablesDictionary(context, dictionary); + } +} + +public class VariablesDictionary : Dictionary; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs b/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs index 009079217..26c70baf8 100644 --- a/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs +++ b/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs @@ -71,7 +71,7 @@ public class WorkflowStateExtractor : IWorkflowStateExtractor private IDictionary GetPersistableInput(WorkflowExecutionContext workflowExecutionContext) { // TODO: This is a temporary solution. We need to find a better way to handle this. - var persistableInput = workflowExecutionContext.Workflow.Inputs.Where(x => x.StorageDriverType == typeof(WorkflowStorageDriver)).ToList(); + var persistableInput = workflowExecutionContext.Workflow.Inputs.Where(x => x.StorageDriverType == typeof(WorkflowStorageDriver) || x.StorageDriverType == typeof(WorkflowInstanceStorageDriver)).ToList(); var input = workflowExecutionContext.Input; var filteredInput = new Dictionary(); diff --git a/src/modules/Elsa.Workflows.Core/Services/WorkflowStorageDriver.cs b/src/modules/Elsa.Workflows.Core/Services/WorkflowStorageDriver.cs index 7cbf7ac8d..b2ee7e8cf 100644 --- a/src/modules/Elsa.Workflows.Core/Services/WorkflowStorageDriver.cs +++ b/src/modules/Elsa.Workflows.Core/Services/WorkflowStorageDriver.cs @@ -8,6 +8,7 @@ namespace Elsa.Workflows.Services; /// A storage driver that stores objects in the workflow state itself. /// [Display(Name = "Workflow")] +[Obsolete("This is no longer used and will be removed in a future version. Use the WorkflowInstanceStorageDriver instead.")] public class WorkflowStorageDriver : IStorageDriver { /// @@ -15,6 +16,9 @@ public class WorkflowStorageDriver : IStorageDriver /// public const string VariablesDictionaryStateKey = "PersistentVariablesDictionary"; + /// + public double Priority => -1; + /// public ValueTask WriteAsync(string id, object value, StorageDriverContext context) { diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs index b04f1b296..06eb5306a 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs @@ -237,7 +237,7 @@ public class BulkDispatchWorkflows : Activity var childInstanceId = new Variable("ChildInstanceId", workflowInstanceId) { - StorageDriverType = typeof(WorkflowStorageDriver) + StorageDriverType = typeof(WorkflowInstanceStorageDriver) }; var variables = new List diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs index 40e5518b2..1141abbbd 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs @@ -97,7 +97,7 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker var driver = variableMetadata?.StorageDriverType; // We only capture output written to the workflow itself. Other drivers like blob storage, etc. will be ignored since the foreground context will be loading those. - if (driver != typeof(WorkflowStorageDriver)) + if (driver != typeof(WorkflowStorageDriver) && driver != typeof(WorkflowInstanceStorageDriver)) continue; var outputValue = activityExecutionContext.Get(memoryBlockReference); diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Variables/CountdownWorkflowTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Variables/CountdownWorkflowTests.cs index b0f3f7b9b..a0067ea35 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Variables/CountdownWorkflowTests.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Variables/CountdownWorkflowTests.cs @@ -48,6 +48,8 @@ public class CountdownWorkflowTests(App app) : AppComponentTest(app) } } - private IDictionary GetVariablesDictionary(ActivityExecutionContextState context) => - context.Properties.GetOrAdd(WorkflowStorageDriver.VariablesDictionaryStateKey, () => new Dictionary()); + private VariablesDictionary GetVariablesDictionary(ActivityExecutionContextState context) + { + return context.Properties.GetOrAdd(WorkflowInstanceStorageDriver.VariablesDictionaryStateKey, () => new VariablesDictionary()); + } } \ No newline at end of file