Fixed Log Persistence Mode Evaluation For Activity Execution (#6595)
* Refactor log persistence mode evaluation for activity execution Updated functionality to evaluate and filter log persistence modes during activity execution. Added middleware, interfaces, and supporting methods to manage log configuration, ensuring only persistable inputs and outputs are retained in execution mapping. Updated pipeline to include the new middleware. * Refactor log persistence mapping and evaluation logic Updated log persistence evaluation to use separate input/output maps, ensuring a more modular and maintainable structure. Adjusted property key handling and introduced a dedicated helper method for evaluating persistence properties. Added middleware for log persistence mode evaluation to the workflow execution pipeline. * Refactor log persistence logic and restructure namespaces. Centralizes log persistence logic under a dedicated `LogPersistence` namespace. Simplifies interfaces and methods to streamline functionality, ensuring clarity and consistency in log persistence evaluations. * Refactor log persistence methods for encapsulation. Converted multiple public methods to private to enhance encapsulation and adherence to the principle of least privilege. Introduced a helper method `ResolveFinalLogPersistenceMode` to improve code clarity and maintainability. * Refactor log persistence with improved type safety and structure Updated methods and properties related to log persistence to remove nullable types, enhance clarity, and ensure type safety. Simplified configurations and refactored logic for evaluating persistence modes, reducing redundancy and improving maintainability. * Add XML documentation to IActivityPropertyLogPersistenceEvaluator This commit introduces XML comments to provide clarity on the purpose and functionality of the interface and its methods, aiding developers in understanding their usage and behavior during workflow execution. * Refine log persistence logic in activity execution mapping. Ensure that log persistence modes are correctly handled by adding explicit checks for `LogPersistenceMode.Inherit`. This prevents potential ambiguities and ensures accurate property mapping during workflow execution. * Refactor state handling and log persistence evaluation. Updated method signatures for stricter type consistency and improved readability. Introduced additional resolution step in log persistence to handle legacy configurations more effectively. This enhances code maintainability and alignment with expected behavior. * Update src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
278ed34981
commit
b88d12cdd8
|
|
@ -101,7 +101,7 @@ public static partial class ActivityExecutionContextExtensions
|
|||
// When input is created from an activity provider, there may be no memory block reference.
|
||||
if (memoryReference?.Id != null!)
|
||||
{
|
||||
// Declare the input memory block on the current context.
|
||||
// Declare the input memory block in the current context.
|
||||
context.ExpressionExecutionContext.Set(memoryReference, value!);
|
||||
}
|
||||
}
|
||||
|
|
@ -110,12 +110,12 @@ public static partial class ActivityExecutionContextExtensions
|
|||
value = input;
|
||||
}
|
||||
|
||||
await StoreInputValueAsync(context, inputDescriptor, value);
|
||||
await StoreInputValueAsync(context, inputDescriptor, value!);
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
private static async Task StoreInputValueAsync(ActivityExecutionContext context, InputDescriptor inputDescriptor, object? value)
|
||||
private static Task StoreInputValueAsync(ActivityExecutionContext context, InputDescriptor inputDescriptor, object value)
|
||||
{
|
||||
// Store the serialized input value in the activity state.
|
||||
// Serializing the value ensures we store a copy of the value and not a reference to the input, which may change over time.
|
||||
|
|
@ -128,5 +128,7 @@ public static partial class ActivityExecutionContextExtensions
|
|||
// var filterResult = await manager.RunFiltersAsync(filterContext);
|
||||
context.ActivityState[inputDescriptor.Name] = value;
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
using Elsa.Extensions;
|
||||
using Elsa.Workflows.Models;
|
||||
|
||||
namespace Elsa.Workflows;
|
||||
|
||||
public static class ActivityExecutionContextExtensions
|
||||
{
|
||||
public static IDictionary<string, object> GetInputs(this ActivityExecutionContext context)
|
||||
{
|
||||
return context.ActivityState!;
|
||||
}
|
||||
|
||||
public static IDictionary<string, object> GetOutputs(this ActivityExecutionContext context)
|
||||
{
|
||||
var activity = context.Activity;
|
||||
var expressionExecutionContext = context.ExpressionExecutionContext;
|
||||
var activityDescriptor = context.ActivityDescriptor;
|
||||
var outputDescriptors = activityDescriptor.Outputs;
|
||||
|
||||
var outputs = outputDescriptors.ToDictionary(x => x.Name, x =>
|
||||
{
|
||||
if (x.IsSerializable == false)
|
||||
return "(not serializable)";
|
||||
|
||||
var cachedValue = activity.GetOutput(expressionExecutionContext, x.Name);
|
||||
|
||||
if (cachedValue != null)
|
||||
return cachedValue;
|
||||
|
||||
if (x.ValueGetter(activity) is Output output && context.TryGet(output.MemoryBlockReference(), out var outputValue))
|
||||
return outputValue!;
|
||||
|
||||
return null!;
|
||||
});
|
||||
|
||||
return outputs;
|
||||
}
|
||||
}
|
||||
|
|
@ -64,7 +64,7 @@ public static class ActivityExtensions
|
|||
/// <param name="context">The activity execution context.</param>
|
||||
/// <param name="outputName">Name of the output.</param>
|
||||
/// <returns>The output value.</returns>
|
||||
public static object? GetOutput(this IActivity activity, ActivityExecutionContext context, string? outputName = default)
|
||||
public static object? GetOutput(this IActivity activity, ActivityExecutionContext context, string? outputName = null)
|
||||
{
|
||||
var workflowExecutionContext = context.WorkflowExecutionContext;
|
||||
var outputRegister = workflowExecutionContext.GetActivityOutputRegister();
|
||||
|
|
@ -84,7 +84,7 @@ public static class ActivityExtensions
|
|||
/// <param name="context">The expression execution context.</param>
|
||||
/// <param name="outputName">Name of the output.</param>
|
||||
/// <returns>The output value.</returns>
|
||||
public static object? GetOutput(this IActivity activity, ExpressionExecutionContext context, string? outputName = default)
|
||||
public static object? GetOutput(this IActivity activity, ExpressionExecutionContext context, string? outputName = null)
|
||||
{
|
||||
var activityExecutionContext = context.GetActivityExecutionContext();
|
||||
|
||||
|
|
|
|||
|
|
@ -10,12 +10,11 @@ public interface IActivityExecutionMapper
|
|||
/// <summary>
|
||||
/// Maps an activity execution context to an activity execution record.
|
||||
/// </summary>
|
||||
Task<ActivityExecutionRecord> MapAsync(ActivityExecutionContext source);
|
||||
|
||||
ActivityExecutionRecord Map(ActivityExecutionContext source);
|
||||
|
||||
/// <summary>
|
||||
/// Retrieves a dictionary containing the persistable output of an activity execution context.
|
||||
/// Maps an activity execution context to an activity execution record.
|
||||
/// </summary>
|
||||
/// <param name="context">The activity execution context to extract persistable output from.</param>
|
||||
/// <returns>A dictionary containing the persistable output.</returns>
|
||||
Task<Dictionary<string, object?>> GetPersistableOutputAsync(ActivityExecutionContext context);
|
||||
[Obsolete( "Use Map instead.", error: false)]
|
||||
Task<ActivityExecutionRecord> MapAsync(ActivityExecutionContext source);
|
||||
}
|
||||
|
|
@ -4,6 +4,12 @@
|
|||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=contracts/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=enums/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=extensions/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Ccontracts/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Cenums/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Cextensions/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Cmodels/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Cservices/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=models/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=requests/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=responses/@EntryIndexedValue">True</s:Boolean>
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ public static class PipelineWorkflowsFeatureExtensions
|
|||
.UseExceptionHandling()
|
||||
.UseExecutionLogging()
|
||||
.UseNotifications()
|
||||
.UseLogPersistenceModeEvaluation()
|
||||
.UseBackgroundActivityInvoker();
|
||||
|
||||
configurePipeline?.Invoke(pipeline);
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ public static class WorkflowsFeatureExtensions
|
|||
workflowsFeature.WithActivityExecutionPipeline(pipeline =>
|
||||
pipeline
|
||||
.UseExceptionHandling()
|
||||
.UseLogPersistenceModeEvaluation()
|
||||
.UseExecutionLogging()
|
||||
.UseNotifications()
|
||||
.UseBackgroundActivityInvoker());
|
||||
|
|
|
|||
|
|
@ -284,6 +284,7 @@ public class WorkflowRuntimeFeature(IModule module) : FeatureBase(module)
|
|||
.AddScoped<IWorkflowRestarter, DefaultWorkflowRestarter>()
|
||||
.AddScoped<IBookmarkQueuePurger, DefaultBookmarkQueuePurger>()
|
||||
.AddScoped<ILogRecordExtractor<WorkflowExecutionLogRecord>, WorkflowExecutionLogRecordExtractor>()
|
||||
.AddScoped<IActivityPropertyLogPersistenceEvaluator, ActivityPropertyLogPersistenceEvaluator>()
|
||||
.AddScoped<IBookmarkQueueProcessor, BookmarkQueueProcessor>()
|
||||
.AddScoped<DefaultCommitStateHandler>()
|
||||
.AddScoped<WorkflowHeartbeatGeneratorFactory>()
|
||||
|
|
|
|||
|
|
@ -1 +0,0 @@
|
|||
namespace Elsa.Workflows.Runtime.Instructions;
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
namespace Elsa.Workflows.Runtime;
|
||||
|
||||
/// <summary>
|
||||
/// Provides functionality for evaluating log persistence settings for activity properties
|
||||
/// during the execution of a workflow.
|
||||
/// </summary>
|
||||
public interface IActivityPropertyLogPersistenceEvaluator
|
||||
{
|
||||
/// <summary>
|
||||
/// Evaluates the log persistence settings for activity properties within the context of a workflow's execution.
|
||||
/// </summary>
|
||||
Task<ActivityLogPersistenceModeMap> EvaluateLogPersistenceModesAsync(ActivityExecutionContext context);
|
||||
|
||||
/// <summary>
|
||||
/// Retrieves a dictionary of persistable output values generated during the execution of an activity.
|
||||
/// </summary>
|
||||
/// <returns>A dictionary where the keys represent output property names and the values represent their persistable data.</returns>
|
||||
Task<Dictionary<string, object>> GetPersistableOutputAsync(ActivityExecutionContext context);
|
||||
}
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
using Elsa.Extensions;
|
||||
|
||||
namespace Elsa.Workflows.Runtime;
|
||||
|
||||
public static class ActivityExecutionContextExtensions
|
||||
{
|
||||
private static object LogPersistenceMapKey { get; } = new();
|
||||
|
||||
public static ActivityLogPersistenceModeMap GetLogPersistenceModeMap(this ActivityExecutionContext context)
|
||||
{
|
||||
return context.TransientProperties.GetValueOrDefault(LogPersistenceMapKey, () => new ActivityLogPersistenceModeMap())!;
|
||||
}
|
||||
|
||||
public static void SetLogPersistenceModeMap(this ActivityExecutionContext context, ActivityLogPersistenceModeMap map)
|
||||
{
|
||||
context.TransientProperties[LogPersistenceMapKey] = map;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Pipelines.ActivityExecution;
|
||||
using Elsa.Workflows.Runtime.Middleware;
|
||||
using Elsa.Workflows.Runtime.Middleware.Activities;
|
||||
|
||||
// ReSharper disable once CheckNamespace
|
||||
|
|
@ -14,4 +15,9 @@ public static class ActivityExecutionPipelineBuilderExtensions
|
|||
/// Installs the <see cref="BackgroundActivityInvokerMiddleware"/>.
|
||||
/// </summary>
|
||||
public static IActivityExecutionPipelineBuilder UseBackgroundActivityInvoker(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware<BackgroundActivityInvokerMiddleware>();
|
||||
|
||||
/// <summary>
|
||||
/// Installs the <see cref="EvaluateLogPersistenceModesMiddleware"/> which evaluates log persistence modes during activity execution.
|
||||
/// </summary>
|
||||
public static IActivityExecutionPipelineBuilder UseLogPersistenceModeEvaluation(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware<EvaluateLogPersistenceModesMiddleware>();
|
||||
}
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
using Elsa.Workflows.Pipelines.ActivityExecution;
|
||||
|
||||
namespace Elsa.Workflows.Runtime.Middleware;
|
||||
|
||||
public class EvaluateLogPersistenceModesMiddleware(ActivityMiddlewareDelegate next, IActivityPropertyLogPersistenceEvaluator persistenceEvaluator) : IActivityExecutionMiddleware
|
||||
{
|
||||
public async ValueTask InvokeAsync(ActivityExecutionContext context)
|
||||
{
|
||||
await next(context);
|
||||
var persistenceLogMap = await persistenceEvaluator.EvaluateLogPersistenceModesAsync(context);
|
||||
context.SetLogPersistenceModeMap(persistenceLogMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
using Elsa.Workflows.LogPersistence;
|
||||
|
||||
namespace Elsa.Workflows.Runtime;
|
||||
|
||||
public class ActivityLogPersistenceModeMap
|
||||
{
|
||||
public IDictionary<string, LogPersistenceMode> Inputs { get; set; } = new Dictionary<string, LogPersistenceMode>();
|
||||
public IDictionary<string, LogPersistenceMode> Outputs { get; set; } = new Dictionary<string, LogPersistenceMode>();
|
||||
}
|
||||
|
|
@ -0,0 +1,230 @@
|
|||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
using Elsa.Expressions.Contracts;
|
||||
using Elsa.Expressions.Models;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.LogPersistence;
|
||||
using Elsa.Workflows.LogPersistence.Strategies;
|
||||
using Elsa.Workflows.Management.Options;
|
||||
using Elsa.Workflows.Models;
|
||||
using Elsa.Workflows.Serialization.Converters;
|
||||
using Humanizer;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace Elsa.Workflows.Runtime;
|
||||
|
||||
/* The following legacy JSON structure is expected to be found in the custom properties of the workflow and activity:
|
||||
* {
|
||||
* "logPersistenceMode": {
|
||||
* "default": "default",
|
||||
* "inputs": { k : v },
|
||||
* "outputs": { k: v }
|
||||
* }
|
||||
* }
|
||||
*/
|
||||
|
||||
/* The following JSON structure is expected to be found in the custom properties of the workflow and activity:
|
||||
* {
|
||||
* "logPersistenceConfig": {
|
||||
* "default": { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." },
|
||||
* "inputs": { "input1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } },
|
||||
* "outputs": { "output1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } }
|
||||
* }
|
||||
* }
|
||||
*/
|
||||
public class ActivityPropertyLogPersistenceEvaluator : IActivityPropertyLogPersistenceEvaluator
|
||||
{
|
||||
private readonly IExpressionEvaluator _expressionEvaluator;
|
||||
private readonly JsonSerializerOptions _jsonOptions;
|
||||
private readonly IDictionary<string, ILogPersistenceStrategy> _strategies;
|
||||
private readonly IOptions<ManagementOptions> _options;
|
||||
private readonly ILogger _logger;
|
||||
|
||||
private const string LegacyKey = "logPersistenceMode";
|
||||
const string ConfigKey = "logPersistenceConfig";
|
||||
|
||||
public ActivityPropertyLogPersistenceEvaluator(
|
||||
ILogPersistenceStrategyService strategyService,
|
||||
IExpressionDescriptorRegistry expressionDescriptorRegistry,
|
||||
IExpressionEvaluator expressionEvaluator,
|
||||
IOptions<ManagementOptions> options,
|
||||
ILogger<ActivityPropertyLogPersistenceEvaluator> logger)
|
||||
{
|
||||
_expressionEvaluator = expressionEvaluator;
|
||||
_options = options;
|
||||
_logger = logger;
|
||||
_strategies = strategyService.ListStrategies().ToDictionary(x => x.GetType().GetSimpleAssemblyQualifiedName(), x => x);
|
||||
_jsonOptions = new JsonSerializerOptions
|
||||
{
|
||||
PropertyNameCaseInsensitive = true
|
||||
}.WithConverters(
|
||||
new ExpressionJsonConverterFactory(expressionDescriptorRegistry),
|
||||
new JsonStringEnumConverter(),
|
||||
new ExpandoObjectConverterFactory());
|
||||
}
|
||||
|
||||
public async Task<ActivityLogPersistenceModeMap> EvaluateLogPersistenceModesAsync(ActivityExecutionContext context)
|
||||
{
|
||||
var cancellationToken = context.CancellationToken;
|
||||
var (legacyProps, configProps, defaultMode) = await GetPersistenceDefaultsAsync(context, cancellationToken);
|
||||
var map = new ActivityLogPersistenceModeMap();
|
||||
|
||||
await EvaluatePropertiesAsync(context, "inputs", context.ActivityDescriptor.Inputs, legacyProps, configProps, defaultMode, map.Inputs, cancellationToken);
|
||||
await EvaluatePropertiesAsync(context, "outputs", context.ActivityDescriptor.Outputs, legacyProps, configProps, defaultMode, map.Outputs, cancellationToken);
|
||||
|
||||
return map;
|
||||
}
|
||||
|
||||
public async Task<Dictionary<string, object>> GetPersistableOutputAsync(ActivityExecutionContext context)
|
||||
{
|
||||
var cancellationToken = context.WorkflowExecutionContext.CancellationToken;
|
||||
var (legacyProps, configProps, defaultMode) = await GetPersistenceDefaultsAsync(context, cancellationToken);
|
||||
var outputs = context.GetOutputs();
|
||||
return await GetPersistablePropertiesAsync(context, outputs, "outputs", legacyProps, configProps, defaultMode, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<(IDictionary<string, object> legacyProps, IDictionary<string, object> configProps, LogPersistenceMode defaultMode)>
|
||||
GetPersistenceDefaultsAsync(ActivityExecutionContext context, CancellationToken cancellationToken)
|
||||
{
|
||||
var legacyProps = context.Activity.CustomProperties.GetValueOrDefault<IDictionary<string, object>>(LegacyKey, () => new Dictionary<string, object>())!;
|
||||
var rootContext = context.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.ParentActivityExecutionContext == null);
|
||||
var workflow = (Workflow?)context.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? context.WorkflowExecutionContext.Workflow;
|
||||
var workflowDefault = await GetDefaultPersistenceModeAsync(rootContext.ExpressionExecutionContext, workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken);
|
||||
var activityDefault = await GetDefaultPersistenceModeAsync(context.ExpressionExecutionContext, context.Activity.CustomProperties, () => workflowDefault, cancellationToken);
|
||||
var configProps = context.Activity.CustomProperties.GetValueOrDefault<IDictionary<string, object>>(ConfigKey, () => new Dictionary<string, object>())!;
|
||||
return (legacyProps, configProps, activityDefault);
|
||||
}
|
||||
|
||||
private async Task EvaluatePropertiesAsync(
|
||||
ActivityExecutionContext context,
|
||||
string key,
|
||||
IEnumerable<PropertyDescriptor> descriptors,
|
||||
IDictionary<string, object> legacyConfig,
|
||||
IDictionary<string, object> currentConfig,
|
||||
LogPersistenceMode defaultMode,
|
||||
IDictionary<string, LogPersistenceMode> resultMap,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var legacySection = legacyConfig.GetValueOrDefault(key, () => new Dictionary<string, object>())!;
|
||||
var currentSection = currentConfig.GetValueOrDefault(key, () => new Dictionary<string, object>())!;
|
||||
|
||||
foreach (var descriptor in descriptors)
|
||||
{
|
||||
resultMap[descriptor.Name] = await EvaluatePropertyModeAsync(context.ExpressionExecutionContext, descriptor, legacySection, currentSection, defaultMode, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<LogPersistenceMode> EvaluatePropertyModeAsync(
|
||||
ExpressionExecutionContext executionContext,
|
||||
PropertyDescriptor descriptor,
|
||||
IDictionary<string, object> legacySection,
|
||||
IDictionary<string, object> currentSection,
|
||||
LogPersistenceMode defaultMode,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var key = descriptor.Name.Camelize();
|
||||
var configObject = currentSection.GetValueOrDefault(key, () => null);
|
||||
var config = ConvertToConfig(configObject);
|
||||
if (config != null)
|
||||
return await EvaluateConfigAsync(config, executionContext, () => defaultMode, cancellationToken);
|
||||
|
||||
var mode = legacySection.GetValueOrDefault(key, () => defaultMode);
|
||||
return ResolveMode(mode, () => defaultMode);
|
||||
}
|
||||
|
||||
private async Task<Dictionary<string, object>> GetPersistablePropertiesAsync(
|
||||
ActivityExecutionContext context,
|
||||
IDictionary<string, object> state,
|
||||
string key,
|
||||
IDictionary<string, object> legacyConfig,
|
||||
IDictionary<string, object> currentConfig,
|
||||
LogPersistenceMode defaultMode,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var result = new Dictionary<string, object>();
|
||||
var legacySection = legacyConfig.GetValueOrDefault(key, () => new Dictionary<string, object>());
|
||||
var currentSection = currentConfig.GetValueOrDefault(key, () => new Dictionary<string, object>());
|
||||
|
||||
foreach (var item in state)
|
||||
{
|
||||
var propKey = item.Key.Camelize();
|
||||
var configObject = currentSection!.GetValueOrDefault(propKey, () => null);
|
||||
var config = ConvertToConfig(configObject);
|
||||
var mode = config != null
|
||||
? await EvaluateConfigAsync(config, context.ExpressionExecutionContext, () => defaultMode, cancellationToken)
|
||||
: legacySection!.GetValueOrDefault(propKey, () => defaultMode);
|
||||
|
||||
if (mode == LogPersistenceMode.Include || (mode == LogPersistenceMode.Inherit && (defaultMode == LogPersistenceMode.Include || defaultMode == LogPersistenceMode.Inherit)))
|
||||
result.Add(item.Key, item.Value);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private async Task<LogPersistenceMode> GetDefaultPersistenceModeAsync(
|
||||
ExpressionExecutionContext executionContext,
|
||||
IDictionary<string, object> properties,
|
||||
Func<LogPersistenceMode> defaultFactory,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var legacyProps = properties.GetValueOrDefault<IDictionary<string, object>>(LegacyKey, () => new Dictionary<string, object>());
|
||||
var configProps = properties.GetValueOrDefault<IDictionary<string, object>>(ConfigKey, () => new Dictionary<string, object>());
|
||||
var defaultObj = configProps!.TryGetValue("default", out var val) ? val : null;
|
||||
if (defaultObj == null)
|
||||
{
|
||||
var legacyDefault = legacyProps!.GetValueOrDefault("default", defaultFactory);
|
||||
return legacyDefault == LogPersistenceMode.Inherit ? defaultFactory() : legacyDefault;
|
||||
}
|
||||
|
||||
var config = ConvertToConfig(defaultObj);
|
||||
return await EvaluateConfigAsync(config, executionContext, defaultFactory, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<LogPersistenceMode> EvaluateConfigAsync(
|
||||
LogPersistenceConfiguration? config,
|
||||
ExpressionExecutionContext executionContext,
|
||||
Func<LogPersistenceMode> defaultFactory,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (config?.EvaluationMode == LogPersistenceEvaluationMode.Strategy)
|
||||
{
|
||||
var strategyType = config.StrategyType ?? typeof(Inherit).GetSimpleAssemblyQualifiedName();
|
||||
if (!_strategies.TryGetValue(strategyType, out var strategy))
|
||||
return defaultFactory();
|
||||
|
||||
var strategyContext = new LogPersistenceStrategyContext(cancellationToken);
|
||||
var mode = await strategy.GetPersistenceModeAsync(strategyContext);
|
||||
return ResolveMode(mode, defaultFactory);
|
||||
}
|
||||
|
||||
if (config?.Expression == null)
|
||||
return defaultFactory();
|
||||
|
||||
try
|
||||
{
|
||||
var mode = await _expressionEvaluator.EvaluateAsync<LogPersistenceMode>(config.Expression, executionContext);
|
||||
return ResolveMode(mode, defaultFactory);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error evaluating log persistence expression");
|
||||
return defaultFactory();
|
||||
}
|
||||
}
|
||||
|
||||
private LogPersistenceMode ResolveMode(LogPersistenceMode mode, Func<LogPersistenceMode> defaultFactory)
|
||||
{
|
||||
var m = mode == LogPersistenceMode.Inherit ? defaultFactory() : mode;
|
||||
return m == LogPersistenceMode.Inherit ? LogPersistenceMode.Include : m;
|
||||
}
|
||||
|
||||
private LogPersistenceConfiguration? ConvertToConfig(object? value)
|
||||
{
|
||||
if (value == null) return null;
|
||||
if (value is LogPersistenceConfiguration config) return config;
|
||||
var json = JsonSerializer.Serialize(value);
|
||||
return JsonSerializer.Deserialize<LogPersistenceConfiguration>(json, _jsonOptions);
|
||||
}
|
||||
}
|
||||
|
|
@ -12,8 +12,5 @@ public class PersistActivityExecutionLogMiddleware(WorkflowMiddlewareDelegate ne
|
|||
public override async ValueTask InvokeAsync(WorkflowExecutionContext context)
|
||||
{
|
||||
await Next(context);
|
||||
|
||||
// Not used anymore.
|
||||
//await sink.PersistExecutionLogsAsync(context);
|
||||
}
|
||||
}
|
||||
|
|
@ -13,8 +13,5 @@ public class PersistWorkflowExecutionLogMiddleware(WorkflowMiddlewareDelegate ne
|
|||
{
|
||||
// Invoke next middleware.
|
||||
await Next(context);
|
||||
|
||||
// Not used anymore.
|
||||
//await sink.PersistExecutionLogsAsync(context);
|
||||
}
|
||||
}
|
||||
|
|
@ -15,7 +15,7 @@ public class BackgroundActivityInvoker(
|
|||
IWorkflowDefinitionService workflowDefinitionService,
|
||||
IVariablePersistenceManager variablePersistenceManager,
|
||||
IActivityInvoker activityInvoker,
|
||||
IActivityExecutionMapper activityExecutionMapper,
|
||||
IActivityPropertyLogPersistenceEvaluator activityPropertyLogPersistenceEvaluator,
|
||||
WorkflowHeartbeatGeneratorFactory workflowHeartbeatGeneratorFactory,
|
||||
IServiceProvider serviceProvider,
|
||||
ILogger<BackgroundActivityInvoker> logger)
|
||||
|
|
@ -66,7 +66,7 @@ public class BackgroundActivityInvoker(
|
|||
var completed = activityExecutionContext.GetBackgroundCompleted();
|
||||
var scheduledActivities = activityExecutionContext.GetBackgroundScheduledActivities().ToList();
|
||||
var workflowInstanceId = scheduledBackgroundActivity.WorkflowInstanceId;
|
||||
var outputValues = await activityExecutionMapper.GetPersistableOutputAsync(activityExecutionContext);
|
||||
var outputValues = await activityPropertyLogPersistenceEvaluator.GetPersistableOutputAsync(activityExecutionContext);
|
||||
var properties = new Dictionary<string, object>
|
||||
{
|
||||
[scheduledActivitiesKey] = JsonSerializer.Serialize(scheduledActivities),
|
||||
|
|
|
|||
|
|
@ -1,88 +1,20 @@
|
|||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
using Elsa.Expressions.Contracts;
|
||||
using Elsa.Expressions.Models;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.LogPersistence;
|
||||
using Elsa.Workflows.LogPersistence.Strategies;
|
||||
using Elsa.Workflows.Management.Options;
|
||||
using Elsa.Workflows.Models;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
using Elsa.Workflows.Serialization.Converters;
|
||||
using Elsa.Workflows.State;
|
||||
using Humanizer;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace Elsa.Workflows.Runtime;
|
||||
|
||||
/// <inheritdoc />
|
||||
public class DefaultActivityExecutionMapper : IActivityExecutionMapper
|
||||
{
|
||||
private readonly JsonSerializerOptions _logPersistenceConfigSerializerOptions;
|
||||
private readonly IOptions<ManagementOptions> _options;
|
||||
private readonly IExpressionEvaluator _expressionEvaluator;
|
||||
private readonly ILogger<DefaultActivityExecutionMapper> _logger;
|
||||
private readonly IDictionary<string, ILogPersistenceStrategy> _logPersistenceStrategies;
|
||||
|
||||
public DefaultActivityExecutionMapper(
|
||||
IOptions<ManagementOptions> options,
|
||||
ILogPersistenceStrategyService logPersistenceStrategyService,
|
||||
IExpressionEvaluator expressionEvaluator,
|
||||
IExpressionDescriptorRegistry expressionDescriptorRegistry,
|
||||
ILogger<DefaultActivityExecutionMapper> logger)
|
||||
public ActivityExecutionRecord Map(ActivityExecutionContext source)
|
||||
{
|
||||
_options = options;
|
||||
_expressionEvaluator = expressionEvaluator;
|
||||
_logger = logger;
|
||||
_logPersistenceStrategies = logPersistenceStrategyService.ListStrategies().ToDictionary(x => x.GetType().GetSimpleAssemblyQualifiedName(), x => x);
|
||||
|
||||
_logPersistenceConfigSerializerOptions = new JsonSerializerOptions
|
||||
{
|
||||
PropertyNameCaseInsensitive = true
|
||||
}.WithConverters(
|
||||
new ExpressionJsonConverterFactory(expressionDescriptorRegistry),
|
||||
new JsonStringEnumConverter(),
|
||||
new ExpandoObjectConverterFactory());
|
||||
}
|
||||
|
||||
private const string LegacyLogPersistenceModeKey = "logPersistenceMode";
|
||||
private const string LogPersistenceConfigKey = "logPersistenceConfig";
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<ActivityExecutionRecord> MapAsync(ActivityExecutionContext source)
|
||||
{
|
||||
/* The following legacy JSON structure is expected to be found in the custom properties of the workflow and activity:
|
||||
* {
|
||||
* "logPersistenceMode": {
|
||||
* "default": "default",
|
||||
* "inputs": { k : v },
|
||||
* "outputs": { k: v }
|
||||
* }
|
||||
* }
|
||||
*/
|
||||
|
||||
/* The following JSON structure is expected to be found in the custom properties of the workflow and activity:
|
||||
* {
|
||||
* "logPersistenceConfig": {
|
||||
* "default": { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." },
|
||||
* "inputs": { "input1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } },
|
||||
* "outputs": { "output1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } }
|
||||
* }
|
||||
* }
|
||||
*/
|
||||
|
||||
var cancellationToken = source.CancellationToken;
|
||||
var legacyActivityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault<IDictionary<string, object?>>(LegacyLogPersistenceModeKey, () => new Dictionary<string, object?>())!;
|
||||
var rootActivityExecutionContext = source.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.ParentActivityExecutionContext == null);
|
||||
var workflow = (Workflow?)source.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? source.WorkflowExecutionContext.Workflow;
|
||||
var workflowPersistenceProperty = await GetDefaultPersistenceModeAsync(rootActivityExecutionContext.ExpressionExecutionContext, workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken);
|
||||
var activityPersistencePropertyDefault = await GetDefaultPersistenceModeAsync(source.ExpressionExecutionContext, source.Activity.CustomProperties, () => workflowPersistenceProperty, cancellationToken);
|
||||
var activityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault<IDictionary<string, object?>>(LogPersistenceConfigKey, () => new Dictionary<string, object?>())!;
|
||||
var payload = GetPayload(source);
|
||||
var outputs = await GetPersistableOutputAsync(source, legacyActivityPersistenceProperties, activityPersistenceProperties, activityPersistencePropertyDefault, cancellationToken);
|
||||
var inputs = await GetPersistableInputAsync(source, legacyActivityPersistenceProperties, activityPersistenceProperties, activityPersistencePropertyDefault, cancellationToken);
|
||||
var outputs = source.GetOutputs();
|
||||
var inputs = source.GetInputs();
|
||||
var persistenceMap = source.GetLogPersistenceModeMap();
|
||||
var persistableInputs = GetPersistableProperties(inputs, persistenceMap.Inputs);
|
||||
var persistableOutputs = GetPersistableProperties(outputs, persistenceMap.Outputs);
|
||||
|
||||
return new()
|
||||
{
|
||||
|
|
@ -92,8 +24,8 @@ public class DefaultActivityExecutionMapper : IActivityExecutionMapper
|
|||
WorkflowInstanceId = source.WorkflowExecutionContext.Id,
|
||||
ActivityType = source.Activity.Type,
|
||||
ActivityName = source.Activity.Name,
|
||||
ActivityState = inputs,
|
||||
Outputs = outputs,
|
||||
ActivityState = persistableInputs,
|
||||
Outputs = persistableOutputs,
|
||||
Properties = source.Properties,
|
||||
Payload = payload,
|
||||
Exception = ExceptionState.FromException(source.Exception),
|
||||
|
|
@ -105,162 +37,25 @@ public class DefaultActivityExecutionMapper : IActivityExecutionMapper
|
|||
};
|
||||
}
|
||||
|
||||
public async Task<Dictionary<string, object?>> GetPersistableOutputAsync(ActivityExecutionContext context)
|
||||
/// <inheritdoc />
|
||||
public Task<ActivityExecutionRecord> MapAsync(ActivityExecutionContext source)
|
||||
{
|
||||
var cancellationToken = context.WorkflowExecutionContext.CancellationToken;
|
||||
var legacyActivityPersistenceProperties = context.Activity.CustomProperties.GetValueOrDefault<IDictionary<string, object?>>(LegacyLogPersistenceModeKey, () => new Dictionary<string, object?>());
|
||||
var rootActivityExecutionContext = context.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.ParentActivityExecutionContext == null);
|
||||
var workflow = (Workflow?)context.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? context.WorkflowExecutionContext.Workflow;
|
||||
var workflowPersistenceProperty = await GetDefaultPersistenceModeAsync(rootActivityExecutionContext.ExpressionExecutionContext, workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken);
|
||||
var activityPersistencePropertyDefault = await GetDefaultPersistenceModeAsync(context.ExpressionExecutionContext, context.Activity.CustomProperties, () => workflowPersistenceProperty, cancellationToken);
|
||||
var activityPersistenceProperties = context.Activity.CustomProperties.GetValueOrDefault<IDictionary<string, object?>>(LogPersistenceConfigKey, () => new Dictionary<string, object?>());
|
||||
|
||||
return await GetPersistableOutputAsync(
|
||||
context,
|
||||
legacyActivityPersistenceProperties,
|
||||
activityPersistenceProperties,
|
||||
activityPersistencePropertyDefault,
|
||||
cancellationToken);
|
||||
return Task.FromResult(Map(source));
|
||||
}
|
||||
|
||||
private async Task<Dictionary<string, object?>> GetPersistableOutputAsync(
|
||||
ActivityExecutionContext context,
|
||||
IDictionary<string, object?> legacyActivityPersistenceProperties,
|
||||
IDictionary<string, object?> activityPersistenceProperties,
|
||||
LogPersistenceMode activityPersistencePropertyDefault,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var outputs = GetOutputs(context);
|
||||
return await GetPersistablePropertiesAsync(context, outputs, "outputs", legacyActivityPersistenceProperties, activityPersistenceProperties, activityPersistencePropertyDefault, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<Dictionary<string, object?>> GetPersistableInputAsync(ActivityExecutionContext context,
|
||||
IDictionary<string, object?> legacyActivityPersistenceProperties,
|
||||
IDictionary<string, object?> activityPersistenceProperties,
|
||||
LogPersistenceMode activityPersistencePropertyDefault,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await GetPersistablePropertiesAsync(
|
||||
context,
|
||||
context.ActivityState!,
|
||||
"inputs",
|
||||
legacyActivityPersistenceProperties,
|
||||
activityPersistenceProperties,
|
||||
activityPersistencePropertyDefault,
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<Dictionary<string, object?>> GetPersistablePropertiesAsync(
|
||||
ActivityExecutionContext context,
|
||||
IDictionary<string, object?> state,
|
||||
string key,
|
||||
IDictionary<string, object?> legacyActivityPersistenceProperties,
|
||||
IDictionary<string, object?> activityPersistenceProperties,
|
||||
LogPersistenceMode activityPersistencePropertyDefault,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await FilterPropertiesUsingPersistenceMode(
|
||||
context.ExpressionExecutionContext,
|
||||
state,
|
||||
legacyActivityPersistenceProperties!.GetValueOrDefault(key, () => new Dictionary<string, object>())!,
|
||||
activityPersistenceProperties!.GetValueOrDefault(key, () => new Dictionary<string, object>())!,
|
||||
activityPersistencePropertyDefault,
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<LogPersistenceMode> GetDefaultPersistenceModeAsync(ExpressionExecutionContext expressionExecutionContext, IDictionary<string, object> customProperties, Func<LogPersistenceMode> defaultFactory, CancellationToken cancellationToken)
|
||||
{
|
||||
var legacyProperties = customProperties.GetValueOrDefault<IDictionary<string, object?>>(LegacyLogPersistenceModeKey, () => new Dictionary<string, object?>());
|
||||
var properties = customProperties.GetValueOrDefault<IDictionary<string, object>>(LogPersistenceConfigKey, () => new Dictionary<string, object>());
|
||||
var defaultPersistenceConfigObject = properties?.TryGetValue("default", out var defaultPersistenceConfigObjectValue) == true ? defaultPersistenceConfigObjectValue : null;
|
||||
|
||||
if (defaultPersistenceConfigObject == null)
|
||||
{
|
||||
var legacyPersistencePropertyDefault = legacyProperties!.GetValueOrDefault("default", defaultFactory);
|
||||
|
||||
if (legacyPersistencePropertyDefault == LogPersistenceMode.Inherit)
|
||||
return defaultFactory();
|
||||
return legacyPersistencePropertyDefault;
|
||||
}
|
||||
|
||||
var defaultPersistenceConfig = Convert(defaultPersistenceConfigObject);
|
||||
return await EvaluateLogPersistenceConfigAsync(defaultPersistenceConfig, expressionExecutionContext, defaultFactory, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<Dictionary<string, object?>> FilterPropertiesUsingPersistenceMode(
|
||||
ExpressionExecutionContext expressionExecutionContext,
|
||||
IDictionary<string, object?> state,
|
||||
IDictionary<string, object> obsoletePersistenceModeConfiguration,
|
||||
IDictionary<string, object> persistenceStrategyConfiguration,
|
||||
LogPersistenceMode defaultLogPersistenceMode,
|
||||
CancellationToken cancellationToken)
|
||||
private IDictionary<string, object?> GetPersistableProperties(IDictionary<string, object> state, IDictionary<string, LogPersistenceMode> map)
|
||||
{
|
||||
var result = new Dictionary<string, object?>();
|
||||
|
||||
foreach (var value in state)
|
||||
foreach (var stateEntry in state)
|
||||
{
|
||||
var propKey = value.Key.Camelize();
|
||||
var logPersistenceConfigObject = persistenceStrategyConfiguration.GetValueOrDefault(propKey, () => null);
|
||||
var logPersistenceConfig = Convert(logPersistenceConfigObject);
|
||||
var mode = logPersistenceConfig != null
|
||||
? await EvaluateLogPersistenceConfigAsync(logPersistenceConfig, expressionExecutionContext, () => defaultLogPersistenceMode, cancellationToken)
|
||||
: obsoletePersistenceModeConfiguration.GetValueOrDefault(propKey, () => defaultLogPersistenceMode);
|
||||
|
||||
if (mode == LogPersistenceMode.Include || mode == LogPersistenceMode.Inherit && defaultLogPersistenceMode is LogPersistenceMode.Include or LogPersistenceMode.Inherit)
|
||||
result.Add(value.Key, value.Value);
|
||||
var mode = map.TryGetValue(stateEntry.Key, out var value) ? value : LogPersistenceMode.Include;
|
||||
if (mode == LogPersistenceMode.Include)
|
||||
result.Add(stateEntry.Key, stateEntry.Value);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private LogPersistenceConfiguration? Convert(object? value)
|
||||
{
|
||||
if (value == null)
|
||||
return null;
|
||||
|
||||
if (value is LogPersistenceConfiguration c)
|
||||
return c;
|
||||
|
||||
var json = JsonSerializer.Serialize(value);
|
||||
var config = JsonSerializer.Deserialize<LogPersistenceConfiguration>(json, _logPersistenceConfigSerializerOptions);
|
||||
|
||||
return config;
|
||||
}
|
||||
|
||||
private async Task<LogPersistenceMode> EvaluateLogPersistenceConfigAsync(LogPersistenceConfiguration? config, ExpressionExecutionContext executionContext, Func<LogPersistenceMode> defaultMode, CancellationToken cancellationToken)
|
||||
{
|
||||
if (config == null)
|
||||
return defaultMode();
|
||||
|
||||
if (config.EvaluationMode == LogPersistenceEvaluationMode.Strategy)
|
||||
{
|
||||
var strategyTypeName = config.StrategyType ?? typeof(Inherit).GetSimpleAssemblyQualifiedName();
|
||||
var strategy = _logPersistenceStrategies.TryGetValue(strategyTypeName, out var v) ? v : null;
|
||||
|
||||
if (strategy == null)
|
||||
return defaultMode();
|
||||
|
||||
var strategyContext = new LogPersistenceStrategyContext(cancellationToken);
|
||||
var logMode = await strategy.GetPersistenceModeAsync(strategyContext);
|
||||
return logMode == LogPersistenceMode.Inherit ? defaultMode() : logMode;
|
||||
}
|
||||
|
||||
if (config.Expression == null)
|
||||
return defaultMode();
|
||||
|
||||
var expression = config.Expression;
|
||||
|
||||
try
|
||||
{
|
||||
return await _expressionEvaluator.EvaluateAsync<LogPersistenceMode>(expression, executionContext);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogWarning(e, "Error evaluating log persistence expression");
|
||||
return defaultMode();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static IDictionary<string, object> GetPayload(ActivityExecutionContext source)
|
||||
{
|
||||
var outcomes = source.JournalData.TryGetValue("Outcomes", out var resultValue) ? resultValue as string[] : null;
|
||||
|
|
@ -271,30 +66,4 @@ public class DefaultActivityExecutionMapper : IActivityExecutionMapper
|
|||
|
||||
return payload;
|
||||
}
|
||||
|
||||
private static IDictionary<string, object?> GetOutputs(ActivityExecutionContext source)
|
||||
{
|
||||
var activity = source.Activity;
|
||||
var expressionExecutionContext = source.ExpressionExecutionContext;
|
||||
var activityDescriptor = source.ActivityDescriptor;
|
||||
var outputDescriptors = activityDescriptor.Outputs;
|
||||
|
||||
var outputs = outputDescriptors.ToDictionary(x => x.Name, x =>
|
||||
{
|
||||
if (x.IsSerializable == false)
|
||||
return "(not serializable)";
|
||||
|
||||
var cachedValue = activity.GetOutput(expressionExecutionContext, x.Name);
|
||||
|
||||
if (cachedValue != null)
|
||||
return cachedValue;
|
||||
|
||||
if (x.ValueGetter(activity) is Output output && source.TryGet(output.MemoryBlockReference(), out var outputValue))
|
||||
return outputValue;
|
||||
|
||||
return null;
|
||||
});
|
||||
|
||||
return outputs;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue