diff --git a/src/api/Elsa.Api/ApiResults/DispatchWorkflowResult.cs b/src/api/Elsa.Api/ApiResults/DispatchWorkflowResult.cs index 931362e63..92e9c9eb0 100644 --- a/src/api/Elsa.Api/ApiResults/DispatchWorkflowResult.cs +++ b/src/api/Elsa.Api/ApiResults/DispatchWorkflowResult.cs @@ -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(); 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); diff --git a/src/api/Elsa.Api/Endpoints/Workflows/Dispatch.cs b/src/api/Elsa.Api/Endpoints/Workflows/Dispatch.cs index e1be129c9..aacc06d77 100644 --- a/src/api/Elsa.Api/Endpoints/Workflows/Dispatch.cs +++ b/src/api/Elsa.Api/Endpoints/Workflows/Dispatch.cs @@ -9,9 +9,9 @@ namespace Elsa.Api.Endpoints.Workflows; public static partial class Workflows { - public static async Task DispatchAsync(string definitionId, IWorkflowRegistry workflowRegistry, HttpResponse response, CancellationToken cancellationToken) + public static async Task 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); } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Builders/WorkflowDefinitionBuilder.cs b/src/core/Elsa.Core/Builders/WorkflowDefinitionBuilder.cs index 6c86e337a..68f63d986 100644 --- a/src/core/Elsa.Core/Builders/WorkflowDefinitionBuilder.cs +++ b/src/core/Elsa.Core/Builders/WorkflowDefinitionBuilder.cs @@ -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; diff --git a/src/core/Elsa.Core/Contracts/IActivity.cs b/src/core/Elsa.Core/Contracts/IActivity.cs index 04fb355a6..13d81d630 100644 --- a/src/core/Elsa.Core/Contracts/IActivity.cs +++ b/src/core/Elsa.Core/Contracts/IActivity.cs @@ -25,7 +25,7 @@ public interface IActivity /// /// Can contain application-specific information about this activity. /// - IDictionary ApplicationProperties { get; set; } + IDictionary ApplicationProperties { get; set; } /// /// Invoked when the activity executes. diff --git a/src/core/Elsa.Core/Contracts/IWorkflowDefinitionBuilder.cs b/src/core/Elsa.Core/Contracts/IWorkflowDefinitionBuilder.cs index 89859fb17..f1e94b4ca 100644 --- a/src/core/Elsa.Core/Contracts/IWorkflowDefinitionBuilder.cs +++ b/src/core/Elsa.Core/Contracts/IWorkflowDefinitionBuilder.cs @@ -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(); } \ No newline at end of file diff --git a/src/core/Elsa.Core/Elsa.Core.csproj b/src/core/Elsa.Core/Elsa.Core.csproj index c2dc36841..ab2dce856 100644 --- a/src/core/Elsa.Core/Elsa.Core.csproj +++ b/src/core/Elsa.Core/Elsa.Core.csproj @@ -9,15 +9,15 @@ - - - - - + + + + + - - - + + + diff --git a/src/core/Elsa.Core/Extensions/DictionaryExtensions.cs b/src/core/Elsa.Core/Extensions/DictionaryExtensions.cs index 1cf3ef8b5..8822f14fe 100644 --- a/src/core/Elsa.Core/Extensions/DictionaryExtensions.cs +++ b/src/core/Elsa.Core/Extensions/DictionaryExtensions.cs @@ -5,6 +5,7 @@ namespace Elsa; public static class DictionaryExtensions { public static bool TryGetValue(this IDictionary dictionary, string key, out T? value) => dictionary.TryGetValue(key, out value); + public static bool TryGetValue(this IDictionary dictionary, string key, out T? value) => dictionary.TryGetValue(key, out value); public static bool TryGetValue(this IDictionary dictionary, TKey key, out T? value) { @@ -18,10 +19,12 @@ public static class DictionaryExtensions return true; } - public static T? GetValue(this IDictionary dictionary, TKey key) => ConvertValue(dictionary[key]); + public static T? GetValue(this IDictionary dictionary, TKey key) => ConvertValue(dictionary[key]); public static T? GetValue(this IDictionary dictionary, string key) => ConvertValue(dictionary[key]); - public static T? GetOrAdd(this IDictionary dictionary, TKey key, Func valueFactory) where T : notnull + public static T? GetOrAdd(this IDictionary dictionary, string key, Func valueFactory) where T : notnull => dictionary.GetOrAdd(key, valueFactory); + + public static T? GetOrAdd(this IDictionary dictionary, TKey key, Func valueFactory) { if (dictionary.TryGetValue(key, out var value)) return value; diff --git a/src/core/Elsa.Core/Models/Activity.cs b/src/core/Elsa.Core/Models/Activity.cs index ff7c9c061..7dcc02a55 100644 --- a/src/core/Elsa.Core/Models/Activity.cs +++ b/src/core/Elsa.Core/Models/Activity.cs @@ -21,11 +21,6 @@ public abstract class Activity : IActivity return ValueTask.CompletedTask; } - public ValueTask> GetBookmarkPayloadsAsync(TriggerIndexingContext context, CancellationToken cancellationToken = default) - { - throw new NotImplementedException(); - } - protected virtual void Execute(ActivityExecutionContext context) { } diff --git a/src/core/Elsa.Core/Models/Variable.cs b/src/core/Elsa.Core/Models/Variable.cs index c7ec48501..edfde66dd 100644 --- a/src/core/Elsa.Core/Models/Variable.cs +++ b/src/core/Elsa.Core/Models/Variable.cs @@ -26,18 +26,18 @@ public class Variable : RegisterLocationReference public class Variable : 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(); public new T? Get(ExpressionExecutionContext context) => base.Get(context).ConvertTo(); } \ No newline at end of file diff --git a/src/core/Elsa.Core/Models/WorkflowExecutionContext.cs b/src/core/Elsa.Core/Models/WorkflowExecutionContext.cs index d911f8d20..c65d885a2 100644 --- a/src/core/Elsa.Core/Models/WorkflowExecutionContext.cs +++ b/src/core/Elsa.Core/Models/WorkflowExecutionContext.cs @@ -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 Nodes => new ReadOnlyCollection(_nodes); public IDictionary NodeIdLookup { get; } public IDictionary NodeActivityLookup { get; } diff --git a/src/core/Elsa.Core/Pipelines/ActivityExecution/ActivityExecutionPipeline.cs b/src/core/Elsa.Core/Pipelines/ActivityExecution/ActivityExecutionPipeline.cs index efdc4242a..f5595bab4 100644 --- a/src/core/Elsa.Core/Pipelines/ActivityExecution/ActivityExecutionPipeline.cs +++ b/src/core/Elsa.Core/Pipelines/ActivityExecution/ActivityExecutionPipeline.cs @@ -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() ); } \ No newline at end of file diff --git a/src/core/Elsa.Core/Pipelines/ActivityExecution/Components/ActivityInvokerMiddleware.cs b/src/core/Elsa.Core/Pipelines/ActivityExecution/Components/ActivityInvokerMiddleware.cs index ea48f84d3..508157831 100644 --- a/src/core/Elsa.Core/Pipelines/ActivityExecution/Components/ActivityInvokerMiddleware.cs +++ b/src/core/Elsa.Core/Pipelines/ActivityExecution/Components/ActivityInvokerMiddleware.cs @@ -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(); + public static IActivityExecutionBuilder UseDefaultActivityInvoker(this IActivityExecutionBuilder builder) => builder.UseMiddleware(); } 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; diff --git a/src/core/Elsa.Core/Services/ExpressionEvaluator.cs b/src/core/Elsa.Core/Services/ExpressionEvaluator.cs index 6a9de0f84..667837e00 100644 --- a/src/core/Elsa.Core/Services/ExpressionEvaluator.cs +++ b/src/core/Elsa.Core/Services/ExpressionEvaluator.cs @@ -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}"); } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Services/ExpressionHandlerRegistry.cs b/src/core/Elsa.Core/Services/ExpressionHandlerRegistry.cs index 0daf4119b..ffc159bc1 100644 --- a/src/core/Elsa.Core/Services/ExpressionHandlerRegistry.cs +++ b/src/core/Elsa.Core/Services/ExpressionHandlerRegistry.cs @@ -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; diff --git a/src/core/Elsa.Core/Services/IdentityGraphService.cs b/src/core/Elsa.Core/Services/IdentityGraphService.cs index 4c18e637f..2f3aa90b5 100644 --- a/src/core/Elsa.Core/Services/IdentityGraphService.cs +++ b/src/core/Elsa.Core/Services/IdentityGraphService.cs @@ -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); diff --git a/src/core/Elsa.Core/Services/WorkflowRunner.cs b/src/core/Elsa.Core/Services/WorkflowRunner.cs index 829ba16d4..bb4c9b44c 100644 --- a/src/core/Elsa.Core/Services/WorkflowRunner.cs +++ b/src/core/Elsa.Core/Services/WorkflowRunner.cs @@ -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 RunAsync(Workflow workflow, IDictionary? 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(); - workflowStateService.WriteState(workflowExecutionContext, workflowState); - } + if (workflowState != null) _workflowStateSerializer.WriteState(workflowExecutionContext, workflowState); return workflowExecutionContext; } diff --git a/src/core/Elsa.Core/Services/WorkflowStateSerializer.cs b/src/core/Elsa.Core/Services/WorkflowStateSerializer.cs index 7b79cba0b..8febcada2 100644 --- a/src/core/Elsa.Core/Services/WorkflowStateSerializer.cs +++ b/src/core/Elsa.Core/Services/WorkflowStateSerializer.cs @@ -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); diff --git a/src/core/Elsa.Core/State/WorkflowState.cs b/src/core/Elsa.Core/State/WorkflowState.cs index 982da5a5b..15b59eea6 100644 --- a/src/core/Elsa.Core/State/WorkflowState.cs +++ b/src/core/Elsa.Core/State/WorkflowState.cs @@ -3,6 +3,7 @@ namespace Elsa.State; public class WorkflowState { public string Id { get; set; } = default!; + public string CorrelationId { get; set; } = default!; public IDictionary> ActivityOutput { get; set; } = new Dictionary>(); public ICollection CompletionCallbacks { get; set; } = new List(); public ICollection ActivityExecutionContexts { get; set; } = new List(); diff --git a/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs b/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs index d27d469c5..8da082816 100644 --- a/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs +++ b/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs @@ -63,14 +63,6 @@ public static class ServiceCollectionExtensions .AddLogging(); } - private static IServiceCollection AddDefaultExpressionHandlers(this IServiceCollection services) => - services - .AddExpressionHandler() - .AddExpressionHandler() - .AddExpressionHandler() - .AddExpressionHandler() - .AddExpressionHandler(); - public static IServiceCollection AddWorkflowProvider(this IServiceCollection services) where T : class, IWorkflowProvider => services.AddSingleton(); public static IServiceCollection AddStimulusHandler(this IServiceCollection services) where T : class, IStimulusHandler => services.AddSingleton(); public static IServiceCollection AddInstructionInterpreter(this IServiceCollection services) where T : class, IWorkflowInstructionInterpreter => services.AddSingleton(); diff --git a/src/modules/Elsa.Modules.Activities/Activities/Workflows/DispatchWorkflow.cs b/src/modules/Elsa.Modules.Activities/Activities/Workflows/DispatchWorkflow.cs index e52e252b0..359ee495b 100644 --- a/src/modules/Elsa.Modules.Activities/Activities/Workflows/DispatchWorkflow.cs +++ b/src/modules/Elsa.Modules.Activities/Activities/Workflows/DispatchWorkflow.cs @@ -9,10 +9,11 @@ namespace Elsa.Modules.Activities.Activities.Workflows; /// /// Creates 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.")] +[Activity("Elsa", "Workflows", "Create a new workflow instance of the specified workflow and dispatches it for execution.")] public class DispatchWorkflow : Activity { public Input WorkflowDefinitionId { get; set; } = new(""); + public Input CorrelationId { get; set; } = new(""); public Input>? Input { get; set; } protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) @@ -28,6 +29,7 @@ public class DispatchWorkflow : Activity var workflowService = context.GetRequiredService(); 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); } } \ No newline at end of file diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Services/Worker.cs b/src/modules/Elsa.Modules.AzureServiceBus/Services/Worker.cs index 46d7cac53..f4108ba76 100644 --- a/src/modules/Elsa.Modules.AzureServiceBus/Services/Worker.cs +++ b/src/modules/Elsa.Modules.AzureServiceBus/Services/Worker.cs @@ -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() { [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); } diff --git a/src/modules/Elsa.Modules.Scheduling/Activities/Delay.cs b/src/modules/Elsa.Modules.Scheduling/Activities/Delay.cs index 27c9f1ef4..36846dc5e 100644 --- a/src/modules/Elsa.Modules.Scheduling/Activities/Delay.cs +++ b/src/modules/Elsa.Modules.Scheduling/Activities/Delay.cs @@ -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); } diff --git a/src/modules/Elsa.Modules.Scheduling/Jobs/RunWorkflowJob.cs b/src/modules/Elsa.Modules.Scheduling/Jobs/RunWorkflowJob.cs index 2b30ce513..84dbb49fc 100644 --- a/src/modules/Elsa.Modules.Scheduling/Jobs/RunWorkflowJob.cs +++ b/src/modules/Elsa.Modules.Scheduling/Jobs/RunWorkflowJob.cs @@ -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(); await workflowDispatcher.DispatchAsync(request, context.CancellationToken); } diff --git a/src/modules/Elsa.Modules.WorkflowContexts/Extensions/ActivityExtensions.cs b/src/modules/Elsa.Modules.WorkflowContexts/Extensions/ActivityExtensions.cs new file mode 100644 index 000000000..08f03fc22 --- /dev/null +++ b/src/modules/Elsa.Modules.WorkflowContexts/Extensions/ActivityExtensions.cs @@ -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 GetWorkflowContextSettings(this IActivity activity) + { + return activity.ApplicationProperties.GetOrAdd(ActivityWorkflowContextSettingsKey, () => new Dictionary())!; + } + + public static TActivity LoadContext(this TActivity activity, WorkflowContext workflowContext) where TActivity : IActivity + { + var settings = activity.GetActivityWorkflowContextSettings(workflowContext); + settings.Load = true; + return activity; + } + + public static TActivity SaveContext(this TActivity activity, WorkflowContext workflowContext) where TActivity : IActivity + { + var settings = activity.GetActivityWorkflowContextSettings(workflowContext); + settings.Save = true; + return activity; + } + + public static ActivityWorkflowContextSettings GetActivityWorkflowContextSettings(this TActivity activity, WorkflowContext workflowContext) where TActivity: IActivity + { + var dictionary = activity.GetWorkflowContextSettings()!; + return dictionary.GetActivityWorkflowContextSettings(workflowContext); + } + + public static ActivityWorkflowContextSettings GetActivityWorkflowContextSettings(this IDictionary dictionary, WorkflowContext workflowContext) + { + var settings = dictionary.ContainsKey(workflowContext) ? dictionary[workflowContext] : default; + + if(settings == null) + { + settings = new ActivityWorkflowContextSettings(); + dictionary[workflowContext] = settings; + } + + return settings; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.WorkflowContexts/Extensions/WorkflowExecutionBuilderExtensions.cs b/src/modules/Elsa.Modules.WorkflowContexts/Extensions/WorkflowExecutionBuilderExtensions.cs index f49046e0b..a1e131048 100644 --- a/src/modules/Elsa.Modules.WorkflowContexts/Extensions/WorkflowExecutionBuilderExtensions.cs +++ b/src/modules/Elsa.Modules.WorkflowContexts/Extensions/WorkflowExecutionBuilderExtensions.cs @@ -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 /// /// Installs middleware that enables the use of workflow context. /// - public static IWorkflowExecutionBuilder UseWorkflowContexts(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware(); + public static IWorkflowExecutionBuilder UseWorkflowContexts(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware(); + + /// + /// Installs middleware that enables the use of workflow context. + /// + public static IActivityExecutionBuilder UseWorkflowContexts(this IActivityExecutionBuilder builder) => builder.UseMiddleware(); } \ No newline at end of file diff --git a/src/modules/Elsa.Modules.WorkflowContexts/Extensions/WorkflowExecutionContextExtensions.cs b/src/modules/Elsa.Modules.WorkflowContexts/Extensions/WorkflowExecutionContextExtensions.cs index afa0a89a4..6dc4a2c18 100644 --- a/src/modules/Elsa.Modules.WorkflowContexts/Extensions/WorkflowExecutionContextExtensions.cs +++ b/src/modules/Elsa.Modules.WorkflowContexts/Extensions/WorkflowExecutionContextExtensions.cs @@ -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(this WorkflowExecutionContext workflowExecutionContext) => (T?)workflowExecutionContext.GetWorkflowContext(new WorkflowContext(typeof(TProvider))); - public static T? GetWorkflowContext(this ExpressionExecutionContext expressionExecutionContext) => (T?)expressionExecutionContext.TransientProperties.GetWorkflowContext(new WorkflowContext(typeof(TProvider))); + public static T? GetWorkflowContext(this WorkflowExecutionContext workflowExecutionContext) => (T?)workflowExecutionContext.TransientProperties.GetWorkflowContextByProviderType(typeof(TProvider)); + public static T? GetWorkflowContext(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 transientProperties, WorkflowContext workflowContext) { - var contextDictionary = transientProperties.GetOrAdd("WorkflowContexts", () => new Dictionary())!; + var contextDictionary = transientProperties.GetWorkflowContextDictionary(); return contextDictionary[workflowContext]; } + + private static object? GetWorkflowContextByProviderType(this IDictionary transientProperties, Type providerType) => + transientProperties.FindWorkflowContext(x => x.ProviderType == providerType); + + private static object? FindWorkflowContext(this IDictionary transientProperties, Func filter) + { + var contextDictionary = transientProperties.GetWorkflowContextDictionary(); + + var query = + from entry in contextDictionary + where filter(entry.Key) + select entry.Value; + + return query.FirstOrDefault(); + } + + private static IDictionary GetWorkflowContextDictionary(this IDictionary transientProperties) => + transientProperties.GetOrAdd("WorkflowContexts", () => new Dictionary())!; } \ No newline at end of file diff --git a/src/modules/Elsa.Modules.WorkflowContexts/Middleware/WorkflowContextActivityExecutionMiddleware.cs b/src/modules/Elsa.Modules.WorkflowContexts/Middleware/WorkflowContextActivityExecutionMiddleware.cs new file mode 100644 index 000000000..74cf443a0 --- /dev/null +++ b/src/modules/Elsa.Modules.WorkflowContexts/Middleware/WorkflowContextActivityExecutionMiddleware.cs @@ -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; + +/// +/// Middleware that loads & save workflow context into the currently executing workflow using installed workflow context providers. +/// +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>("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); + } + } + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.WorkflowContexts/Middleware/WorkflowContextMiddleware.cs b/src/modules/Elsa.Modules.WorkflowContexts/Middleware/WorkflowContextWorkflowExecutionMiddleware.cs similarity index 90% rename from src/modules/Elsa.Modules.WorkflowContexts/Middleware/WorkflowContextMiddleware.cs rename to src/modules/Elsa.Modules.WorkflowContexts/Middleware/WorkflowContextWorkflowExecutionMiddleware.cs index 864e5fe00..723127597 100644 --- a/src/modules/Elsa.Modules.WorkflowContexts/Middleware/WorkflowContextMiddleware.cs +++ b/src/modules/Elsa.Modules.WorkflowContexts/Middleware/WorkflowContextWorkflowExecutionMiddleware.cs @@ -11,12 +11,12 @@ namespace Elsa.Modules.WorkflowContexts.Middleware; /// /// Middleware that loads & save workflow context into the currently executing workflow using installed workflow context providers. /// -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; diff --git a/src/modules/Elsa.Modules.WorkflowContexts/Models/ActivityWorkflowContextSettings.cs b/src/modules/Elsa.Modules.WorkflowContexts/Models/ActivityWorkflowContextSettings.cs new file mode 100644 index 000000000..408e2e25b --- /dev/null +++ b/src/modules/Elsa.Modules.WorkflowContexts/Models/ActivityWorkflowContextSettings.cs @@ -0,0 +1,7 @@ +namespace Elsa.Modules.WorkflowContexts.Models; + +public class ActivityWorkflowContextSettings +{ + public bool Load { get; set; } + public bool Save { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.WorkflowContexts/Models/WorkflowContext.cs b/src/modules/Elsa.Modules.WorkflowContexts/Models/WorkflowContext.cs index 2687f9e6f..2f6dc3a4a 100644 --- a/src/modules/Elsa.Modules.WorkflowContexts/Models/WorkflowContext.cs +++ b/src/modules/Elsa.Modules.WorkflowContexts/Models/WorkflowContext.cs @@ -22,7 +22,6 @@ public class WorkflowContext : WorkflowContext where TProvider:IWo public T? Get(ExpressionExecutionContext context) { var workflowContexts = (IDictionary)context.TransientProperties["WorkflowContexts"]!; - return (T?)workflowContexts[this]; - + return workflowContexts.TryGetValue(this, out var workflowContext) ? (T?)workflowContext : default; } } \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Serialization/WorkflowExecutionLogRecordSerializer.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Serialization/WorkflowExecutionLogRecordSerializer.cs index 55ee5f7a7..cbc25039d 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Serialization/WorkflowExecutionLogRecordSerializer.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Serialization/WorkflowExecutionLogRecordSerializer.cs @@ -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 { - 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(json, options)!; - } + if (!string.IsNullOrWhiteSpace(json)) entity.Payload = JsonSerializer.Deserialize(json, _options)!; } } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ServiceCollectionExtensions.cs b/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ServiceCollectionExtensions.cs index a583718fc..9e55fc3e6 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ServiceCollectionExtensions.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Extensions/ServiceCollectionExtensions.cs @@ -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) diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowDefinitionGrain.cs b/src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowDefinitionGrain.cs index 6f3b3717f..9d184fd42 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowDefinitionGrain.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Grains/WorkflowDefinitionGrain.cs @@ -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 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 CreateWorkflowInstanceAsync(string workflowDefinitionId, CancellationToken cancellationToken) + private async Task 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 diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Protos/Messages.proto b/src/runtime/Elsa.Runtime.ProtoActor/Protos/Messages.proto index 1898e0a95..805ec0f0c 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Protos/Messages.proto +++ b/src/runtime/Elsa.Runtime.ProtoActor/Protos/Messages.proto @@ -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 { diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs b/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs index d3a4c181e..6365b652f 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs @@ -29,13 +29,14 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker public async Task 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 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."); diff --git a/src/runtime/Elsa.Runtime/Contracts/IWorkflowService.cs b/src/runtime/Elsa.Runtime/Contracts/IWorkflowService.cs index 85baed93f..a5ef25f9f 100644 --- a/src/runtime/Elsa.Runtime/Contracts/IWorkflowService.cs +++ b/src/runtime/Elsa.Runtime/Contracts/IWorkflowService.cs @@ -9,11 +9,11 @@ namespace Elsa.Runtime.Contracts; /// public interface IWorkflowService { - Task ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default); - Task ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default); - Task DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default); - Task DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default); - Task> ExecuteStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken); - Task> DispatchStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken); + Task ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, string? correlationId = default, CancellationToken cancellationToken = default); + Task ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, string? correlationId = default, CancellationToken cancellationToken = default); + Task DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, string? correlationId = default, CancellationToken cancellationToken = default); + Task DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, string? correlationId = default, CancellationToken cancellationToken = default); + Task> ExecuteStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken = default); + Task> DispatchStimulusAsync(IStimulus stimulus, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs b/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs index 6efef1d13..49e72e69e 100644 --- a/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs +++ b/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs @@ -70,11 +70,12 @@ public static class ServiceCollectionExtensions private static ChannelReader CreateChannelReader(IServiceProvider serviceProvider) => serviceProvider.GetRequiredService>().Reader; private static ChannelWriter CreateChannelWriter(IServiceProvider serviceProvider) => serviceProvider.GetRequiredService>().Writer; - private static IServiceCollection AddDefaultExpressionHandlers(this IServiceCollection services) => + public static IServiceCollection AddDefaultExpressionHandlers(this IServiceCollection services) => services .AddExpressionHandler() .AddExpressionHandler() .AddExpressionHandler() + .AddExpressionHandler() .AddExpressionHandler(); public static IServiceCollection AddWorkflowProvider(this IServiceCollection services) where T : class, IWorkflowProvider => services.AddSingleton(); diff --git a/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowDefinitionWorker.cs b/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowDefinitionWorker.cs index 62eca7554..a2bc272d5 100644 --- a/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowDefinitionWorker.cs +++ b/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowDefinitionWorker.cs @@ -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 { diff --git a/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowInstanceWorker.cs b/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowInstanceWorker.cs index c1f9614e8..52706bcda 100644 --- a/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowInstanceWorker.cs +++ b/src/runtime/Elsa.Runtime/HostedServices/DispatchedWorkflowInstanceWorker.cs @@ -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 { diff --git a/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs b/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs index 685c01aa5..11cf47c46 100644 --- a/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs +++ b/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs @@ -10,7 +10,7 @@ using Microsoft.Extensions.Logging; namespace Elsa.Runtime.Interpreters; -public record ResumeWorkflowInstruction(WorkflowBookmark WorkflowBookmark, IDictionary? Input) : IWorkflowInstruction; +public record ResumeWorkflowInstruction(WorkflowBookmark WorkflowBookmark, IDictionary? Input, string? CorrelationId) : IWorkflowInstruction; public class ResumeWorkflowInstructionInterpreter : WorkflowInstructionInterpreter { diff --git a/src/runtime/Elsa.Runtime/Interpreters/TriggerWorkflowInstructionInterpreter.cs b/src/runtime/Elsa.Runtime/Interpreters/TriggerWorkflowInstructionInterpreter.cs index 6855df43c..7a0b4af76 100644 --- a/src/runtime/Elsa.Runtime/Interpreters/TriggerWorkflowInstructionInterpreter.cs +++ b/src/runtime/Elsa.Runtime/Interpreters/TriggerWorkflowInstructionInterpreter.cs @@ -8,7 +8,7 @@ using Microsoft.Extensions.Logging; namespace Elsa.Runtime.Interpreters; -public record TriggerWorkflowInstruction(WorkflowTrigger WorkflowTrigger, IDictionary? Input) : IWorkflowInstruction; +public record TriggerWorkflowInstruction(WorkflowTrigger WorkflowTrigger, IDictionary? Input, string? CorrelationId) : IWorkflowInstruction; public class TriggerWorkflowInstructionInterpreter : WorkflowInstructionInterpreter { @@ -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(); diff --git a/src/runtime/Elsa.Runtime/Models/DispatchWorkflowDefinitionRequest.cs b/src/runtime/Elsa.Runtime/Models/DispatchWorkflowDefinitionRequest.cs index 4c4f530be..6d8d2ce9a 100644 --- a/src/runtime/Elsa.Runtime/Models/DispatchWorkflowDefinitionRequest.cs +++ b/src/runtime/Elsa.Runtime/Models/DispatchWorkflowDefinitionRequest.cs @@ -2,4 +2,4 @@ using Elsa.Persistence.Models; namespace Elsa.Runtime.Models; -public record DispatchWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary? Input = default); \ No newline at end of file +public record DispatchWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary? Input = default, string? CorrelationId = default); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/DispatchWorkflowInstanceRequest.cs b/src/runtime/Elsa.Runtime/Models/DispatchWorkflowInstanceRequest.cs index cc8032b15..172949b63 100644 --- a/src/runtime/Elsa.Runtime/Models/DispatchWorkflowInstanceRequest.cs +++ b/src/runtime/Elsa.Runtime/Models/DispatchWorkflowInstanceRequest.cs @@ -2,4 +2,4 @@ using Elsa.Models; namespace Elsa.Runtime.Models; -public record DispatchWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary? Input = default); \ No newline at end of file +public record DispatchWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary? Input = default, string? CorrelationId = default); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/InvokeWorkflowDefinitionRequest.cs b/src/runtime/Elsa.Runtime/Models/InvokeWorkflowDefinitionRequest.cs index f0fcd25f0..586eb3743 100644 --- a/src/runtime/Elsa.Runtime/Models/InvokeWorkflowDefinitionRequest.cs +++ b/src/runtime/Elsa.Runtime/Models/InvokeWorkflowDefinitionRequest.cs @@ -2,4 +2,4 @@ using Elsa.Persistence.Models; namespace Elsa.Runtime.Models; -public record InvokeWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary? Input = default); \ No newline at end of file +public record InvokeWorkflowDefinitionRequest(string DefinitionId, VersionOptions VersionOptions, IDictionary? Input = default, string? CorrelationId = default); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/InvokeWorkflowInstanceRequest.cs b/src/runtime/Elsa.Runtime/Models/InvokeWorkflowInstanceRequest.cs index 7a730f93b..2ff0340d8 100644 --- a/src/runtime/Elsa.Runtime/Models/InvokeWorkflowInstanceRequest.cs +++ b/src/runtime/Elsa.Runtime/Models/InvokeWorkflowInstanceRequest.cs @@ -2,4 +2,4 @@ using Elsa.Models; namespace Elsa.Runtime.Models; -public record InvokeWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary? Input = default); \ No newline at end of file +public record InvokeWorkflowInstanceRequest(string InstanceId, Bookmark? Bookmark = default, IDictionary? Input = default, string? CorrelationId = default); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/Stimulus.cs b/src/runtime/Elsa.Runtime/Models/Stimulus.cs index f00299cd9..f8b6eda44 100644 --- a/src/runtime/Elsa.Runtime/Models/Stimulus.cs +++ b/src/runtime/Elsa.Runtime/Models/Stimulus.cs @@ -4,7 +4,7 @@ namespace Elsa.Runtime.Models; public static class Stimulus { - public static StandardStimulus Standard(string activityTypeName, string? hash = default, IDictionary? 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? 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); } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Services/WorkflowService.cs b/src/runtime/Elsa.Runtime/Services/WorkflowService.cs index 9c1ea931c..2b94b63da 100644 --- a/src/runtime/Elsa.Runtime/Services/WorkflowService.cs +++ b/src/runtime/Elsa.Runtime/Services/WorkflowService.cs @@ -24,27 +24,27 @@ public class WorkflowService : IWorkflowService _stimulusInterpreter = stimulusInterpreter; } - public async Task ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default) + public async Task ExecuteWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? 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 ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default) + public async Task ExecuteWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? 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 DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? input = default, CancellationToken cancellationToken = default) + public async Task DispatchWorkflowAsync(string definitionId, VersionOptions versionOptions, IDictionary? 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 DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? input = default, CancellationToken cancellationToken = default) + public async Task DispatchWorkflowAsync(string instanceId, Bookmark bookmark, IDictionary? 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); } diff --git a/src/runtime/Elsa.Runtime/Stimuli/Handlers/ResumeWorkflowsStimulusHandler.cs b/src/runtime/Elsa.Runtime/Stimuli/Handlers/ResumeWorkflowsStimulusHandler.cs index c7047e946..eb7f847b4 100644 --- a/src/runtime/Elsa.Runtime/Stimuli/Handlers/ResumeWorkflowsStimulusHandler.cs +++ b/src/runtime/Elsa.Runtime/Stimuli/Handlers/ResumeWorkflowsStimulusHandler.cs @@ -14,6 +14,6 @@ public class ResumeWorkflowsStimulusHandler : StimulusHandler protected override async ValueTask> 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)); } } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Stimuli/Handlers/TriggerWorkflowsStimulusHandler.cs b/src/runtime/Elsa.Runtime/Stimuli/Handlers/TriggerWorkflowsStimulusHandler.cs index 7d96675ce..c4ecd0101 100644 --- a/src/runtime/Elsa.Runtime/Stimuli/Handlers/TriggerWorkflowsStimulusHandler.cs +++ b/src/runtime/Elsa.Runtime/Stimuli/Handlers/TriggerWorkflowsStimulusHandler.cs @@ -14,6 +14,6 @@ public class TriggerWorkflowsStimulusHandler : StimulusHandler protected override async ValueTask> 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)); } } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Stimuli/StandardStimulus.cs b/src/runtime/Elsa.Runtime/Stimuli/StandardStimulus.cs index fd2a53377..2c989af71 100644 --- a/src/runtime/Elsa.Runtime/Stimuli/StandardStimulus.cs +++ b/src/runtime/Elsa.Runtime/Stimuli/StandardStimulus.cs @@ -5,4 +5,4 @@ namespace Elsa.Runtime.Stimuli; /// /// Represents a simple stimulus that targets a specific activity type and optionally a corresponding hash to use as a bookmark lookup. /// -public record StandardStimulus(string ActivityTypeName, string? Hash = default, IDictionary? Input = default) : IStimulus; \ No newline at end of file +public record StandardStimulus(string ActivityTypeName, string? Hash = default, IDictionary? Input = default, string? CorrelationId = default) : IStimulus; \ No newline at end of file diff --git a/src/samples/console/Elsa.Samples.Console1/Program.cs b/src/samples/console/Elsa.Samples.Console1/Program.cs index e52e7027e..adbc63151 100644 --- a/src/samples/console/Elsa.Samples.Console1/Program.cs +++ b/src/samples/console/Elsa.Samples.Console1/Program.cs @@ -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(BlockingParallelForEachWorkflow.Create); var workflow14 = new Func(FlowchartWorkflow.Create); - var workflowFactory = workflow3; + var workflowFactory = workflow1; var workflowGraph = workflowFactory(); var workflow = Workflow.FromActivity(workflowGraph); diff --git a/src/samples/console/Elsa.Samples.Console1/Workflows/HelloWorldWorkflow.cs b/src/samples/console/Elsa.Samples.Console1/Workflows/HelloWorldWorkflow.cs index 5dd01841a..750b05b36 100644 --- a/src/samples/console/Elsa.Samples.Console1/Workflows/HelloWorldWorkflow.cs +++ b/src/samples/console/Elsa.Samples.Console1/Workflows/HelloWorldWorkflow.cs @@ -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" }; } \ No newline at end of file