diff --git a/src/modules/Elsa.Expressions/Models/ExpressionExecutionContext.cs b/src/modules/Elsa.Expressions/Models/ExpressionExecutionContext.cs index fc505fee3..901706e84 100644 --- a/src/modules/Elsa.Expressions/Models/ExpressionExecutionContext.cs +++ b/src/modules/Elsa.Expressions/Models/ExpressionExecutionContext.cs @@ -43,5 +43,5 @@ public class ExpressionExecutionContext public T GetRequiredService() where T : notnull => _serviceProvider.GetRequiredService(); //private MemoryDatum? GetMemoryDatumInternal(MemoryDatumReference locationReference) => MemoryRegister.TryGetMemoryDatum(locationReference.Id, out var location) ? location : ParentContext?.GetMemoryDatumInternal(locationReference); - private MemoryBlock? GetMemoryDatumInternal(MemoryReference reference) => Memory.TryGetMemoryDatum(reference.Id, out var location) ? location : default; + private MemoryBlock? GetMemoryDatumInternal(MemoryReference reference) => Memory.TryGetBlock(reference.Id, out var location) ? location : default; } \ No newline at end of file diff --git a/src/modules/Elsa.Expressions/Models/MemoryReference.cs b/src/modules/Elsa.Expressions/Models/MemoryReference.cs index a7f77a150..537cb7ae8 100644 --- a/src/modules/Elsa.Expressions/Models/MemoryReference.cs +++ b/src/modules/Elsa.Expressions/Models/MemoryReference.cs @@ -13,11 +13,11 @@ public abstract class MemoryReference public string Id { get; set; } = default!; public abstract MemoryBlock Declare(); - public object? Get(MemoryRegister memoryRegister) => GetLocation(memoryRegister).Value; + public object? Get(MemoryRegister memoryRegister) => GetBlock(memoryRegister).Value; public T? Get(MemoryRegister memoryRegister) => (T?)Get(memoryRegister); public object? Get(ExpressionExecutionContext context) => context.Get(this); public T? Get(ExpressionExecutionContext context) => (T?)Get(context); - public void Set(MemoryRegister memoryRegister, object? value) => GetLocation(memoryRegister).Value = value; + public void Set(MemoryRegister memoryRegister, object? value) => GetBlock(memoryRegister).Value = value; public void Set(ExpressionExecutionContext context, object? value) => context.Set(this, value); - public MemoryBlock GetLocation(MemoryRegister memoryRegister) => memoryRegister.TryGetMemoryDatum(Id, out var location) ? location : memoryRegister.Declare(this); + public MemoryBlock GetBlock(MemoryRegister memoryRegister) => memoryRegister.TryGetBlock(Id, out var location) ? location : memoryRegister.Declare(this); } \ No newline at end of file diff --git a/src/modules/Elsa.Expressions/Models/MemoryRegister.cs b/src/modules/Elsa.Expressions/Models/MemoryRegister.cs index 1fb285310..eb363f972 100644 --- a/src/modules/Elsa.Expressions/Models/MemoryRegister.cs +++ b/src/modules/Elsa.Expressions/Models/MemoryRegister.cs @@ -14,14 +14,14 @@ public class MemoryRegister public MemoryRegister? Parent { get; } public IDictionary Blocks { get; } - public bool TryGetMemoryDatum(string id, out MemoryBlock datum) + public bool TryGetBlock(string id, out MemoryBlock datum) { datum = null!; if (Blocks.TryGetValue(id, out datum!)) return true; - return Parent?.TryGetMemoryDatum(id, out datum) == true; + return Parent?.TryGetBlock(id, out datum) == true; } public void Declare(IEnumerable references) diff --git a/src/modules/Elsa.JavaScript/Handlers/ConfigureJavaScriptEngineWithActivityOutput.cs b/src/modules/Elsa.JavaScript/Handlers/ConfigureJavaScriptEngineWithActivityOutput.cs index 16a59524b..703bf9e23 100644 --- a/src/modules/Elsa.JavaScript/Handlers/ConfigureJavaScriptEngineWithActivityOutput.cs +++ b/src/modules/Elsa.JavaScript/Handlers/ConfigureJavaScriptEngineWithActivityOutput.cs @@ -32,7 +32,7 @@ public class ConfigureJavaScriptEngineWithActivityOutput : INotificationHandler< foreach (var property in properties) { - if (register.TryGetMemoryDatum(property.Value.MemoryReference.Id, out var location)) + if (register.TryGetBlock(property.Value.MemoryReference.Id, out var location)) jsActivity[property.Name] = location.Value; } diff --git a/src/modules/Elsa.Workflows.Core/Activities/ForEach.cs b/src/modules/Elsa.Workflows.Core/Activities/ForEach.cs index 8da143f71..e48aa3733 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/ForEach.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/ForEach.cs @@ -18,6 +18,11 @@ public class ForEach : Activity Behaviors.Remove(); } + public ForEach(ICollection items) : this() + { + Items = new Input>(items); + } + /// /// The set of values to iterate. /// diff --git a/src/modules/Elsa.Workflows.Core/Behaviors/ExecutionLoggingBehavior.cs b/src/modules/Elsa.Workflows.Core/Behaviors/ExecutionLoggingBehavior.cs index 0f0948bdd..097772d2c 100644 --- a/src/modules/Elsa.Workflows.Core/Behaviors/ExecutionLoggingBehavior.cs +++ b/src/modules/Elsa.Workflows.Core/Behaviors/ExecutionLoggingBehavior.cs @@ -1,4 +1,3 @@ -using System.Runtime.CompilerServices; using Elsa.Workflows.Core.Models; using Elsa.Workflows.Core.Services; using Elsa.Workflows.Core.Signals; diff --git a/src/modules/Elsa.Workflows.Core/Behaviors/ScheduledChildCallbackBehavior.cs b/src/modules/Elsa.Workflows.Core/Behaviors/ScheduledChildCallbackBehavior.cs index 9e70d0e38..d0de87f60 100644 --- a/src/modules/Elsa.Workflows.Core/Behaviors/ScheduledChildCallbackBehavior.cs +++ b/src/modules/Elsa.Workflows.Core/Behaviors/ScheduledChildCallbackBehavior.cs @@ -24,6 +24,8 @@ public class ScheduledChildCallbackBehavior : Behavior if (callbackEntry == null) return; + // Before invoking the parent activity, make sure its properties are evaluated. + await activityExecutionContext.EvaluateInputPropertiesAsync(); await callbackEntry(activityExecutionContext, childActivityExecutionContext); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Builders/WorkflowDefinitionBuilder.cs b/src/modules/Elsa.Workflows.Core/Builders/WorkflowDefinitionBuilder.cs index d4056ba27..fb052fe0f 100644 --- a/src/modules/Elsa.Workflows.Core/Builders/WorkflowDefinitionBuilder.cs +++ b/src/modules/Elsa.Workflows.Core/Builders/WorkflowDefinitionBuilder.cs @@ -39,6 +39,32 @@ public class WorkflowDefinitionBuilder : IWorkflowDefinitionBuilder return this; } + public Variable WithVariable(string? driveId = default) + { + var variable = new Variable() + { + DriveId = driveId + }; + Variables.Add(variable); + return variable; + } + + public Variable WithVariable(string name, T value, string? driveId = default) + { + var variable = value != null ? new Variable(name, value) : new Variable(name); + variable.DriveId = driveId; + Variables.Add(variable); + return variable; + } + + public Variable WithVariable(T value, string? driveId = default) + { + var variable = value != null ? new Variable(value) : new Variable(); + variable.DriveId = driveId; + Variables.Add(variable); + return variable; + } + public IWorkflowDefinitionBuilder WithVariable(Variable variable) { Variables.Add(variable); diff --git a/src/modules/Elsa.Workflows.Core/DataDriveNames.cs b/src/modules/Elsa.Workflows.Core/DataDriveNames.cs new file mode 100644 index 000000000..8f8034a41 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/DataDriveNames.cs @@ -0,0 +1,7 @@ +namespace Elsa.Workflows.Core; + +public static class DataDriveNames +{ + public const string Workflow = "Workflow"; + public const string Memory = "Memory"; +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Extensions/DependencyInjectionExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/DependencyInjectionExtensions.cs index c59125510..972583176 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/DependencyInjectionExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/DependencyInjectionExtensions.cs @@ -1,5 +1,7 @@ using Elsa.Features.Services; using Elsa.Workflows.Core.Features; +using Elsa.Workflows.Core.Services; +using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Core; @@ -10,4 +12,9 @@ public static class DependencyInjectionExtensions configuration.Configure(configure); return configuration; } + + public static IServiceCollection AddDataDrive(this IServiceCollection services) where T : class, IDataDrive + { + return services.AddSingleton(); + } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Extensions/DictionaryExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/DictionaryExtensions.cs index 979e8c566..0b6bf91e5 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/DictionaryExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/DictionaryExtensions.cs @@ -22,7 +22,7 @@ public static class DictionaryExtensions public static T? GetValue(this IDictionary dictionary, TKey key) => ConvertValue(dictionary[key]); public static T? GetValue(this IDictionary dictionary, string key) => ConvertValue(dictionary[key]); - public static T? GetOrAdd(this IDictionary dictionary, string key, Func valueFactory) where T : notnull => dictionary.GetOrAdd(key, valueFactory); + public static T GetOrAdd(this IDictionary dictionary, string key, Func valueFactory) where T : notnull => dictionary.GetOrAdd(key, valueFactory); public static T GetOrAdd(this IDictionary dictionary, TKey key, Func valueFactory) { diff --git a/src/modules/Elsa.Workflows.Core/Extensions/VariableExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/VariableExtensions.cs new file mode 100644 index 000000000..e82eff7b2 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Extensions/VariableExtensions.cs @@ -0,0 +1,15 @@ +using Elsa.Workflows.Core.Models; + +namespace Elsa.Workflows.Core; + +public static class VariableExtensions +{ + public static Variable WithWorkflowDrive(this Variable variable) => variable.WithDrive(DataDriveNames.Workflow); + public static Variable WithMemoryDrive(this Variable variable) => variable.WithDrive(DataDriveNames.Memory); + + public static Variable WithDrive(this Variable variable, string driveId) + { + variable.DriveId = driveId; + return variable; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs index 9b1cd4d15..7165044eb 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs @@ -69,7 +69,7 @@ public static class WorkflowExecutionContextExtensions /// Gets the specified workflow variable by name. /// public static T? GetVariable(this WorkflowExecutionContext workflowExecutionContext, string name) => (T?)workflowExecutionContext.GetVariable(name); - + /// /// Gets the specified workflow variable by name, where the name is implied by the type name. /// @@ -88,7 +88,7 @@ public static class WorkflowExecutionContextExtensions /// Sets the specified workflow variable by name, where the name is implied by the type name. /// public static Variable SetVariable(this WorkflowExecutionContext workflowExecutionContext, T? value) => workflowExecutionContext.SetVariable(typeof(T).Name, value); - + /// /// Sets the specified workflow variable by name. /// @@ -99,7 +99,14 @@ public static class WorkflowExecutionContextExtensions /// public static Variable SetVariable(this WorkflowExecutionContext workflowExecutionContext, string name, object? value) { - var variable = workflowExecutionContext.Workflow.Variables.FirstOrDefault(x => x.Name == name) ?? new Variable(name, value); + var variable = workflowExecutionContext.Workflow.Variables.FirstOrDefault(x => x.Name == name); + + if (variable == null) + { + variable = new Variable(name, value); + workflowExecutionContext.Workflow.Variables.Add(variable); + } + variable.Set(workflowExecutionContext.MemoryRegister, value); return variable; } diff --git a/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs b/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs index e87c8c2eb..a76520108 100644 --- a/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs +++ b/src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs @@ -79,6 +79,11 @@ public class WorkflowsFeature : FeatureBase // Stream providers. .AddSingleton(StandardInStreamProvider) .AddSingleton(StandardOutStreamProvider) + + // Data drives. + .AddSingleton() + .AddDataDrive() + .AddDataDrive() // Logging .AddLogging(); diff --git a/src/modules/Elsa.Workflows.Core/Implementations/DataDriveManager.cs b/src/modules/Elsa.Workflows.Core/Implementations/DataDriveManager.cs new file mode 100644 index 000000000..3794773c0 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Implementations/DataDriveManager.cs @@ -0,0 +1,10 @@ +using Elsa.Workflows.Core.Services; + +namespace Elsa.Workflows.Core.Implementations; + +public class DataDriveManager : IDataDriveManager +{ + private readonly IEnumerable _drives; + public DataDriveManager(IEnumerable drives) => _drives = drives; + public IDataDrive? GetDriveById(string id) => _drives.FirstOrDefault(x => x.Id == id); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Implementations/MemoryDataDrive.cs b/src/modules/Elsa.Workflows.Core/Implementations/MemoryDataDrive.cs new file mode 100644 index 000000000..d6108036e --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Implementations/MemoryDataDrive.cs @@ -0,0 +1,27 @@ +using Elsa.Workflows.Core.Services; + +namespace Elsa.Workflows.Core.Implementations; + +public class MemoryDataDrive : IDataDrive +{ + private readonly IDictionary _dictionary = new Dictionary(); + public string Id => DataDriveNames.Memory; + + public ValueTask WriteAsync(string id, object value, DataDriveContext context) + { + _dictionary[id] = value; + return ValueTask.CompletedTask; + } + + public ValueTask ReadAsync(string id, DataDriveContext context) + { + var value = _dictionary.TryGetValue(id, out var v) ? v : default; + return new (value); + } + + public ValueTask DeleteAsync(string id, DataDriveContext context) + { + _dictionary.Remove(id); + return ValueTask.CompletedTask; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Implementations/WorkflowRunner.cs b/src/modules/Elsa.Workflows.Core/Implementations/WorkflowRunner.cs index a93437740..6fca7da2d 100644 --- a/src/modules/Elsa.Workflows.Core/Implementations/WorkflowRunner.cs +++ b/src/modules/Elsa.Workflows.Core/Implementations/WorkflowRunner.cs @@ -84,7 +84,7 @@ public class WorkflowRunner : IWorkflowRunner await _pipeline.ExecuteAsync(workflowExecutionContext); // Extract workflow state. - var workflowState = _workflowStateSerializer.ReadState(workflowExecutionContext); + var workflowState = _workflowStateSerializer.SerializeState(workflowExecutionContext); // Return workflow execution result containing state + bookmarks. return new InvokeWorkflowResult(workflowState, workflowExecutionContext.Bookmarks); @@ -116,7 +116,7 @@ public class WorkflowRunner : IWorkflowRunner var workflowExecutionContext = new WorkflowExecutionContext(serviceProvider, id, correlationId, workflow, graph, scheduler, bookmark, input, executeActivityDelegate, cancellationToken); // Restore workflow execution context from state, if provided. - if (workflowState != null) _workflowStateSerializer.WriteState(workflowExecutionContext, workflowState); + if (workflowState != null) _workflowStateSerializer.DeserializeState(workflowExecutionContext, workflowState); return workflowExecutionContext; } diff --git a/src/modules/Elsa.Workflows.Core/Implementations/WorkflowStateDataDrive.cs b/src/modules/Elsa.Workflows.Core/Implementations/WorkflowStateDataDrive.cs new file mode 100644 index 000000000..28c69b58b --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Implementations/WorkflowStateDataDrive.cs @@ -0,0 +1,35 @@ +using Elsa.Workflows.Core.Services; + +namespace Elsa.Workflows.Core.Implementations; + +/// +/// A data drive that stores objects in the workflow state itself. +/// +public class WorkflowStateDataDrive : IDataDrive +{ + public const string VariablesDictionaryStateKey = "PersistentVariablesDictionary"; + public string Id => DataDriveNames.Workflow; + + public ValueTask WriteAsync(string id, object value, DataDriveContext context) + { + var dictionary = GetVariablesDictionary(context); + dictionary[id] = value; + return ValueTask.CompletedTask; + } + + public ValueTask ReadAsync(string id, DataDriveContext context) + { + var dictionary = GetVariablesDictionary(context); + var value = dictionary.TryGetValue(id, out var v) ? v : default; + return new(value); + } + + public ValueTask DeleteAsync(string id, DataDriveContext context) + { + var dictionary = GetVariablesDictionary(context); + dictionary.Remove(id); + return ValueTask.CompletedTask; + } + + private IDictionary GetVariablesDictionary(DataDriveContext context) => context.WorkflowState.Properties.GetOrAdd(VariablesDictionaryStateKey, () => new Dictionary()); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Implementations/WorkflowStateSerializer.cs b/src/modules/Elsa.Workflows.Core/Implementations/WorkflowStateSerializer.cs index 0168d3039..f0b3f0e1a 100644 --- a/src/modules/Elsa.Workflows.Core/Implementations/WorkflowStateSerializer.cs +++ b/src/modules/Elsa.Workflows.Core/Implementations/WorkflowStateSerializer.cs @@ -16,7 +16,7 @@ public class WorkflowStateSerializer : IWorkflowStateSerializer _serviceProvider = serviceProvider; } - public WorkflowState ReadState(WorkflowExecutionContext workflowExecutionContext) + public WorkflowState SerializeState(WorkflowExecutionContext workflowExecutionContext) { var state = new WorkflowState { @@ -26,31 +26,31 @@ public class WorkflowStateSerializer : IWorkflowStateSerializer SubStatus = workflowExecutionContext.SubStatus }; - //GetOutput(state, workflowExecutionContext); - GetProperties(state, workflowExecutionContext); - GetCompletionCallbacks(state, workflowExecutionContext); - GetActivityExecutionContexts(state, workflowExecutionContext); + SerializeProperties(state, workflowExecutionContext); + SerializeCompletionCallbacks(state, workflowExecutionContext); + SerializeActivityExecutionContexts(state, workflowExecutionContext); + SerializePersistentVariables(state, workflowExecutionContext); return state; } - public void WriteState(WorkflowExecutionContext workflowExecutionContext, WorkflowState state) + public void DeserializeState(WorkflowExecutionContext workflowExecutionContext, WorkflowState state) { workflowExecutionContext.Id = state.Id; workflowExecutionContext.CorrelationId = state.CorrelationId; - workflowExecutionContext.SubStatus = state.SubStatus; - //SetOutput(state, workflowExecutionContext); - SetProperties(state, workflowExecutionContext); - SetActivityExecutionContexts(state, workflowExecutionContext); - SetCompletionCallbacks(state, workflowExecutionContext); + workflowExecutionContext.SubStatus = state.SubStatus; + DeserializeProperties(state, workflowExecutionContext); + DeserializeActivityExecutionContexts(state, workflowExecutionContext); + DeserializeCompletionCallbacks(state, workflowExecutionContext); + //DeserializePersistentVariables(state, workflowExecutionContext); } - private void GetProperties(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) + private void SerializeProperties(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) { state.Properties = workflowExecutionContext.Properties; } - private void SetProperties(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) + private void DeserializeProperties(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) { workflowExecutionContext.Properties = state.Properties; } @@ -87,7 +87,7 @@ public class WorkflowStateSerializer : IWorkflowStateSerializer } } - private void SetCompletionCallbacks(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) + private void DeserializeCompletionCallbacks(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) { foreach (var completionCallbackEntry in state.CompletionCallbacks) { @@ -99,13 +99,13 @@ public class WorkflowStateSerializer : IWorkflowStateSerializer } } - private void GetCompletionCallbacks(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) + private void SerializeCompletionCallbacks(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) { var completionCallbacks = workflowExecutionContext.CompletionCallbacks.Select(x => new CompletionCallbackState(x.Owner.Id, x.Child.Id, x.CompletionCallback.Method.Name)); state.CompletionCallbacks = completionCallbacks.ToList(); } - private void GetActivityExecutionContexts(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) + private void SerializeActivityExecutionContexts(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) { ActivityExecutionContextState CreateActivityExecutionContextState(ActivityExecutionContext activityExecutionContext) { @@ -125,7 +125,7 @@ public class WorkflowStateSerializer : IWorkflowStateSerializer state.ActivityExecutionContexts = workflowExecutionContext.ActivityExecutionContexts.Reverse().Select(CreateActivityExecutionContextState).ToList(); } - private void SetActivityExecutionContexts(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) + private void DeserializeActivityExecutionContexts(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) { ActivityExecutionContext CreateActivityExecutionContext(ActivityExecutionContextState activityExecutionContextState) { @@ -163,6 +163,32 @@ public class WorkflowStateSerializer : IWorkflowStateSerializer workflowExecutionContext.ActivityExecutionContexts = activityExecutionContexts; } + private void SerializePersistentVariables(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) + { + var workflow = workflowExecutionContext.Workflow; + + state.PersistentVariables = workflow.Variables + .Where(x => x.DriveId != null) + .Select(x => new PersistentVariableState(x.Name, x.DriveId!)) + .ToList(); + } + + // private void DeserializePersistentVariables(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) + // { + // var persistentVariables = state.PersistentVariables; + // var workflow = workflowExecutionContext.Workflow; + // + // foreach (var persistentVariable in persistentVariables) + // { + // var variable = workflow.Variables.FirstOrDefault(x => x.Name == persistentVariable.Name); + // + // if (variable == null) + // continue; + // + // variable.Set(workflowExecutionContext.MemoryRegister, variable.Value); + // } + // } + private IDictionary GetOutputFrom(ActivityNode activityNode) => activityNode.GetType().GetProperties(BindingFlags.Public).Where(x => x.GetCustomAttribute() != null).ToDictionary(x => x.Name, x => (object?)x.GetValue(activityNode)); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Models/ActivityExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Models/ActivityExecutionContext.cs index 381cbb759..43c1f18a1 100644 --- a/src/modules/Elsa.Workflows.Core/Models/ActivityExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Models/ActivityExecutionContext.cs @@ -146,19 +146,19 @@ public class ActivityExecutionContext public object? GetService(Type serviceType) => WorkflowExecutionContext.GetService(serviceType); public T? Get(Input? input) => input == null ? default : Get(input.MemoryReference); - public object? Get(MemoryReference locationReference) + public object? Get(MemoryReference reference) { - var location = GetLocation(locationReference) ?? throw new InvalidOperationException($"No location found with ID {locationReference.Id}. Did you forget to declare a variable with a container?"); + var location = GetBlock(reference) ?? throw new InvalidOperationException($"No location found with ID {reference.Id}. Did you forget to declare a variable with a container?"); return location.Value; } - public T? Get(MemoryReference locationReference) + public T? Get(MemoryReference reference) { - var value = Get(locationReference); + var value = Get(reference); return value != default ? (T?)(value) : default; } - public void Set(MemoryReference locationReference, object? value) => ExpressionExecutionContext.Set(locationReference, value); + public void Set(MemoryReference reference, object? value) => ExpressionExecutionContext.Set(reference, value); public void Set(Output output, object? value) => ExpressionExecutionContext.Set(output, value); public void Set(Output output, T value) => ExpressionExecutionContext.Set(output, value); @@ -176,8 +176,8 @@ public class ActivityExecutionContext WorkflowExecutionContext.RemoveCompletionCallbacks(entriesToRemove); } - private MemoryBlock? GetLocation(MemoryReference locationReference) => - ExpressionExecutionContext.Memory.TryGetMemoryDatum(locationReference.Id, out var location) + private MemoryBlock? GetBlock(MemoryReference locationReference) => + ExpressionExecutionContext.Memory.TryGetBlock(locationReference.Id, out var location) ? location - : ParentActivityExecutionContext?.GetLocation(locationReference); + : ParentActivityExecutionContext?.GetBlock(locationReference); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Models/DataDriveDefinition.cs b/src/modules/Elsa.Workflows.Core/Models/DataDriveDefinition.cs new file mode 100644 index 000000000..42b656c72 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Models/DataDriveDefinition.cs @@ -0,0 +1,25 @@ +using System.Text.Json.Serialization; + +namespace Elsa.Workflows.Core.Models; + +/// +/// Represents a workflow-specific memory drive definition that is configured with a specific storage driver. +/// +public class DataDriveDefinition +{ + [JsonConstructor] + public DataDriveDefinition() + { + } + + public DataDriveDefinition(string id, string driverId, string displayName) + { + Id = id; + DriverId = driverId; + DisplayName = displayName; + } + + public string Id { get; set; } = default!; + public string DriverId { get; set; } = default!; + public string DisplayName { get; set; } = default!; +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Models/Variable.cs b/src/modules/Elsa.Workflows.Core/Models/Variable.cs index e8dfd2bcf..0ee288153 100644 --- a/src/modules/Elsa.Workflows.Core/Models/Variable.cs +++ b/src/modules/Elsa.Workflows.Core/Models/Variable.cs @@ -25,7 +25,14 @@ public class Variable : MemoryReference get => Id; set => Id = value; } + public object? Value { get; set; } + + /// + /// When specified, the variable's value will be stored on the specified data drive. + /// + public string? DriveId { get; set; } + public override MemoryBlock Declare() => new(Value); } diff --git a/src/modules/Elsa.Workflows.Core/Models/Workflow.cs b/src/modules/Elsa.Workflows.Core/Models/Workflow.cs index 3e57dcfda..4950a7db4 100644 --- a/src/modules/Elsa.Workflows.Core/Models/Workflow.cs +++ b/src/modules/Elsa.Workflows.Core/Models/Workflow.cs @@ -47,6 +47,15 @@ public class Workflow : Composite, ICloneable public WorkflowPublication Publication { get; set; } public WorkflowMetadata WorkflowMetadata { get; set; } public ICollection Variables { get; init; } + + /// + /// A list of storage drive definitions that can be used by variables to persist their values. + /// + /// + /// Depending on the drive's driver capabilities, only serializable variables can be persisted. + /// + public ICollection Drives { get; set; } = new List(); + public Workflow Clone() => (Workflow)((ICloneable)this).Clone(); object ICloneable.Clone() => MemberwiseClone(); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Services/IDataDrive.cs b/src/modules/Elsa.Workflows.Core/Services/IDataDrive.cs new file mode 100644 index 000000000..abd81a31a --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Services/IDataDrive.cs @@ -0,0 +1,16 @@ +using Elsa.Workflows.Core.State; + +namespace Elsa.Workflows.Core.Services; + +/// +/// Represents a storage driver for workflows to write state to. +/// +public interface IDataDrive +{ + string Id { get; } + ValueTask WriteAsync(string id, object value, DataDriveContext context); + ValueTask ReadAsync(string id, DataDriveContext context); + ValueTask DeleteAsync(string id, DataDriveContext context); +} + +public record DataDriveContext(WorkflowState WorkflowState, CancellationToken CancellationToken); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Services/IDataDriveManager.cs b/src/modules/Elsa.Workflows.Core/Services/IDataDriveManager.cs new file mode 100644 index 000000000..b1dab05fd --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Services/IDataDriveManager.cs @@ -0,0 +1,6 @@ +namespace Elsa.Workflows.Core.Services; + +public interface IDataDriveManager +{ + IDataDrive? GetDriveById(string id); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Services/IWorkflowDefinitionBuilder.cs b/src/modules/Elsa.Workflows.Core/Services/IWorkflowDefinitionBuilder.cs index 4ba48b777..49958874b 100644 --- a/src/modules/Elsa.Workflows.Core/Services/IWorkflowDefinitionBuilder.cs +++ b/src/modules/Elsa.Workflows.Core/Services/IWorkflowDefinitionBuilder.cs @@ -12,6 +12,9 @@ public interface IWorkflowDefinitionBuilder IWorkflowDefinitionBuilder WithDefinitionId(string definitionId); IWorkflowDefinitionBuilder WithVersion(int version); IWorkflowDefinitionBuilder WithRoot(IActivity root); + Variable WithVariable(string? driveId = default); + Variable WithVariable(string name, T value, string? driveId = default); + Variable WithVariable(T value, string? driveId = default); IWorkflowDefinitionBuilder WithVariable(Variable variable); IWorkflowDefinitionBuilder WithVariables(params Variable[] variables); IWorkflowDefinitionBuilder WithApplicationProperty(string name, object value); diff --git a/src/modules/Elsa.Workflows.Core/Services/IWorkflowStateSerializer.cs b/src/modules/Elsa.Workflows.Core/Services/IWorkflowStateSerializer.cs index 34631791f..562647af5 100644 --- a/src/modules/Elsa.Workflows.Core/Services/IWorkflowStateSerializer.cs +++ b/src/modules/Elsa.Workflows.Core/Services/IWorkflowStateSerializer.cs @@ -5,6 +5,6 @@ namespace Elsa.Workflows.Core.Services; public interface IWorkflowStateSerializer { - WorkflowState ReadState(WorkflowExecutionContext workflowExecutionContext); - void WriteState(WorkflowExecutionContext workflowExecutionContext, WorkflowState state); + WorkflowState SerializeState(WorkflowExecutionContext workflowExecutionContext); + void DeserializeState(WorkflowExecutionContext workflowExecutionContext, WorkflowState state); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/State/PersistentVariableState.cs b/src/modules/Elsa.Workflows.Core/State/PersistentVariableState.cs new file mode 100644 index 000000000..ce6bb8502 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/State/PersistentVariableState.cs @@ -0,0 +1,20 @@ +using System.Text.Json.Serialization; + +namespace Elsa.Workflows.Core.State; + +public class PersistentVariableState +{ + [JsonConstructor] + public PersistentVariableState() + { + } + + public PersistentVariableState(string name, string driveId) + { + Name = name; + DriveId = driveId; + } + + public string Name { get; set; } = default!; + public string DriveId { get; set; } = default!; +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/State/WorkflowState.cs b/src/modules/Elsa.Workflows.Core/State/WorkflowState.cs index 80a6875bf..797b06c92 100644 --- a/src/modules/Elsa.Workflows.Core/State/WorkflowState.cs +++ b/src/modules/Elsa.Workflows.Core/State/WorkflowState.cs @@ -30,7 +30,7 @@ public class WorkflowState /// /// A dictionary of activity output. /// - public IDictionary> ActivityOutput { get; set; } = new Dictionary>(); + public IDictionary> ActivityOutput { get; set; } = new Dictionary>(); /// /// A list of callbacks that activities registered in order to be notified when the activities they scheduled complete. @@ -45,5 +45,10 @@ public class WorkflowState /// /// A global property bag that contains properties set by application code and/or activities. /// - public IDictionary Properties { get; set; } = new Dictionary(); + public IDictionary Properties { get; set; } = new Dictionary(); + + /// + /// A list of variables that can be persisted. + /// + public ICollection PersistentVariables { get; set; } = new List(); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Middleware/PersistWorkflowInstanceMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/Middleware/PersistWorkflowInstanceMiddleware.cs index 37d85002c..58122248c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Middleware/PersistWorkflowInstanceMiddleware.cs +++ b/src/modules/Elsa.Workflows.Runtime/Middleware/PersistWorkflowInstanceMiddleware.cs @@ -25,6 +25,7 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware private readonly IRequestSender _requestSender; private readonly IEventPublisher _eventPublisher; private readonly IWorkflowStateSerializer _workflowStateSerializer; + private readonly IDataDriveManager _dataDriveManager; private readonly IBookmarkManager _bookmarkManager; private readonly IIdentityGenerator _identityGenerator; private readonly ISystemClock _clock; @@ -37,6 +38,7 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware IEventPublisher eventPublisher, IBookmarkManager bookmarkManager, IWorkflowStateSerializer workflowStateSerializer, + IDataDriveManager dataDriveManager, IIdentityGenerator identityGenerator, ISystemClock clock) : base(next) { @@ -46,6 +48,7 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware _eventPublisher = eventPublisher; _bookmarkManager = bookmarkManager; _workflowStateSerializer = workflowStateSerializer; + _dataDriveManager = dataDriveManager; _identityGenerator = identityGenerator; _clock = clock; } @@ -74,7 +77,7 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware Status = WorkflowStatus.Running, SubStatus = WorkflowSubStatus.Executing, CorrelationId = _identityGenerator.GenerateId(), - WorkflowState = _workflowStateSerializer.ReadState(context), + WorkflowState = _workflowStateSerializer.SerializeState(context), Name = workflowInstanceName }; @@ -95,12 +98,26 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware // Persist workflow instance. await _workflowInstanceStore.SaveAsync(workflowInstance, cancellationToken); + + // Load persistent variables. + var dataDriveContext = new DataDriveContext(workflowInstance.WorkflowState, cancellationToken); + + foreach (var variableState in workflowInstance.WorkflowState.PersistentVariables) + { + var drive = _dataDriveManager.GetDriveById(variableState.DriveId); + if (drive == null) continue; + var id = $"{context.Id}:{variableState.Name}"; + var value = await drive.ReadAsync(id, dataDriveContext); + if (value == null) continue; + var variable = new Variable(variableState.Name, value); + context.MemoryRegister.Declare(variable); + } // Invoke next middleware. await Next(context); // Update workflow instance. - var workflowState = _workflowStateSerializer.ReadState(context); + var workflowState = _workflowStateSerializer.SerializeState(context); workflowInstance.WorkflowState = workflowState; workflowInstance.Status = workflowState.Status; workflowInstance.SubStatus = workflowState.SubStatus; @@ -128,6 +145,19 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware if (context.TransientProperties.TryGetValue(WorkflowInstanceNameKey, out name)) workflowInstance.Name = (string?)name; + // Persist variables. + dataDriveContext = new DataDriveContext(workflowState, cancellationToken); + + foreach (var variableState in workflowState.PersistentVariables) + { + var drive = _dataDriveManager.GetDriveById(variableState.DriveId); + if (drive == null) continue; + if (!context.MemoryRegister.TryGetBlock(variableState.Name, out var block)) continue; + if (block.Value == null) continue; + var id = $"{context.Id}:{variableState.Name}"; + await drive.WriteAsync(id, block.Value, dataDriveContext); + } + // Persist updated workflow instance. await _workflowInstanceStore.SaveAsync(workflowInstance, cancellationToken); diff --git a/test/Elsa.IntegrationTests/Scenarios/PersistentVariables/Tests.cs b/test/Elsa.IntegrationTests/Scenarios/PersistentVariables/Tests.cs new file mode 100644 index 000000000..7ad8196c2 --- /dev/null +++ b/test/Elsa.IntegrationTests/Scenarios/PersistentVariables/Tests.cs @@ -0,0 +1,89 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Elsa.Extensions; +using Elsa.Testing.Shared; +using Elsa.Workflows.Core; +using Elsa.Workflows.Core.Builders; +using Elsa.Workflows.Core.Implementations; +using Elsa.Workflows.Core.Models; +using Elsa.Workflows.Core.Pipelines.WorkflowExecution.Components; +using Elsa.Workflows.Core.Services; +using Elsa.Workflows.Core.State; +using Elsa.Workflows.Runtime.Extensions; +using Microsoft.Extensions.DependencyInjection; +using Xunit; +using Xunit.Abstractions; + +namespace Elsa.IntegrationTests.Scenarios.PersistentVariables; + +public class WorkflowInstancePersistenceTests +{ + private readonly IWorkflowRunner _workflowRunner; + private readonly CapturingTextWriter _capturingTextWriter = new(); + + public WorkflowInstancePersistenceTests(ITestOutputHelper testOutputHelper) + { + var services = new TestApplicationBuilder(testOutputHelper).WithCapturingTextWriter(_capturingTextWriter).Build(); + _workflowRunner = services.GetRequiredService(); + + services.ConfigureDefaultWorkflowExecutionPipeline(pipeline => pipeline + .UsePersistence() + .UseStackBasedActivityScheduler()); + } + + [Fact(DisplayName = "Persistent variables are persisted after workflow gets blocked")] + public async Task Test1() + { + // Build the workflow. + var workflow = await new WorkflowDefinitionBuilder().BuildWorkflowAsync(); + + // Run the workflow. + var result = await _workflowRunner.RunAsync(workflow); + + // Assert the variable was persisted. + var persistentVariables = result.WorkflowState.PersistentVariables; + var currentLanguageVariable = persistentVariables.FirstOrDefault(); + + Assert.NotNull(currentLanguageVariable); + } + + [Fact(DisplayName = "Persistent variables are applied before workflow gets resumed")] + public async Task Test2() + { + // Build the workflow. + var languages = new[] { "C#", "JavaScript", "Haskell" }; + var workflow = await new WorkflowDefinitionBuilder().BuildWorkflowAsync(new BlockingWorkflow(languages)); + var currentIndex = 0; + + var bookmark = default(Bookmark?); + var workflowState = default(WorkflowState?); + var variable = workflow.Variables.First(); + + // Continuously resume the workflow until no more bookmarks are returned. + while (true) + { + // Run/resume the workflow. + var result = workflowState == null + ? await _workflowRunner.RunAsync(workflow) + : await _workflowRunner.RunAsync(workflow, workflowState, bookmark); + + bookmark = result.Bookmarks.FirstOrDefault(); + + if (bookmark == null) + break; + + workflowState = result.WorkflowState; + + // Assert that the expected variable is persisted. + var persistentVariablesDictionary = (IDictionary)workflowState.Properties[WorkflowStateDataDrive.VariablesDictionaryStateKey]; + var stateId = $"{workflowState.Id}:{variable.Name}"; + var persistedValue = persistentVariablesDictionary[stateId]; + var expectedValue = languages[currentIndex]; + + Assert.Equal(expectedValue, persistedValue); + + currentIndex ++; + } + } +} \ No newline at end of file diff --git a/test/Elsa.IntegrationTests/Scenarios/PersistentVariables/Workflows.cs b/test/Elsa.IntegrationTests/Scenarios/PersistentVariables/Workflows.cs new file mode 100644 index 000000000..4ed75af0f --- /dev/null +++ b/test/Elsa.IntegrationTests/Scenarios/PersistentVariables/Workflows.cs @@ -0,0 +1,45 @@ +using Elsa.Expressions.Models; +using Elsa.Workflows.Core; +using Elsa.Workflows.Core.Activities; +using Elsa.Workflows.Core.Models; +using Elsa.Workflows.Core.Services; + +namespace Elsa.IntegrationTests.Scenarios.PersistentVariables; + +public class BlockingWorkflow : WorkflowBase +{ + public BlockingWorkflow(string[] languages) + { + Languages = languages; + } + + public string[] Languages { get; } + + protected override void Build(IWorkflowDefinitionBuilder workflow) + { + var currentLanguage = workflow.WithVariable().WithWorkflowDrive(); + + workflow.WithRoot(new Sequence + { + Activities = + { + new WriteLine("Start"), + new ForEach(Languages) + { + CurrentValue = new Output(currentLanguage), + Body = new Sequence + { + Activities = + { + new WriteLine("Waiting for event..."), + new Event("Resume") { Id = "Resume" }, + new WriteLine(context => currentLanguage.Get(context)) + } + } + }, + + new WriteLine("Done") + } + }); + } +} \ No newline at end of file