Incremental work on correlation
This commit is contained in:
Sipke Schoorstra
parent
78a799edb9
commit
3120429f89
|
|
@ -11,15 +11,21 @@ namespace Elsa.Api.ApiResults;
|
|||
|
||||
public class DispatchWorkflowResult : IResult
|
||||
{
|
||||
public DispatchWorkflowResult(Workflow workflow) => Workflow = workflow;
|
||||
public DispatchWorkflowResult(Workflow workflow, string? correlationId = default)
|
||||
{
|
||||
Workflow = workflow;
|
||||
CorrelationId = correlationId;
|
||||
}
|
||||
|
||||
public Workflow Workflow { get; }
|
||||
public string? CorrelationId { get; set; }
|
||||
|
||||
public async Task ExecuteAsync(HttpContext httpContext)
|
||||
{
|
||||
var response = httpContext.Response;
|
||||
var workflowDispatcher = httpContext.RequestServices.GetRequiredService<IWorkflowDispatcher>();
|
||||
var definitionId = Workflow.Identity.DefinitionId;
|
||||
var result = await workflowDispatcher.DispatchAsync(new DispatchWorkflowDefinitionRequest(definitionId, VersionOptions.Published));
|
||||
var result = await workflowDispatcher.DispatchAsync(new DispatchWorkflowDefinitionRequest(definitionId, VersionOptions.Published, CorrelationId: CorrelationId));
|
||||
|
||||
response.StatusCode = (int)HttpStatusCode.OK;
|
||||
await response.WriteAsJsonAsync(result, httpContext.RequestAborted);
|
||||
|
|
|
|||
|
|
@ -9,9 +9,9 @@ namespace Elsa.Api.Endpoints.Workflows;
|
|||
|
||||
public static partial class Workflows
|
||||
{
|
||||
public static async Task<IResult> DispatchAsync(string definitionId, IWorkflowRegistry workflowRegistry, HttpResponse response, CancellationToken cancellationToken)
|
||||
public static async Task<IResult> DispatchAsync(string definitionId, IWorkflowRegistry workflowRegistry, HttpResponse response, string? correlationId = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflow = await workflowRegistry.FindByIdAsync(definitionId, VersionOptions.Published, cancellationToken);
|
||||
return workflow == null ? Results.NotFound() : new DispatchWorkflowResult(workflow);
|
||||
return workflow == null ? Results.NotFound() : new DispatchWorkflowResult(workflow, correlationId);
|
||||
}
|
||||
}
|
||||
|
|
@ -44,6 +44,12 @@ public class WorkflowDefinitionBuilder : IWorkflowDefinitionBuilder
|
|||
return this;
|
||||
}
|
||||
|
||||
public IWorkflowDefinitionBuilder WithVariables(params Variable[] variables)
|
||||
{
|
||||
foreach (var variable in variables) Variables.Add(variable);
|
||||
return this;
|
||||
}
|
||||
|
||||
public IWorkflowDefinitionBuilder WithMetadata(string name, object value)
|
||||
{
|
||||
Metadata[name] = value;
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ public interface IActivity
|
|||
/// <summary>
|
||||
/// Can contain application-specific information about this activity.
|
||||
/// </summary>
|
||||
IDictionary<string, object> ApplicationProperties { get; set; }
|
||||
IDictionary<string, object?> ApplicationProperties { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Invoked when the activity executes.
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ public interface IWorkflowDefinitionBuilder
|
|||
IWorkflowDefinitionBuilder WithVersion(int version);
|
||||
IWorkflowDefinitionBuilder WithRoot(IActivity root);
|
||||
IWorkflowDefinitionBuilder WithVariable(Variable variable);
|
||||
IWorkflowDefinitionBuilder WithVariables(params Variable[] variables);
|
||||
IWorkflowDefinitionBuilder WithApplicationProperty(string name, object value);
|
||||
Workflow BuildWorkflow();
|
||||
}
|
||||
|
|
@ -9,15 +9,15 @@
|
|||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<!-- <PackageReference Include="Dahomey.Json" Version="1.12.2" />-->
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="6.0.1" />
|
||||
<PackageReference Include="Microsoft.Extensions.Options" Version="6.0.0" />
|
||||
<PackageReference Include="System.ComponentModel.Annotations" Version="6.0.0-preview.4.21253.7" />
|
||||
<PackageReference Include="System.Text.Json" Version="6.0.2" />
|
||||
<PackageReference Include="Dahomey.Json" Version="1.12.2"/>
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="6.0.1"/>
|
||||
<PackageReference Include="Microsoft.Extensions.Options" Version="6.0.0"/>
|
||||
<PackageReference Include="System.ComponentModel.Annotations" Version="6.0.0-preview.4.21253.7"/>
|
||||
<PackageReference Include="System.Text.Json" Version="6.0.2"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\..\..\Dahomey.Json\src\Dahomey.Json\Dahomey.Json.csproj" />
|
||||
</ItemGroup>
|
||||
<!-- <ItemGroup>-->
|
||||
<!-- <ProjectReference Include="..\..\..\..\..\Dahomey.Json\src\Dahomey.Json\Dahomey.Json.csproj" />-->
|
||||
<!-- </ItemGroup>-->
|
||||
|
||||
</Project>
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ namespace Elsa;
|
|||
public static class DictionaryExtensions
|
||||
{
|
||||
public static bool TryGetValue<T>(this IDictionary<string, object?> dictionary, string key, out T? value) => dictionary.TryGetValue<string, T>(key, out value);
|
||||
public static bool TryGetValue<T>(this IDictionary<object, object?> dictionary, string key, out T? value) => dictionary.TryGetValue<object, T>(key, out value);
|
||||
|
||||
public static bool TryGetValue<TKey, T>(this IDictionary<TKey, object?> dictionary, TKey key, out T? value)
|
||||
{
|
||||
|
|
@ -18,10 +19,12 @@ public static class DictionaryExtensions
|
|||
return true;
|
||||
}
|
||||
|
||||
public static T? GetValue<TKey, T>(this IDictionary<TKey, object?> dictionary, TKey key) => ConvertValue<T>(dictionary[key]);
|
||||
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<TKey, T>(this IDictionary<TKey, object?> dictionary, TKey key, Func<T> valueFactory) where T : notnull
|
||||
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)
|
||||
{
|
||||
if (dictionary.TryGetValue<TKey, T>(key, out var value))
|
||||
return value;
|
||||
|
|
|
|||
|
|
@ -21,11 +21,6 @@ public abstract class Activity : IActivity
|
|||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public ValueTask<IEnumerable<object>> GetBookmarkPayloadsAsync(TriggerIndexingContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
throw new NotImplementedException();
|
||||
}
|
||||
|
||||
protected virtual void Execute(ActivityExecutionContext context)
|
||||
{
|
||||
}
|
||||
|
|
|
|||
|
|
@ -26,18 +26,18 @@ public class Variable : RegisterLocationReference
|
|||
|
||||
public class Variable<T> : Variable
|
||||
{
|
||||
public Variable() : base(default(T))
|
||||
public Variable() : base(typeof(T).Name, default(T))
|
||||
{
|
||||
}
|
||||
|
||||
public Variable(T value) : base(value ?? default)
|
||||
public Variable(T value) : base(typeof(T).Name, value ?? default)
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
public Variable(string name, T value = default!) : base(name, value ?? default)
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
public new T? Get(ActivityExecutionContext context) => base.Get(context).ConvertTo<T?>();
|
||||
public new T? Get(ExpressionExecutionContext context) => base.Get(context).ConvertTo<T?>();
|
||||
}
|
||||
|
|
@ -16,6 +16,8 @@ public class WorkflowExecutionContext
|
|||
|
||||
public WorkflowExecutionContext(
|
||||
IServiceProvider serviceProvider,
|
||||
string id,
|
||||
string correlationId,
|
||||
Workflow workflow,
|
||||
ActivityNode graph,
|
||||
IActivityScheduler scheduler,
|
||||
|
|
@ -27,7 +29,8 @@ public class WorkflowExecutionContext
|
|||
_serviceProvider = serviceProvider;
|
||||
Workflow = workflow;
|
||||
Graph = graph;
|
||||
Id = Guid.NewGuid().ToString("N");
|
||||
Id = id;
|
||||
CorrelationId = correlationId;
|
||||
_nodes = graph.Flatten().Distinct().ToList();
|
||||
Scheduler = scheduler;
|
||||
Bookmark = bookmark;
|
||||
|
|
@ -43,6 +46,7 @@ public class WorkflowExecutionContext
|
|||
public ActivityNode Graph { get; }
|
||||
public Register Register { get; }
|
||||
public string Id { get; set; }
|
||||
public string CorrelationId { get; set; }
|
||||
public IReadOnlyCollection<ActivityNode> Nodes => new ReadOnlyCollection<ActivityNode>(_nodes);
|
||||
public IDictionary<string, ActivityNode> NodeIdLookup { get; }
|
||||
public IDictionary<IActivity, ActivityNode> NodeActivityLookup { get; }
|
||||
|
|
|
|||
|
|
@ -26,6 +26,6 @@ public class ActivityExecutionPipeline : IActivityExecutionPipeline
|
|||
public async Task ExecuteAsync(ActivityExecutionContext context) => await Pipeline(context);
|
||||
|
||||
private ActivityMiddlewareDelegate CreateDefaultPipeline() => Setup(x => x
|
||||
.UseActivityDrivers()
|
||||
.UseDefaultActivityInvoker()
|
||||
);
|
||||
}
|
||||
|
|
@ -1,13 +1,16 @@
|
|||
using System.Text.Json;
|
||||
using Elsa.Contracts;
|
||||
using Elsa.Models;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Delegate = System.Delegate;
|
||||
using JsonNode = System.Text.Json.Nodes.JsonNode;
|
||||
using JsonObject = System.Text.Json.Nodes.JsonObject;
|
||||
|
||||
namespace Elsa.Pipelines.ActivityExecution.Components;
|
||||
|
||||
public static class InvokeDriversMiddlewareExtensions
|
||||
{
|
||||
public static IActivityExecutionBuilder UseActivityDrivers(this IActivityExecutionBuilder builder) => builder.UseMiddleware<ActivityInvokerMiddleware>();
|
||||
public static IActivityExecutionBuilder UseDefaultActivityInvoker(this IActivityExecutionBuilder builder) => builder.UseMiddleware<ActivityInvokerMiddleware>();
|
||||
}
|
||||
|
||||
public class ActivityInvokerMiddleware : IActivityExecutionMiddleware
|
||||
|
|
@ -41,7 +44,8 @@ public class ActivityInvokerMiddleware : IActivityExecutionMiddleware
|
|||
await executeDelegate(context);
|
||||
|
||||
// Record executed event.
|
||||
LogExecutionRecord(context, WorkflowExecutionLogEventNames.Executed);
|
||||
var payload = context.JournalData.Any() ? context.JournalData : default;
|
||||
LogExecutionRecord(context, WorkflowExecutionLogEventNames.Executed, payload: payload);
|
||||
|
||||
// Reset execute delegate.
|
||||
workflowExecution.ExecuteDelegate = null;
|
||||
|
|
|
|||
|
|
@ -20,8 +20,7 @@ public class ExpressionEvaluator : IExpressionEvaluator
|
|||
|
||||
if (handler != null)
|
||||
return await handler.EvaluateAsync(input, returnType, context);
|
||||
|
||||
var expressionType = input.GetType().GetGenericTypeDefinition();
|
||||
throw new InvalidOperationException($"Could not find handler for expression type {expressionType}");
|
||||
|
||||
throw new InvalidOperationException($"Could not find handler for expression type {input.GetType().FullName}");
|
||||
}
|
||||
}
|
||||
|
|
@ -23,6 +23,9 @@ public class ExpressionHandlerRegistry : IExpressionHandlerRegistry
|
|||
{
|
||||
var expressionType = input.GetType();
|
||||
|
||||
if (expressionType.IsConstructedGenericType)
|
||||
expressionType = expressionType.BaseType;
|
||||
|
||||
if (!Dictionary.TryGetValue(expressionType, out var handlerType))
|
||||
return null;
|
||||
|
||||
|
|
|
|||
|
|
@ -69,7 +69,7 @@ public class IdentityGraphService : IIdentityGraphService
|
|||
{
|
||||
if (!string.IsNullOrWhiteSpace(activityNode.NodeId))
|
||||
return activityNode.NodeId;
|
||||
|
||||
|
||||
var fullTypeName = activityNode.Activity.TypeName;
|
||||
var shortTypeName = fullTypeName.Split('.').Last();
|
||||
var index = GetNextIndexFor(shortTypeName, identityCounters);
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ public class WorkflowRunner : IWorkflowRunner
|
|||
private readonly IWorkflowStateSerializer _workflowStateSerializer;
|
||||
private readonly IIdentityGraphService _identityGraphService;
|
||||
private readonly IActivitySchedulerFactory _schedulerFactory;
|
||||
private readonly IIdentityGenerator _identityGenerator;
|
||||
|
||||
public WorkflowRunner(
|
||||
IServiceScopeFactory serviceScopeFactory,
|
||||
|
|
@ -20,7 +21,8 @@ public class WorkflowRunner : IWorkflowRunner
|
|||
IWorkflowExecutionPipeline pipeline,
|
||||
IWorkflowStateSerializer workflowStateSerializer,
|
||||
IIdentityGraphService identityGraphService,
|
||||
IActivitySchedulerFactory schedulerFactory)
|
||||
IActivitySchedulerFactory schedulerFactory,
|
||||
IIdentityGenerator identityGenerator)
|
||||
{
|
||||
_serviceScopeFactory = serviceScopeFactory;
|
||||
_activityWalker = activityWalker;
|
||||
|
|
@ -28,6 +30,7 @@ public class WorkflowRunner : IWorkflowRunner
|
|||
_workflowStateSerializer = workflowStateSerializer;
|
||||
_identityGraphService = identityGraphService;
|
||||
_schedulerFactory = schedulerFactory;
|
||||
_identityGenerator = identityGenerator;
|
||||
}
|
||||
|
||||
public async Task<InvokeWorkflowResult> RunAsync(Workflow workflow, IDictionary<string, object>? input, CancellationToken cancellationToken = default)
|
||||
|
|
@ -105,14 +108,12 @@ public class WorkflowRunner : IWorkflowRunner
|
|||
var scheduler = _schedulerFactory.CreateScheduler();
|
||||
|
||||
// Setup a workflow execution context.
|
||||
var workflowExecutionContext = new WorkflowExecutionContext(serviceProvider, workflow, graph, scheduler, bookmark, input, executeActivityDelegate, cancellationToken);
|
||||
var id = workflowState?.Id ?? _identityGenerator.GenerateId();
|
||||
var correlationId = workflowState?.CorrelationId ?? _identityGenerator.GenerateId();
|
||||
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)
|
||||
{
|
||||
var workflowStateService = serviceProvider.GetRequiredService<IWorkflowStateSerializer>();
|
||||
workflowStateService.WriteState(workflowExecutionContext, workflowState);
|
||||
}
|
||||
if (workflowState != null) _workflowStateSerializer.WriteState(workflowExecutionContext, workflowState);
|
||||
|
||||
return workflowExecutionContext;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,7 +19,8 @@ public class WorkflowStateSerializer : IWorkflowStateSerializer
|
|||
{
|
||||
var state = new WorkflowState
|
||||
{
|
||||
Id = workflowExecutionContext.Id
|
||||
Id = workflowExecutionContext.Id,
|
||||
CorrelationId = workflowExecutionContext.CorrelationId
|
||||
};
|
||||
|
||||
//GetOutput(state, workflowExecutionContext);
|
||||
|
|
@ -33,6 +34,7 @@ public class WorkflowStateSerializer : IWorkflowStateSerializer
|
|||
public void WriteState(WorkflowExecutionContext workflowExecutionContext, WorkflowState state)
|
||||
{
|
||||
workflowExecutionContext.Id = state.Id;
|
||||
workflowExecutionContext.CorrelationId = state.CorrelationId;
|
||||
//SetOutput(state, workflowExecutionContext);
|
||||
SetProperties(state, workflowExecutionContext);
|
||||
SetActivityExecutionContexts(state, workflowExecutionContext);
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ namespace Elsa.State;
|
|||
public class WorkflowState
|
||||
{
|
||||
public string Id { get; set; } = default!;
|
||||
public string CorrelationId { get; set; } = default!;
|
||||
public IDictionary<string, IDictionary<string, object?>> ActivityOutput { get; set; } = new Dictionary<string, IDictionary<string, object?>>();
|
||||
public ICollection<CompletionCallbackState> CompletionCallbacks { get; set; } = new List<CompletionCallbackState>();
|
||||
public ICollection<ActivityExecutionContextState> ActivityExecutionContexts { get; set; } = new List<ActivityExecutionContextState>();
|
||||
|
|
|
|||
|
|
@ -63,14 +63,6 @@ public static class ServiceCollectionExtensions
|
|||
.AddLogging();
|
||||
}
|
||||
|
||||
private static IServiceCollection AddDefaultExpressionHandlers(this IServiceCollection services) =>
|
||||
services
|
||||
.AddExpressionHandler<LiteralExpressionHandler, LiteralExpression>()
|
||||
.AddExpressionHandler<DelegateExpressionHandler, DelegateExpression>()
|
||||
.AddExpressionHandler<VariableExpressionHandler, VariableExpression>()
|
||||
.AddExpressionHandler<JsonExpressionHandler, JsonExpression>()
|
||||
.AddExpressionHandler<ElsaExpressionHandler, ElsaExpression>();
|
||||
|
||||
public static IServiceCollection AddWorkflowProvider<T>(this IServiceCollection services) where T : class, IWorkflowProvider => services.AddSingleton<IWorkflowProvider, T>();
|
||||
public static IServiceCollection AddStimulusHandler<T>(this IServiceCollection services) where T : class, IStimulusHandler => services.AddSingleton<IStimulusHandler, T>();
|
||||
public static IServiceCollection AddInstructionInterpreter<T>(this IServiceCollection services) where T : class, IWorkflowInstructionInterpreter => services.AddSingleton<IWorkflowInstructionInterpreter, T>();
|
||||
|
|
|
|||
|
|
@ -9,10 +9,11 @@ namespace Elsa.Modules.Activities.Activities.Workflows;
|
|||
/// <summary>
|
||||
/// Creates a new workflow instance of the specified workflow and dispatches it for execution.
|
||||
/// </summary>
|
||||
[Activity("Elsa", "Workflows","Create a new workflow instance of the specified workflow and dispatches it for execution.")]
|
||||
[Activity("Elsa", "Workflows", "Create a new workflow instance of the specified workflow and dispatches it for execution.")]
|
||||
public class DispatchWorkflow : Activity
|
||||
{
|
||||
public Input<string> WorkflowDefinitionId { get; set; } = new("");
|
||||
public Input<string> CorrelationId { get; set; } = new("");
|
||||
public Input<IDictionary<string, object>>? Input { get; set; }
|
||||
|
||||
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
|
||||
|
|
@ -28,6 +29,7 @@ public class DispatchWorkflow : Activity
|
|||
|
||||
var workflowService = context.GetRequiredService<IWorkflowService>();
|
||||
var input = context.Get(Input);
|
||||
await workflowService.DispatchWorkflowAsync(workflowDefinitionId, VersionOptions.Published, input, context.CancellationToken);
|
||||
var correlationId = CorrelationId.Get(context);
|
||||
await workflowService.DispatchWorkflowAsync(workflowDefinitionId, VersionOptions.Published, input, correlationId, context.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
|
@ -72,10 +72,11 @@ public class Worker : IAsyncDisposable
|
|||
{
|
||||
var payload = new MessageReceivedTriggerPayload(QueueOrTopic, Subscription);
|
||||
var hash = _hasher.Hash(payload);
|
||||
var correlationId = message.CorrelationId;
|
||||
var messageModel = CreateMessageModel(message);
|
||||
var input = new Dictionary<string, object>() { [MessageReceived.InputKey] = messageModel };
|
||||
var stimulus = Stimulus.Standard(BookmarkName, hash, input);
|
||||
var executionResults = (await _workflowService.ExecuteStimulusAsync(stimulus, cancellationToken)).ToList();
|
||||
var stimulus = Stimulus.Standard(BookmarkName, hash, input, correlationId);
|
||||
var executionResults = (await _workflowService.DispatchStimulusAsync(stimulus, cancellationToken)).ToList();
|
||||
|
||||
_logger.LogInformation("Triggered {WorkflowCount} workflows", executionResults.Count);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ public class Delay : Activity
|
|||
var resumeAt = clock.UtcNow.Add(timeSpan);
|
||||
var payload = new DelayPayload(resumeAt);
|
||||
|
||||
context.JournalData.Add("ResumeAt", resumeAt);
|
||||
context.CreateBookmark(payload);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -15,16 +15,18 @@ public class RunWorkflowJob : Job
|
|||
{
|
||||
}
|
||||
|
||||
public RunWorkflowJob(string workflowId)
|
||||
public RunWorkflowJob(string workflowId, string? correlationId = default)
|
||||
{
|
||||
WorkflowId = workflowId;
|
||||
CorrelationId = correlationId;
|
||||
}
|
||||
|
||||
public string WorkflowId { get; set; } = default!;
|
||||
public string? CorrelationId { get; set; }
|
||||
|
||||
protected override async ValueTask ExecuteAsync(JobExecutionContext context)
|
||||
{
|
||||
var request = new DispatchWorkflowDefinitionRequest(WorkflowId, VersionOptions.Published);
|
||||
var request = new DispatchWorkflowDefinitionRequest(WorkflowId, VersionOptions.Published, CorrelationId: CorrelationId);
|
||||
var workflowDispatcher = context.GetRequiredService<IWorkflowDispatcher>();
|
||||
await workflowDispatcher.DispatchAsync(request, context.CancellationToken);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,47 @@
|
|||
using Elsa.Contracts;
|
||||
using Elsa.Modules.WorkflowContexts.Models;
|
||||
|
||||
namespace Elsa.Modules.WorkflowContexts.Extensions;
|
||||
|
||||
public static class ActivityExtensions
|
||||
{
|
||||
private const string ActivityWorkflowContextSettingsKey = "ActivityWorkflowContextSettingsKey";
|
||||
|
||||
public static IDictionary<WorkflowContext, ActivityWorkflowContextSettings> GetWorkflowContextSettings(this IActivity activity)
|
||||
{
|
||||
return activity.ApplicationProperties.GetOrAdd(ActivityWorkflowContextSettingsKey, () => new Dictionary<WorkflowContext, ActivityWorkflowContextSettings>())!;
|
||||
}
|
||||
|
||||
public static TActivity LoadContext<TActivity>(this TActivity activity, WorkflowContext workflowContext) where TActivity : IActivity
|
||||
{
|
||||
var settings = activity.GetActivityWorkflowContextSettings(workflowContext);
|
||||
settings.Load = true;
|
||||
return activity;
|
||||
}
|
||||
|
||||
public static TActivity SaveContext<TActivity>(this TActivity activity, WorkflowContext workflowContext) where TActivity : IActivity
|
||||
{
|
||||
var settings = activity.GetActivityWorkflowContextSettings(workflowContext);
|
||||
settings.Save = true;
|
||||
return activity;
|
||||
}
|
||||
|
||||
public static ActivityWorkflowContextSettings GetActivityWorkflowContextSettings<TActivity>(this TActivity activity, WorkflowContext workflowContext) where TActivity: IActivity
|
||||
{
|
||||
var dictionary = activity.GetWorkflowContextSettings()!;
|
||||
return dictionary.GetActivityWorkflowContextSettings(workflowContext);
|
||||
}
|
||||
|
||||
public static ActivityWorkflowContextSettings GetActivityWorkflowContextSettings(this IDictionary<WorkflowContext, ActivityWorkflowContextSettings> dictionary, WorkflowContext workflowContext)
|
||||
{
|
||||
var settings = dictionary.ContainsKey(workflowContext) ? dictionary[workflowContext] : default;
|
||||
|
||||
if(settings == null)
|
||||
{
|
||||
settings = new ActivityWorkflowContextSettings();
|
||||
dictionary[workflowContext] = settings;
|
||||
}
|
||||
|
||||
return settings;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
using Elsa.Contracts;
|
||||
using Elsa.Modules.WorkflowContexts.Middleware;
|
||||
using Elsa.Pipelines.ActivityExecution;
|
||||
using Elsa.Pipelines.WorkflowExecution;
|
||||
|
||||
namespace Elsa.Modules.WorkflowContexts.Extensions;
|
||||
|
|
@ -9,5 +10,10 @@ public static class WorkflowExecutionBuilderExtensions
|
|||
/// <summary>
|
||||
/// Installs middleware that enables the use of workflow context.
|
||||
/// </summary>
|
||||
public static IWorkflowExecutionBuilder UseWorkflowContexts(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware<WorkflowContextMiddleware>();
|
||||
public static IWorkflowExecutionBuilder UseWorkflowContexts(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware<WorkflowContextWorkflowExecutionMiddleware>();
|
||||
|
||||
/// <summary>
|
||||
/// Installs middleware that enables the use of workflow context.
|
||||
/// </summary>
|
||||
public static IActivityExecutionBuilder UseWorkflowContexts(this IActivityExecutionBuilder builder) => builder.UseMiddleware<WorkflowContextActivityExecutionMiddleware>();
|
||||
}
|
||||
|
|
@ -7,8 +7,8 @@ public static class WorkflowExecutionContextExtensions
|
|||
{
|
||||
public static void SetWorkflowContext(this WorkflowExecutionContext workflowExecutionContext, WorkflowContext workflowContext, object? value) => workflowExecutionContext.TransientProperties.SetWorkflowContext(workflowContext, value);
|
||||
public static void SetWorkflowContext(this ExpressionExecutionContext expressionExecutionContext, WorkflowContext workflowContext, object? value) => expressionExecutionContext.TransientProperties.SetWorkflowContext(workflowContext, value);
|
||||
public static T? GetWorkflowContext<T, TProvider>(this WorkflowExecutionContext workflowExecutionContext) => (T?)workflowExecutionContext.GetWorkflowContext(new WorkflowContext(typeof(TProvider)));
|
||||
public static T? GetWorkflowContext<T, TProvider>(this ExpressionExecutionContext expressionExecutionContext) => (T?)expressionExecutionContext.TransientProperties.GetWorkflowContext(new WorkflowContext(typeof(TProvider)));
|
||||
public static T? GetWorkflowContext<T, TProvider>(this WorkflowExecutionContext workflowExecutionContext) => (T?)workflowExecutionContext.TransientProperties.GetWorkflowContextByProviderType(typeof(TProvider));
|
||||
public static T? GetWorkflowContext<T, TProvider>(this ExpressionExecutionContext expressionExecutionContext) => (T?)expressionExecutionContext.TransientProperties.GetWorkflowContextByProviderType(typeof(TProvider));
|
||||
public static object? GetWorkflowContext(this WorkflowExecutionContext workflowExecutionContext, WorkflowContext workflowContext) => workflowExecutionContext.TransientProperties.GetWorkflowContext(workflowContext);
|
||||
public static object? GetWorkflowContext(this ExpressionExecutionContext expressionExecutionContext, WorkflowContext workflowContext) => expressionExecutionContext.TransientProperties.GetWorkflowContext(workflowContext);
|
||||
|
||||
|
|
@ -20,7 +20,25 @@ public static class WorkflowExecutionContextExtensions
|
|||
|
||||
private static object? GetWorkflowContext(this IDictionary<object, object?> transientProperties, WorkflowContext workflowContext)
|
||||
{
|
||||
var contextDictionary = transientProperties.GetOrAdd("WorkflowContexts", () => new Dictionary<WorkflowContext, object?>())!;
|
||||
var contextDictionary = transientProperties.GetWorkflowContextDictionary();
|
||||
return contextDictionary[workflowContext];
|
||||
}
|
||||
|
||||
private static object? GetWorkflowContextByProviderType(this IDictionary<object, object?> transientProperties, Type providerType) =>
|
||||
transientProperties.FindWorkflowContext(x => x.ProviderType == providerType);
|
||||
|
||||
private static object? FindWorkflowContext(this IDictionary<object, object?> transientProperties, Func<WorkflowContext, bool> filter)
|
||||
{
|
||||
var contextDictionary = transientProperties.GetWorkflowContextDictionary();
|
||||
|
||||
var query =
|
||||
from entry in contextDictionary
|
||||
where filter(entry.Key)
|
||||
select entry.Value;
|
||||
|
||||
return query.FirstOrDefault();
|
||||
}
|
||||
|
||||
private static IDictionary<WorkflowContext, object?> GetWorkflowContextDictionary(this IDictionary<object, object?> transientProperties) =>
|
||||
transientProperties.GetOrAdd("WorkflowContexts", () => new Dictionary<WorkflowContext, object?>())!;
|
||||
}
|
||||
|
|
@ -0,0 +1,69 @@
|
|||
using Elsa.Contracts;
|
||||
using Elsa.Models;
|
||||
using Elsa.Modules.WorkflowContexts.Contracts;
|
||||
using Elsa.Modules.WorkflowContexts.Extensions;
|
||||
using Elsa.Modules.WorkflowContexts.Models;
|
||||
using Elsa.Pipelines.ActivityExecution;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.Modules.WorkflowContexts.Middleware;
|
||||
|
||||
/// <summary>
|
||||
/// Middleware that loads & save workflow context into the currently executing workflow using installed workflow context providers.
|
||||
/// </summary>
|
||||
public class WorkflowContextActivityExecutionMiddleware : IActivityExecutionMiddleware
|
||||
{
|
||||
private readonly ActivityMiddlewareDelegate _next;
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
|
||||
public WorkflowContextActivityExecutionMiddleware(ActivityMiddlewareDelegate next, IServiceScopeFactory serviceScopeFactory)
|
||||
{
|
||||
_next = next;
|
||||
_serviceScopeFactory = serviceScopeFactory;
|
||||
}
|
||||
|
||||
public async ValueTask InvokeAsync(ActivityExecutionContext context)
|
||||
{
|
||||
// Check if the workflow contains any workflow contexts.
|
||||
if (!context.WorkflowExecutionContext.Workflow.ApplicationProperties!.TryGetValue<ICollection<WorkflowContext>>("Elsa:WorkflowContexts", out var workflowContexts))
|
||||
{
|
||||
await _next(context);
|
||||
return;
|
||||
}
|
||||
|
||||
// For each workflow context, invoke its provider.
|
||||
using (var scope = _serviceScopeFactory.CreateScope())
|
||||
{
|
||||
foreach (var workflowContext in workflowContexts!)
|
||||
{
|
||||
var load = context.Activity.GetActivityWorkflowContextSettings(workflowContext).Load;
|
||||
if (!load) continue;
|
||||
|
||||
var provider = (IWorkflowContextProvider)ActivatorUtilities.GetServiceOrCreateInstance(scope.ServiceProvider, workflowContext.ProviderType);
|
||||
var value = await provider.LoadAsync(context.WorkflowExecutionContext);
|
||||
|
||||
// Store the loaded value into the workflow execution context.
|
||||
context.WorkflowExecutionContext.SetWorkflowContext(workflowContext, value);
|
||||
}
|
||||
}
|
||||
|
||||
// Invoke the next middleware.
|
||||
await _next(context);
|
||||
|
||||
// For each workflow context, invoke its provider to update the context.
|
||||
using (var scope = _serviceScopeFactory.CreateScope())
|
||||
{
|
||||
foreach (var workflowContext in workflowContexts!)
|
||||
{
|
||||
var save = context.Activity.GetActivityWorkflowContextSettings(workflowContext).Load;
|
||||
if (!save) continue;
|
||||
|
||||
// Get the loaded value from the workflow execution context.
|
||||
var value = context.WorkflowExecutionContext.GetWorkflowContext(workflowContext);
|
||||
|
||||
var provider = (IWorkflowContextProvider)ActivatorUtilities.GetServiceOrCreateInstance(scope.ServiceProvider, workflowContext.ProviderType);
|
||||
await provider.SaveAsync(context.WorkflowExecutionContext, value);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -11,12 +11,12 @@ namespace Elsa.Modules.WorkflowContexts.Middleware;
|
|||
/// <summary>
|
||||
/// Middleware that loads & save workflow context into the currently executing workflow using installed workflow context providers.
|
||||
/// </summary>
|
||||
public class WorkflowContextMiddleware : WorkflowExecutionMiddleware
|
||||
public class WorkflowContextWorkflowExecutionMiddleware : WorkflowExecutionMiddleware
|
||||
{
|
||||
private readonly IServiceProvider _serviceProvider;
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
|
||||
public WorkflowContextMiddleware(WorkflowMiddlewareDelegate next, IServiceProvider serviceProvider, IServiceScopeFactory serviceScopeFactory) : base(next)
|
||||
public WorkflowContextWorkflowExecutionMiddleware(WorkflowMiddlewareDelegate next, IServiceProvider serviceProvider, IServiceScopeFactory serviceScopeFactory) : base(next)
|
||||
{
|
||||
_serviceProvider = serviceProvider;
|
||||
_serviceScopeFactory = serviceScopeFactory;
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
namespace Elsa.Modules.WorkflowContexts.Models;
|
||||
|
||||
public class ActivityWorkflowContextSettings
|
||||
{
|
||||
public bool Load { get; set; }
|
||||
public bool Save { get; set; }
|
||||
}
|
||||
|
|
@ -22,7 +22,6 @@ public class WorkflowContext<T, TProvider> : WorkflowContext where TProvider:IWo
|
|||
public T? Get(ExpressionExecutionContext context)
|
||||
{
|
||||
var workflowContexts = (IDictionary<WorkflowContext, object?>)context.TransientProperties["WorkflowContexts"]!;
|
||||
return (T?)workflowContexts[this];
|
||||
|
||||
return workflowContexts.TryGetValue(this, out var workflowContext) ? (T?)workflowContext : default;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,24 +1,25 @@
|
|||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
using Elsa.Persistence.Entities;
|
||||
using Elsa.Persistence.EntityFrameworkCore.Contracts;
|
||||
using Elsa.Serialization;
|
||||
using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
|
||||
|
||||
namespace Elsa.Persistence.EntityFrameworkCore.Handlers.Serialization;
|
||||
|
||||
public class WorkflowExecutionLogRecordSerializer : IEntitySerializer<WorkflowExecutionLogRecord>
|
||||
{
|
||||
private readonly WorkflowSerializerOptionsProvider _workflowSerializerOptionsProvider;
|
||||
private readonly JsonSerializerOptions _options;
|
||||
|
||||
public WorkflowExecutionLogRecordSerializer(WorkflowSerializerOptionsProvider workflowSerializerOptionsProvider)
|
||||
public WorkflowExecutionLogRecordSerializer()
|
||||
{
|
||||
_workflowSerializerOptionsProvider = workflowSerializerOptionsProvider;
|
||||
_options = new JsonSerializerOptions();
|
||||
_options.Converters.Add(new JsonStringEnumConverter());
|
||||
}
|
||||
|
||||
public void Serialize(ElsaDbContext dbContext, WorkflowExecutionLogRecord entity)
|
||||
{
|
||||
var options = _workflowSerializerOptionsProvider.CreatePersistenceOptions();
|
||||
var json = entity.Payload != null ? JsonSerializer.Serialize(entity.Payload, options) : default!;
|
||||
|
||||
var json = entity.Payload != null ? JsonSerializer.Serialize(entity.Payload, _options) : default!;
|
||||
dbContext.Entry(entity).Property("PayloadData").CurrentValue = json;
|
||||
}
|
||||
|
||||
|
|
@ -26,10 +27,6 @@ public class WorkflowExecutionLogRecordSerializer : IEntitySerializer<WorkflowEx
|
|||
{
|
||||
var json = (string?)dbContext.Entry(entity).Property("PayloadData").CurrentValue;
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(json))
|
||||
{
|
||||
var options = _workflowSerializerOptionsProvider.CreatePersistenceOptions();
|
||||
entity.Payload = JsonSerializer.Deserialize<object>(json, options)!;
|
||||
}
|
||||
if (!string.IsNullOrWhiteSpace(json)) entity.Payload = JsonSerializer.Deserialize<object>(json, _options)!;
|
||||
}
|
||||
}
|
||||
|
|
@ -59,6 +59,8 @@ public static class ServiceCollectionExtensions
|
|||
private static ActorSystemConfig GetSystemConfig() =>
|
||||
ActorSystemConfig
|
||||
.Setup()
|
||||
.WithDeveloperSupervisionLogging(true)
|
||||
.WithDeveloperReceiveLogging(TimeSpan.FromHours(1))
|
||||
.WithDeadLetterThrottleCount(3)
|
||||
.WithDeadLetterThrottleInterval(TimeSpan.FromSeconds(10000))
|
||||
.WithDeveloperSupervisionLogging(true)
|
||||
|
|
|
|||
|
|
@ -7,7 +7,6 @@ using Elsa.Persistence.Commands;
|
|||
using Elsa.Persistence.Entities;
|
||||
using Elsa.Persistence.Models;
|
||||
using Elsa.Runtime.Contracts;
|
||||
using Elsa.Runtime.ProtoActor.Services;
|
||||
using Elsa.Runtime.Protos;
|
||||
using Elsa.State;
|
||||
using Proto;
|
||||
|
|
@ -23,38 +22,37 @@ public class WorkflowDefinitionGrain : WorkflowDefinitionGrainBase
|
|||
private readonly ICommandSender _commandSender;
|
||||
private readonly IIdentityGenerator _identityGenerator;
|
||||
private readonly ISystemClock _systemClock;
|
||||
private readonly GrainClientFactory _grainClientFactory;
|
||||
|
||||
public WorkflowDefinitionGrain(
|
||||
IWorkflowRegistry workflowRegistry,
|
||||
ICommandSender commandSender,
|
||||
IIdentityGenerator identityGenerator,
|
||||
IWorkflowRegistry workflowRegistry,
|
||||
ICommandSender commandSender,
|
||||
IIdentityGenerator identityGenerator,
|
||||
ISystemClock systemClock,
|
||||
GrainClientFactory grainClientFactory,
|
||||
IContext context) : base(context)
|
||||
{
|
||||
_workflowRegistry = workflowRegistry;
|
||||
_commandSender = commandSender;
|
||||
_identityGenerator = identityGenerator;
|
||||
_systemClock = systemClock;
|
||||
_grainClientFactory = grainClientFactory;
|
||||
}
|
||||
|
||||
|
||||
public override async Task<ExecuteWorkflowDefinitionResponse> Execute(ExecuteWorkflowDefinitionRequest request)
|
||||
{
|
||||
var workflowDefinitionId = request.Id;
|
||||
var correlationId = request.CorrelationId;
|
||||
var cancellationToken = Context.CancellationToken;
|
||||
var workflowInstance = await CreateWorkflowInstanceAsync(workflowDefinitionId, cancellationToken);
|
||||
var workflowInstance = await CreateWorkflowInstanceAsync(workflowDefinitionId, correlationId, cancellationToken);
|
||||
|
||||
var executeWorkflowInstanceMessage = new ExecuteWorkflowInstanceRequest
|
||||
{
|
||||
Id = workflowInstance.Id,
|
||||
Input = request.Input
|
||||
Input = request.Input,
|
||||
CorrelationId = correlationId
|
||||
};
|
||||
|
||||
var workflowInstanceGrainClient = Context.GetWorkflowInstanceGrain(workflowInstance.Id);
|
||||
var workflowInstanceResponse = await workflowInstanceGrainClient.Execute(executeWorkflowInstanceMessage, cancellationToken);
|
||||
|
||||
|
||||
if (workflowInstanceResponse == null)
|
||||
throw new TimeoutException("Did not receive a response from the WorkflowInstance actor within the configured amount of time.");
|
||||
|
||||
|
|
@ -67,19 +65,21 @@ public class WorkflowDefinitionGrain : WorkflowDefinitionGrainBase
|
|||
return response;
|
||||
}
|
||||
|
||||
private async Task<WorkflowInstance> CreateWorkflowInstanceAsync(string workflowDefinitionId, CancellationToken cancellationToken)
|
||||
private async Task<WorkflowInstance> CreateWorkflowInstanceAsync(string workflowDefinitionId, string? correlationId, CancellationToken cancellationToken)
|
||||
{
|
||||
var workflow = (await _workflowRegistry.FindByIdAsync(workflowDefinitionId, VersionOptions.Published, cancellationToken))!;
|
||||
var workflowInstanceId = _identityGenerator.GenerateId();
|
||||
var correlationId = _identityGenerator.GenerateId();
|
||||
|
||||
if (string.IsNullOrWhiteSpace(correlationId))
|
||||
correlationId = _identityGenerator.GenerateId();
|
||||
|
||||
var workflowInstance = new WorkflowInstance
|
||||
{
|
||||
Id = workflowInstanceId,
|
||||
Version = workflow.Identity.Version,
|
||||
DefinitionId = workflowDefinitionId,
|
||||
DefinitionVersionId = workflow.Identity.Id,
|
||||
CorrelationId = correlationId,
|
||||
CorrelationId = correlationId,
|
||||
CreatedAt = _systemClock.UtcNow,
|
||||
WorkflowStatus = WorkflowStatus.Idle,
|
||||
WorkflowState = new WorkflowState
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ message ExecuteWorkflowDefinitionRequest {
|
|||
string id = 1;
|
||||
string versionOptions = 2;
|
||||
optional Input input = 3;
|
||||
optional string CorrelationId = 4;
|
||||
}
|
||||
|
||||
message ExecuteWorkflowDefinitionResponse {
|
||||
|
|
@ -21,6 +22,7 @@ message ExecuteWorkflowInstanceRequest {
|
|||
string id = 1;
|
||||
optional Bookmark bookmark = 2;
|
||||
optional Input input = 3;
|
||||
optional string CorrelationId = 4;
|
||||
}
|
||||
|
||||
message ExecuteWorkflowInstanceResponse {
|
||||
|
|
|
|||
|
|
@ -29,13 +29,14 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker
|
|||
|
||||
public async Task<InvokeWorkflowResult> InvokeAsync(InvokeWorkflowDefinitionRequest request, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var (definitionId, versionOptions, input) = request;
|
||||
var (definitionId, versionOptions, input, correlationId) = request;
|
||||
|
||||
var message = new ExecuteWorkflowDefinitionRequest
|
||||
{
|
||||
Id = definitionId,
|
||||
VersionOptions = versionOptions.ToString(),
|
||||
Input = input!?.Serialize()
|
||||
Input = input!?.Serialize(),
|
||||
CorrelationId = correlationId ?? ""
|
||||
};
|
||||
|
||||
var client = _grainClientFactory.CreateWorkflowDefinitionGrainClient(definitionId, versionOptions);
|
||||
|
|
@ -52,18 +53,19 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker
|
|||
|
||||
public async Task<InvokeWorkflowResult> InvokeAsync(InvokeWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var (instanceId, bookmark, input) = request;
|
||||
var (instanceId, bookmark, input, correlationId) = request;
|
||||
var bookmarkMessage = MapBookmark(bookmark);
|
||||
|
||||
var message = new ExecuteWorkflowInstanceRequest
|
||||
{
|
||||
Id = instanceId,
|
||||
Bookmark = bookmarkMessage,
|
||||
Input = input!?.Serialize()
|
||||
Input = input!?.Serialize(),
|
||||
CorrelationId = correlationId ?? ""
|
||||
};
|
||||
|
||||
var client = _grainClientFactory.CreateWorkflowInstanceGrainClient(instanceId);
|
||||
var response = await client.Execute(message, cancellationToken);
|
||||
var response = await client.Execute(message, CancellationTokens.FromSeconds(600));
|
||||
|
||||
if (response == null)
|
||||
throw new TimeoutException("Did not receive a response from the WorkflowInstance actor within the configured amount of time.");
|
||||
|
|
|
|||
|
|
@ -9,11 +9,11 @@ namespace Elsa.Runtime.Contracts;
|
|||
/// </summary>
|
||||
public interface IWorkflowService
|
||||
{
|
||||
Task<InvokeWorkflowResult> ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default);
|
||||
Task<InvokeWorkflowResult> ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default);
|
||||
Task<DispatchWorkflowDefinitionResponse> DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default);
|
||||
Task<DispatchWorkflowInstanceResponse> DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default);
|
||||
Task<IEnumerable<ExecuteWorkflowInstructionResult>> ExecuteStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken);
|
||||
Task<IEnumerable<DispatchWorkflowInstructionResult>> DispatchStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken);
|
||||
Task<InvokeWorkflowResult> ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary<string, object>? input = default, string? correlationId = default, CancellationToken cancellationToken = default);
|
||||
Task<InvokeWorkflowResult> ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary<string, object>? input = default, string? correlationId = default, CancellationToken cancellationToken = default);
|
||||
Task<DispatchWorkflowDefinitionResponse> DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary<string, object>? input = default, string? correlationId = default, CancellationToken cancellationToken = default);
|
||||
Task<DispatchWorkflowInstanceResponse> DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary<string, object>? input = default, string? correlationId = default, CancellationToken cancellationToken = default);
|
||||
Task<IEnumerable<ExecuteWorkflowInstructionResult>> ExecuteStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken = default);
|
||||
Task<IEnumerable<DispatchWorkflowInstructionResult>> DispatchStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken = default);
|
||||
|
||||
}
|
||||
|
|
@ -70,11 +70,12 @@ public static class ServiceCollectionExtensions
|
|||
private static ChannelReader<T> CreateChannelReader<T>(IServiceProvider serviceProvider) => serviceProvider.GetRequiredService<Channel<T>>().Reader;
|
||||
private static ChannelWriter<T> CreateChannelWriter<T>(IServiceProvider serviceProvider) => serviceProvider.GetRequiredService<Channel<T>>().Writer;
|
||||
|
||||
private static IServiceCollection AddDefaultExpressionHandlers(this IServiceCollection services) =>
|
||||
public static IServiceCollection AddDefaultExpressionHandlers(this IServiceCollection services) =>
|
||||
services
|
||||
.AddExpressionHandler<LiteralExpressionHandler, LiteralExpression>()
|
||||
.AddExpressionHandler<DelegateExpressionHandler, DelegateExpression>()
|
||||
.AddExpressionHandler<VariableExpressionHandler, VariableExpression>()
|
||||
.AddExpressionHandler<OutputExpressionHandler, OutputExpression>()
|
||||
.AddExpressionHandler<ElsaExpressionHandler, ElsaExpression>();
|
||||
|
||||
public static IServiceCollection AddWorkflowProvider<T>(this IServiceCollection services) where T : class, IWorkflowProvider => services.AddSingleton<IWorkflowProvider, T>();
|
||||
|
|
|
|||
|
|
@ -25,9 +25,9 @@ public class DispatchedWorkflowDefinitionWorker : BackgroundService
|
|||
|
||||
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (var (definitionId, versionOptions, input) in _channelReader.ReadAllAsync(cancellationToken))
|
||||
await foreach (var (definitionId, versionOptions, input, correlationId) in _channelReader.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
var request = new InvokeWorkflowDefinitionRequest(definitionId, versionOptions, input);
|
||||
var request = new InvokeWorkflowDefinitionRequest(definitionId, versionOptions, input, correlationId);
|
||||
|
||||
try
|
||||
{
|
||||
|
|
|
|||
|
|
@ -24,9 +24,9 @@ public class DispatchedWorkflowInstanceWorker : BackgroundService
|
|||
|
||||
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (var (instanceId, bookmark, input) in _channelReader.ReadAllAsync(cancellationToken))
|
||||
await foreach (var (instanceId, bookmark, input, correlationId) in _channelReader.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
var request = new InvokeWorkflowInstanceRequest(instanceId, bookmark, input);
|
||||
var request = new InvokeWorkflowInstanceRequest(instanceId, bookmark, input, correlationId);
|
||||
|
||||
try
|
||||
{
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ using Microsoft.Extensions.Logging;
|
|||
|
||||
namespace Elsa.Runtime.Interpreters;
|
||||
|
||||
public record ResumeWorkflowInstruction(WorkflowBookmark WorkflowBookmark, IDictionary<string, object>? Input) : IWorkflowInstruction;
|
||||
public record ResumeWorkflowInstruction(WorkflowBookmark WorkflowBookmark, IDictionary<string, object>? Input, string? CorrelationId) : IWorkflowInstruction;
|
||||
|
||||
public class ResumeWorkflowInstructionInterpreter : WorkflowInstructionInterpreter<ResumeWorkflowInstruction>
|
||||
{
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ using Microsoft.Extensions.Logging;
|
|||
|
||||
namespace Elsa.Runtime.Interpreters;
|
||||
|
||||
public record TriggerWorkflowInstruction(WorkflowTrigger WorkflowTrigger, IDictionary<string, object>? Input) : IWorkflowInstruction;
|
||||
public record TriggerWorkflowInstruction(WorkflowTrigger WorkflowTrigger, IDictionary<string, object>? Input, string? CorrelationId) : IWorkflowInstruction;
|
||||
|
||||
public class TriggerWorkflowInstructionInterpreter : WorkflowInstructionInterpreter<TriggerWorkflowInstruction>
|
||||
{
|
||||
|
|
@ -41,7 +41,7 @@ public class TriggerWorkflowInstructionInterpreter : WorkflowInstructionInterpre
|
|||
return null;
|
||||
|
||||
// Execute workflow.
|
||||
var executeRequest = new InvokeWorkflowDefinitionRequest(workflowId, VersionOptions.Published, instruction.Input);
|
||||
var executeRequest = new InvokeWorkflowDefinitionRequest(workflowId, VersionOptions.Published, instruction.Input, instruction.CorrelationId);
|
||||
var workflowExecutionResult = await _workflowInvoker.InvokeAsync(executeRequest, cancellationToken);
|
||||
|
||||
return new ExecuteWorkflowInstructionResult(workflow, workflowExecutionResult);
|
||||
|
|
@ -59,7 +59,7 @@ public class TriggerWorkflowInstructionInterpreter : WorkflowInstructionInterpre
|
|||
return null;
|
||||
|
||||
// Execute workflow.
|
||||
var dispatchRequest = new DispatchWorkflowDefinitionRequest(definitionId, VersionOptions.Published, instruction.Input);
|
||||
var dispatchRequest = new DispatchWorkflowDefinitionRequest(definitionId, VersionOptions.Published, instruction.Input, instruction.CorrelationId);
|
||||
await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken);
|
||||
|
||||
return new DispatchWorkflowInstructionResult();
|
||||
|
|
|
|||
|
|
@ -2,4 +2,4 @@ using Elsa.Persistence.Models;
|
|||
|
||||
namespace Elsa.Runtime.Models;
|
||||
|
||||
public record DispatchWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary<string, object>? Input = default);
|
||||
public record DispatchWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary<string, object>? Input = default, string? CorrelationId = default);
|
||||
|
|
@ -2,4 +2,4 @@ using Elsa.Models;
|
|||
|
||||
namespace Elsa.Runtime.Models;
|
||||
|
||||
public record DispatchWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary<string, object>? Input = default);
|
||||
public record DispatchWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary<string, object>? Input = default, string? CorrelationId = default);
|
||||
|
|
@ -2,4 +2,4 @@ using Elsa.Persistence.Models;
|
|||
|
||||
namespace Elsa.Runtime.Models;
|
||||
|
||||
public record InvokeWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary<string, object>? Input = default);
|
||||
public record InvokeWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary<string, object>? Input = default, string? CorrelationId = default);
|
||||
|
|
@ -2,4 +2,4 @@ using Elsa.Models;
|
|||
|
||||
namespace Elsa.Runtime.Models;
|
||||
|
||||
public record InvokeWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary<string, object>? Input = default);
|
||||
public record InvokeWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary<string, object>? Input = default, string? CorrelationId = default);
|
||||
|
|
@ -4,7 +4,7 @@ namespace Elsa.Runtime.Models;
|
|||
|
||||
public static class Stimulus
|
||||
{
|
||||
public static StandardStimulus Standard(string activityTypeName, string? hash = default, IDictionary<string, object>? input = default) => new(activityTypeName, hash, input);
|
||||
public static StandardStimulus Standard(string activityTypeName, string? hash, object input) => new(activityTypeName, hash, input.ToDictionary());
|
||||
public static StandardStimulus Standard(string activityTypeName, object input) => new(activityTypeName, default, input.ToDictionary());
|
||||
public static StandardStimulus Standard(string activityTypeName, string? hash = default, IDictionary<string, object>? input = default, string? correlationId = default) => new(activityTypeName, hash, input, correlationId);
|
||||
public static StandardStimulus Standard(string activityTypeName, string? hash, object input, string? correlationId = default) => new(activityTypeName, hash, input.ToDictionary(), correlationId);
|
||||
public static StandardStimulus Standard(string activityTypeName, object input, string? correlationId = default) => new(activityTypeName, default, input.ToDictionary(), correlationId);
|
||||
}
|
||||
|
|
@ -24,27 +24,27 @@ public class WorkflowService : IWorkflowService
|
|||
_stimulusInterpreter = stimulusInterpreter;
|
||||
}
|
||||
|
||||
public async Task<InvokeWorkflowResult> ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default)
|
||||
public async Task<InvokeWorkflowResult> ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary<string, object>? input = default, string? correlationId = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var executeRequest = new InvokeWorkflowDefinitionRequest(definitionId, versionOptions, input);
|
||||
var executeRequest = new InvokeWorkflowDefinitionRequest(definitionId, versionOptions, input, correlationId);
|
||||
return await _workflowInvoker.InvokeAsync(executeRequest, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<InvokeWorkflowResult> ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default)
|
||||
public async Task<InvokeWorkflowResult> ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary<string, object>? input = default, string? correlationId = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var request = new InvokeWorkflowInstanceRequest(instanceId, bookmark, input);
|
||||
var request = new InvokeWorkflowInstanceRequest(instanceId, bookmark, input, correlationId);
|
||||
return await _workflowInvoker.InvokeAsync(request, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<DispatchWorkflowDefinitionResponse> DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default)
|
||||
public async Task<DispatchWorkflowDefinitionResponse> DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary<string, object>? input = default, string? correlationId = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var executeRequest = new DispatchWorkflowDefinitionRequest(definitionId, versionOptions, input);
|
||||
var executeRequest = new DispatchWorkflowDefinitionRequest(definitionId, versionOptions, input, correlationId);
|
||||
return await _workflowDispatcher.DispatchAsync(executeRequest, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<DispatchWorkflowInstanceResponse> DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default)
|
||||
public async Task<DispatchWorkflowInstanceResponse> DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary<string, object>? input = default, string? correlationId = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var request = new DispatchWorkflowInstanceRequest(instanceId, bookmark, input);
|
||||
var request = new DispatchWorkflowInstanceRequest(instanceId, bookmark, input, correlationId);
|
||||
return await _workflowDispatcher.DispatchAsync(request, cancellationToken);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -14,6 +14,6 @@ public class ResumeWorkflowsStimulusHandler : StimulusHandler<StandardStimulus>
|
|||
protected override async ValueTask<IEnumerable<IWorkflowInstruction>> GetInstructionsAsync(StandardStimulus stimulus, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowBookmarks = (await _mediator.RequestAsync(new FindWorkflowBookmarks(stimulus.ActivityTypeName, stimulus.Hash), cancellationToken)).ToList();
|
||||
return workflowBookmarks.Select(x => new ResumeWorkflowInstruction(x, stimulus.Input));
|
||||
return workflowBookmarks.Select(x => new ResumeWorkflowInstruction(x, stimulus.Input, stimulus.CorrelationId));
|
||||
}
|
||||
}
|
||||
|
|
@ -14,6 +14,6 @@ public class TriggerWorkflowsStimulusHandler : StimulusHandler<StandardStimulus>
|
|||
protected override async ValueTask<IEnumerable<IWorkflowInstruction>> GetInstructionsAsync(StandardStimulus stimulus, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowTriggers = (await _mediator.RequestAsync(new FindWorkflowTriggersByName(stimulus.ActivityTypeName, stimulus.Hash), cancellationToken)).ToList();
|
||||
return workflowTriggers.Select(x => new TriggerWorkflowInstruction(x, stimulus.Input));
|
||||
return workflowTriggers.Select(x => new TriggerWorkflowInstruction(x, stimulus.Input, stimulus.CorrelationId));
|
||||
}
|
||||
}
|
||||
|
|
@ -5,4 +5,4 @@ namespace Elsa.Runtime.Stimuli;
|
|||
/// <summary>
|
||||
/// Represents a simple stimulus that targets a specific activity type and optionally a corresponding hash to use as a bookmark lookup.
|
||||
/// </summary>
|
||||
public record StandardStimulus(string ActivityTypeName, string? Hash = default, IDictionary<string, object>? Input = default) : IStimulus;
|
||||
public record StandardStimulus(string ActivityTypeName, string? Hash = default, IDictionary<string, object>? Input = default, string? CorrelationId = default) : IStimulus;
|
||||
|
|
@ -25,7 +25,7 @@ class Program
|
|||
// Configure activity execution pipeline.
|
||||
.ConfigureDefaultActivityExecutionPipeline(pipeline => pipeline
|
||||
.UseLogging()
|
||||
.UseActivityDrivers()
|
||||
.UseDefaultActivityInvoker()
|
||||
)
|
||||
|
||||
// Configure workflow engine execution pipeline.
|
||||
|
|
@ -49,7 +49,7 @@ class Program
|
|||
var workflow13 = new Func<IActivity>(BlockingParallelForEachWorkflow.Create);
|
||||
var workflow14 = new Func<IActivity>(FlowchartWorkflow.Create);
|
||||
|
||||
var workflowFactory = workflow3;
|
||||
var workflowFactory = workflow1;
|
||||
var workflowGraph = workflowFactory();
|
||||
var workflow = Workflow.FromActivity(workflowGraph);
|
||||
|
||||
|
|
|
|||
|
|
@ -5,5 +5,5 @@ namespace Elsa.Samples.Console1.Workflows;
|
|||
|
||||
public static class HelloWorldWorkflow
|
||||
{
|
||||
public static IActivity Create() => new WriteLine("Hello World!");
|
||||
public static IActivity Create() => new WriteLine("Hello World!") { Id = "WriteLine1" };
|
||||
}
|
||||
Loading…
Reference in a new issue