Incremental work on persistent variables
This commit is contained in:
Sipke Schoorstra
parent
d5257a639c
commit
e67e40bd7b
|
|
@ -43,5 +43,5 @@ public class ExpressionExecutionContext
|
|||
public T GetRequiredService<T>() where T : notnull => _serviceProvider.GetRequiredService<T>();
|
||||
|
||||
//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;
|
||||
}
|
||||
|
|
@ -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<T>(MemoryRegister memoryRegister) => (T?)Get(memoryRegister);
|
||||
public object? Get(ExpressionExecutionContext context) => context.Get(this);
|
||||
public T? Get<T>(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);
|
||||
}
|
||||
|
|
@ -14,14 +14,14 @@ public class MemoryRegister
|
|||
public MemoryRegister? Parent { get; }
|
||||
public IDictionary<string, MemoryBlock> 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<MemoryReference> references)
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -18,6 +18,11 @@ public class ForEach : Activity
|
|||
Behaviors.Remove<AutoCompleteBehavior>();
|
||||
}
|
||||
|
||||
public ForEach(ICollection<object> items) : this()
|
||||
{
|
||||
Items = new Input<ICollection<object>>(items);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The set of values to iterate.
|
||||
/// </summary>
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
using System.Runtime.CompilerServices;
|
||||
using Elsa.Workflows.Core.Models;
|
||||
using Elsa.Workflows.Core.Services;
|
||||
using Elsa.Workflows.Core.Signals;
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
@ -39,6 +39,32 @@ public class WorkflowDefinitionBuilder : IWorkflowDefinitionBuilder
|
|||
return this;
|
||||
}
|
||||
|
||||
public Variable<T> WithVariable<T>(string? driveId = default)
|
||||
{
|
||||
var variable = new Variable<T>()
|
||||
{
|
||||
DriveId = driveId
|
||||
};
|
||||
Variables.Add(variable);
|
||||
return variable;
|
||||
}
|
||||
|
||||
public Variable<T> WithVariable<T>(string name, T value, string? driveId = default)
|
||||
{
|
||||
var variable = value != null ? new Variable<T>(name, value) : new Variable<T>(name);
|
||||
variable.DriveId = driveId;
|
||||
Variables.Add(variable);
|
||||
return variable;
|
||||
}
|
||||
|
||||
public Variable<T> WithVariable<T>(T value, string? driveId = default)
|
||||
{
|
||||
var variable = value != null ? new Variable<T>(value) : new Variable<T>();
|
||||
variable.DriveId = driveId;
|
||||
Variables.Add(variable);
|
||||
return variable;
|
||||
}
|
||||
|
||||
public IWorkflowDefinitionBuilder WithVariable(Variable variable)
|
||||
{
|
||||
Variables.Add(variable);
|
||||
|
|
|
|||
7
src/modules/Elsa.Workflows.Core/DataDriveNames.cs
Normal file
7
src/modules/Elsa.Workflows.Core/DataDriveNames.cs
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
namespace Elsa.Workflows.Core;
|
||||
|
||||
public static class DataDriveNames
|
||||
{
|
||||
public const string Workflow = "Workflow";
|
||||
public const string Memory = "Memory";
|
||||
}
|
||||
|
|
@ -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<T>(this IServiceCollection services) where T : class, IDataDrive
|
||||
{
|
||||
return services.AddSingleton<IDataDrive, T>();
|
||||
}
|
||||
}
|
||||
|
|
@ -22,7 +22,7 @@ public static class DictionaryExtensions
|
|||
public static T? GetValue<TKey, T>(this IDictionary<TKey, T> dictionary, TKey key) => ConvertValue<T>(dictionary[key]);
|
||||
public static T? GetValue<T>(this IDictionary<string, object> dictionary, string key) => ConvertValue<T>(dictionary[key]);
|
||||
|
||||
public static T? GetOrAdd<T>(this IDictionary<string, object> dictionary, string key, Func<T> valueFactory) where T : notnull => dictionary.GetOrAdd<string, T>(key, valueFactory);
|
||||
public static T GetOrAdd<T>(this IDictionary<string, object> dictionary, string key, Func<T> valueFactory) where T : notnull => dictionary.GetOrAdd<string, T>(key, valueFactory);
|
||||
|
||||
public static T GetOrAdd<TKey, T>(this IDictionary<TKey, object> dictionary, TKey key, Func<T> valueFactory)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
@ -69,7 +69,7 @@ public static class WorkflowExecutionContextExtensions
|
|||
/// Gets the specified workflow variable by name.
|
||||
/// </summary>
|
||||
public static T? GetVariable<T>(this WorkflowExecutionContext workflowExecutionContext, string name) => (T?)workflowExecutionContext.GetVariable(name);
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// Gets the specified workflow variable by name, where the name is implied by the type name.
|
||||
/// </summary>
|
||||
|
|
@ -88,7 +88,7 @@ public static class WorkflowExecutionContextExtensions
|
|||
/// Sets the specified workflow variable by name, where the name is implied by the type name.
|
||||
/// </summary>
|
||||
public static Variable SetVariable<T>(this WorkflowExecutionContext workflowExecutionContext, T? value) => workflowExecutionContext.SetVariable(typeof(T).Name, value);
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// Sets the specified workflow variable by name.
|
||||
/// </summary>
|
||||
|
|
@ -99,7 +99,14 @@ public static class WorkflowExecutionContextExtensions
|
|||
/// </summary>
|
||||
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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -79,6 +79,11 @@ public class WorkflowsFeature : FeatureBase
|
|||
// Stream providers.
|
||||
.AddSingleton(StandardInStreamProvider)
|
||||
.AddSingleton(StandardOutStreamProvider)
|
||||
|
||||
// Data drives.
|
||||
.AddSingleton<IDataDriveManager, DataDriveManager>()
|
||||
.AddDataDrive<WorkflowStateDataDrive>()
|
||||
.AddDataDrive<MemoryDataDrive>()
|
||||
|
||||
// Logging
|
||||
.AddLogging();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,10 @@
|
|||
using Elsa.Workflows.Core.Services;
|
||||
|
||||
namespace Elsa.Workflows.Core.Implementations;
|
||||
|
||||
public class DataDriveManager : IDataDriveManager
|
||||
{
|
||||
private readonly IEnumerable<IDataDrive> _drives;
|
||||
public DataDriveManager(IEnumerable<IDataDrive> drives) => _drives = drives;
|
||||
public IDataDrive? GetDriveById(string id) => _drives.FirstOrDefault(x => x.Id == id);
|
||||
}
|
||||
|
|
@ -0,0 +1,27 @@
|
|||
using Elsa.Workflows.Core.Services;
|
||||
|
||||
namespace Elsa.Workflows.Core.Implementations;
|
||||
|
||||
public class MemoryDataDrive : IDataDrive
|
||||
{
|
||||
private readonly IDictionary<string, object> _dictionary = new Dictionary<string, object>();
|
||||
public string Id => DataDriveNames.Memory;
|
||||
|
||||
public ValueTask WriteAsync(string id, object value, DataDriveContext context)
|
||||
{
|
||||
_dictionary[id] = value;
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public ValueTask<object?> 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;
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,35 @@
|
|||
using Elsa.Workflows.Core.Services;
|
||||
|
||||
namespace Elsa.Workflows.Core.Implementations;
|
||||
|
||||
/// <summary>
|
||||
/// A data drive that stores objects in the workflow state itself.
|
||||
/// </summary>
|
||||
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<object?> 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<string, object> GetVariablesDictionary(DataDriveContext context) => context.WorkflowState.Properties.GetOrAdd(VariablesDictionaryStateKey, () => new Dictionary<string, object>());
|
||||
}
|
||||
|
|
@ -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<string, object?> GetOutputFrom(ActivityNode activityNode) =>
|
||||
activityNode.GetType().GetProperties(BindingFlags.Public).Where(x => x.GetCustomAttribute<OutputAttribute>() != null).ToDictionary(x => x.Name, x => (object?)x.GetValue(activityNode));
|
||||
}
|
||||
|
|
@ -146,19 +146,19 @@ public class ActivityExecutionContext
|
|||
public object? GetService(Type serviceType) => WorkflowExecutionContext.GetService(serviceType);
|
||||
public T? Get<T>(Input<T>? input) => input == null ? default : Get<T>(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<T>(MemoryReference locationReference)
|
||||
public T? Get<T>(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<T>(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);
|
||||
}
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
using System.Text.Json.Serialization;
|
||||
|
||||
namespace Elsa.Workflows.Core.Models;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a workflow-specific memory drive definition that is configured with a specific storage driver.
|
||||
/// </summary>
|
||||
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!;
|
||||
}
|
||||
|
|
@ -25,7 +25,14 @@ public class Variable : MemoryReference
|
|||
get => Id;
|
||||
set => Id = value;
|
||||
}
|
||||
|
||||
public object? Value { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// When specified, the variable's value will be stored on the specified data drive.
|
||||
/// </summary>
|
||||
public string? DriveId { get; set; }
|
||||
|
||||
public override MemoryBlock Declare() => new(Value);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -47,6 +47,15 @@ public class Workflow : Composite, ICloneable
|
|||
public WorkflowPublication Publication { get; set; }
|
||||
public WorkflowMetadata WorkflowMetadata { get; set; }
|
||||
public ICollection<Variable> Variables { get; init; }
|
||||
|
||||
/// <summary>
|
||||
/// A list of storage drive definitions that can be used by variables to persist their values.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Depending on the drive's driver capabilities, only serializable variables can be persisted.
|
||||
/// </remarks>
|
||||
public ICollection<DataDriveDefinition> Drives { get; set; } = new List<DataDriveDefinition>();
|
||||
|
||||
public Workflow Clone() => (Workflow)((ICloneable)this).Clone();
|
||||
object ICloneable.Clone() => MemberwiseClone();
|
||||
}
|
||||
16
src/modules/Elsa.Workflows.Core/Services/IDataDrive.cs
Normal file
16
src/modules/Elsa.Workflows.Core/Services/IDataDrive.cs
Normal file
|
|
@ -0,0 +1,16 @@
|
|||
using Elsa.Workflows.Core.State;
|
||||
|
||||
namespace Elsa.Workflows.Core.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a storage driver for workflows to write state to.
|
||||
/// </summary>
|
||||
public interface IDataDrive
|
||||
{
|
||||
string Id { get; }
|
||||
ValueTask WriteAsync(string id, object value, DataDriveContext context);
|
||||
ValueTask<object?> ReadAsync(string id, DataDriveContext context);
|
||||
ValueTask DeleteAsync(string id, DataDriveContext context);
|
||||
}
|
||||
|
||||
public record DataDriveContext(WorkflowState WorkflowState, CancellationToken CancellationToken);
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
namespace Elsa.Workflows.Core.Services;
|
||||
|
||||
public interface IDataDriveManager
|
||||
{
|
||||
IDataDrive? GetDriveById(string id);
|
||||
}
|
||||
|
|
@ -12,6 +12,9 @@ public interface IWorkflowDefinitionBuilder
|
|||
IWorkflowDefinitionBuilder WithDefinitionId(string definitionId);
|
||||
IWorkflowDefinitionBuilder WithVersion(int version);
|
||||
IWorkflowDefinitionBuilder WithRoot(IActivity root);
|
||||
Variable<T> WithVariable<T>(string? driveId = default);
|
||||
Variable<T> WithVariable<T>(string name, T value, string? driveId = default);
|
||||
Variable<T> WithVariable<T>(T value, string? driveId = default);
|
||||
IWorkflowDefinitionBuilder WithVariable(Variable variable);
|
||||
IWorkflowDefinitionBuilder WithVariables(params Variable[] variables);
|
||||
IWorkflowDefinitionBuilder WithApplicationProperty(string name, object value);
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
|
@ -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!;
|
||||
}
|
||||
|
|
@ -30,7 +30,7 @@ public class WorkflowState
|
|||
/// <summary>
|
||||
/// A dictionary of activity output.
|
||||
/// </summary>
|
||||
public IDictionary<string, IDictionary<string, object?>> ActivityOutput { get; set; } = new Dictionary<string, IDictionary<string, object?>>();
|
||||
public IDictionary<string, IDictionary<string, object>> ActivityOutput { get; set; } = new Dictionary<string, IDictionary<string, object>>();
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
|||
/// <summary>
|
||||
/// A global property bag that contains properties set by application code and/or activities.
|
||||
/// </summary>
|
||||
public IDictionary<string, object?> Properties { get; set; } = new Dictionary<string, object?>();
|
||||
public IDictionary<string, object> Properties { get; set; } = new Dictionary<string, object>();
|
||||
|
||||
/// <summary>
|
||||
/// A list of variables that can be persisted.
|
||||
/// </summary>
|
||||
public ICollection<PersistentVariableState> PersistentVariables { get; set; } = new List<PersistentVariableState>();
|
||||
}
|
||||
|
|
@ -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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<IWorkflowRunner>();
|
||||
|
||||
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<BlockingWorkflow>();
|
||||
|
||||
// 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<string, object>)workflowState.Properties[WorkflowStateDataDrive.VariablesDictionaryStateKey];
|
||||
var stateId = $"{workflowState.Id}:{variable.Name}";
|
||||
var persistedValue = persistentVariablesDictionary[stateId];
|
||||
var expectedValue = languages[currentIndex];
|
||||
|
||||
Assert.Equal(expectedValue, persistedValue);
|
||||
|
||||
currentIndex ++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<string>().WithWorkflowDrive();
|
||||
|
||||
workflow.WithRoot(new Sequence
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
new WriteLine("Start"),
|
||||
new ForEach(Languages)
|
||||
{
|
||||
CurrentValue = new Output<MemoryReference?>(currentLanguage),
|
||||
Body = new Sequence
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
new WriteLine("Waiting for event..."),
|
||||
new Event("Resume") { Id = "Resume" },
|
||||
new WriteLine(context => currentLanguage.Get<string>(context))
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
new WriteLine("Done")
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue