diff --git a/src/bundles/Elsa.Server.Web/MyEndpoint.cs b/src/bundles/Elsa.Server.Web/MyEndpoint.cs new file mode 100644 index 000000000..12ff777c2 --- /dev/null +++ b/src/bundles/Elsa.Server.Web/MyEndpoint.cs @@ -0,0 +1,27 @@ +using Elsa.Abstractions; +using Elsa.Workflows.Runtime.Contracts; + +namespace Elsa.Server.Web; + +public class MyEndpoint : ElsaEndpointWithoutRequest +{ + private readonly IEventPublisher _eventPublisher; + + public MyEndpoint(IEventPublisher eventPublisher) + { + _eventPublisher = eventPublisher; + } + + public override void Configure() + { + Get("/my-event-workflow"); + AllowAnonymous(); + } + + public override async Task HandleAsync(CancellationToken ct) + { + Console.WriteLine("Publishing MyEvent"); + var results = await _eventPublisher.PublishAsync("MyEvent", cancellationToken: ct); + Console.WriteLine($"Affected workflows: {results.Count}"); + } +} \ No newline at end of file diff --git a/src/bundles/Elsa.Server.Web/MyEventWorkflow.cs b/src/bundles/Elsa.Server.Web/MyEventWorkflow.cs new file mode 100644 index 000000000..2439e537c --- /dev/null +++ b/src/bundles/Elsa.Server.Web/MyEventWorkflow.cs @@ -0,0 +1,39 @@ +using Elsa.Workflows; +using Elsa.Workflows.Activities; +using Elsa.Workflows.Contracts; +using Elsa.Workflows.Runtime.Activities; + +namespace Elsa.Server.Web; + +public class OnMyEventWorkflow : WorkflowBase +{ + protected override void Build(IWorkflowBuilder builder) + { + builder.Version = 1; + builder.Id = "OnMyEventWorkflow"; + builder.Root = new Sequence + { + Activities = + { + new Event("MyEvent") + { + CanStartWorkflow = true + }, + new Inline(async () => + { + // IEventPublisher.PublishAsync returns before this executes + await SomeCallAsync(); + }), + new WriteLine("End of workflow"), + new Finish() + } + }; + } + + private async Task SomeCallAsync() + { + Console.WriteLine("Hello from OnMyEventWorkflow"); + await Task.Delay(1000); + Console.WriteLine("Goodbye from OnMyEventWorkflow"); + } +} \ No newline at end of file diff --git a/src/bundles/Elsa.Server.Web/Program.cs b/src/bundles/Elsa.Server.Web/Program.cs index 229c9380f..7df6fea32 100644 --- a/src/bundles/Elsa.Server.Web/Program.cs +++ b/src/bundles/Elsa.Server.Web/Program.cs @@ -24,7 +24,7 @@ using Proto.Persistence.SqlServer; const bool useMongoDb = false; const bool useSqlServer = false; const bool useDapper = false; -const bool useProtoActor = true; +const bool useProtoActor = false; const bool useHangfire = false; const bool useQuartz = true; const bool useMassTransit = true; diff --git a/src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs b/src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs index 416962a0b..b80136e94 100644 --- a/src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs +++ b/src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs @@ -52,7 +52,6 @@ public static class DependencyInjectionExtensions { var builderOptions = new ElsaClientBuilderOptions(); configureClient.Invoke(builderOptions); - builderOptions.ConfigureHttpClientBuilder += builder => builder.AddHttpMessageHandler(sp => (DelegatingHandler)sp.GetRequiredService(builderOptions.AuthenticationHandler)); services.AddScoped(builderOptions.AuthenticationHandler); @@ -63,7 +62,19 @@ public static class DependencyInjectionExtensions options.ConfigureHttpClient = builderOptions.ConfigureHttpClient; options.ApiKey = builderOptions.ApiKey; }); + + var builderOptionsWithoutRetryPolicy = new ElsaClientBuilderOptions + { + ApiKey = builderOptions.ApiKey, + AuthenticationHandler = builderOptions.AuthenticationHandler, + BaseAddress = builderOptions.BaseAddress, + ConfigureHttpClient = builderOptions.ConfigureHttpClient, + ConfigureHttpClientBuilder = builderOptions.ConfigureHttpClientBuilder, + ConfigureRetryPolicy = null + }; + services.AddApi(builderOptions); + services.AddApi(builderOptionsWithoutRetryPolicy); services.AddApi(builderOptions); services.AddApi(builderOptions); services.AddApi(builderOptions); @@ -89,8 +100,20 @@ public static class DependencyInjectionExtensions public static void AddApi(this IServiceCollection services, ElsaClientBuilderOptions? httpClientBuilderOptions = default) where T : class { var builder = services.AddRefitClient(CreateRefitSettings, typeof(T).Name).ConfigureHttpClient(ConfigureElsaApiHttpClient); - httpClientBuilderOptions?.ConfigureHttpClientBuilder?.Invoke(builder); - builder.AddTransientHttpErrorPolicy(p => p.WaitAndRetryAsync(3, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)))); + httpClientBuilderOptions?.ConfigureHttpClientBuilder(builder); + httpClientBuilderOptions?.ConfigureRetryPolicy?.Invoke(builder); + } + + /// + /// Adds a refit client for the specified API type. + /// + /// The service collection. + /// An options object that can be used to configure the HTTP client builder. + /// The type representing the API. + public static void AddApiWithoutRetryPolicy(this IServiceCollection services, ElsaClientBuilderOptions? httpClientBuilderOptions = default) where T : class + { + var builder = services.AddRefitClient(CreateRefitSettings, typeof(T).Name).ConfigureHttpClient(ConfigureElsaApiHttpClient); + httpClientBuilderOptions?.ConfigureHttpClientBuilder(builder); } /// diff --git a/src/clients/Elsa.Api.Client/Options/ElsaClientBuilderOptions.cs b/src/clients/Elsa.Api.Client/Options/ElsaClientBuilderOptions.cs index 998b6d499..a7312eb60 100644 --- a/src/clients/Elsa.Api.Client/Options/ElsaClientBuilderOptions.cs +++ b/src/clients/Elsa.Api.Client/Options/ElsaClientBuilderOptions.cs @@ -1,5 +1,6 @@ using Elsa.Api.Client.HttpMessageHandlers; using Microsoft.Extensions.DependencyInjection; +using Polly; namespace Elsa.Api.Client.Options; @@ -33,4 +34,9 @@ public class ElsaClientBuilderOptions /// Gets or sets a delegate that can be used to configure the HTTP client builder. /// public Action ConfigureHttpClientBuilder { get; set; } = _ => { }; + + /// + /// Gets or sets a delegate that can be used to configure the retry policy. + /// + public Action? ConfigureRetryPolicy { get; set; } = builder => builder.AddTransientHttpErrorPolicy(p => p.WaitAndRetryAsync(3, attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)))); } \ No newline at end of file diff --git a/src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IExecuteWorkflowApi.cs b/src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IExecuteWorkflowApi.cs new file mode 100644 index 000000000..9b2ec493c --- /dev/null +++ b/src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IExecuteWorkflowApi.cs @@ -0,0 +1,32 @@ +using Elsa.Api.Client.Resources.WorkflowDefinitions.Requests; +using JetBrains.Annotations; +using Refit; + +namespace Elsa.Api.Client.Resources.WorkflowDefinitions.Contracts; + +/// +/// Represents a client for the workflow definitions API. +/// +[PublicAPI] +public interface IExecuteWorkflowApi +{ + /// + /// Executes a workflow definition. + /// + /// The definition ID of the workflow definition to execute. + /// An optional request containing options for executing the workflow definition. + /// An optional cancellation token. + /// A response containing information about the workflow instance that was created. + [Post("/workflow-definitions/{definitionId}/execute")] + Task ExecuteAsync(string definitionId, ExecuteWorkflowDefinitionRequest? request, CancellationToken cancellationToken = default); + + /// + /// Dispatches a request to execute the specified workflow definition. + /// + /// The definition ID of the workflow definition to dispatch request. + /// An optional request containing options for dispatching a request to execute the specified workflow definition. + /// An optional cancellation token. + /// A response containing information about the workflow instance that was created. + [Post("/workflow-definitions/{definitionId}/dispatch")] + Task DispatchAsync(string definitionId, DispatchWorkflowDefinitionRequest? request, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IWorkflowDefinitionsApi.cs b/src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IWorkflowDefinitionsApi.cs index 73205909b..b165cec8f 100644 --- a/src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IWorkflowDefinitionsApi.cs +++ b/src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IWorkflowDefinitionsApi.cs @@ -196,24 +196,4 @@ public interface IWorkflowDefinitionsApi /// An optional cancellation token. [Post("/workflow-definitions/{definitionId}/revert/{version}")] Task RevertVersionAsync(string definitionId, int version, CancellationToken cancellationToken = default); - - /// - /// Executes a workflow definition. - /// - /// The definition ID of the workflow definition to execute. - /// An optional request containing options for executing the workflow definition. - /// An optional cancellation token. - /// A response containing information about the workflow instance that was created. - [Post("/workflow-definitions/{definitionId}/execute")] - Task ExecuteAsync(string definitionId, ExecuteWorkflowDefinitionRequest? request, CancellationToken cancellationToken = default); - - /// - /// Dispatches a request to execute the specified workflow definition. - /// - /// The definition ID of the workflow definition to dispatch request. - /// An optional request containing options for dispatching a request to execute the specified workflow definition. - /// An optional cancellation token. - /// A response containing information about the workflow instance that was created. - [Post("/workflow-definitions/{definitionId}/dispatch")] - Task DispatchAsync(string definitionId, DispatchWorkflowDefinitionRequest? request, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/common/Elsa.Api.Common/Elsa.Api.Common.csproj b/src/common/Elsa.Api.Common/Elsa.Api.Common.csproj index 9c992b1df..eab2e3e1b 100644 --- a/src/common/Elsa.Api.Common/Elsa.Api.Common.csproj +++ b/src/common/Elsa.Api.Common/Elsa.Api.Common.csproj @@ -19,9 +19,9 @@ - - - + + + diff --git a/src/modules/Elsa.Http/Activities/FlowSendHttpRequest.cs b/src/modules/Elsa.Http/Activities/FlowSendHttpRequest.cs index 2960b5887..6b6e8b1d7 100644 --- a/src/modules/Elsa.Http/Activities/FlowSendHttpRequest.cs +++ b/src/modules/Elsa.Http/Activities/FlowSendHttpRequest.cs @@ -43,8 +43,6 @@ public class FlowSendHttpRequest : SendHttpRequestBase, IActivityPropertyDefault outcomes.Add(outcome); outcomes.Add("Done"); - - context.JournalData["StatusCode"] = statusCode; await context.CompleteActivityWithOutcomesAsync(outcomes.ToArray()); } diff --git a/src/modules/Elsa.Http/Activities/SendHttpRequestBase.cs b/src/modules/Elsa.Http/Activities/SendHttpRequestBase.cs index 161892751..53cc92daa 100644 --- a/src/modules/Elsa.Http/Activities/SendHttpRequestBase.cs +++ b/src/modules/Elsa.Http/Activities/SendHttpRequestBase.cs @@ -77,12 +77,24 @@ public abstract class SendHttpRequestBase : Activity )] public Input RequestHeaders { get; set; } = new(new HttpHeaders()); + /// + /// The HTTP response status code + /// + [Output(Description = "The HTTP response status code")] + public Output StatusCode { get; set; } = default!; + /// /// The parsed content, if any. /// [Output(Description = "The parsed content, if any.")] public Output ParsedContent { get; set; } = default!; + /// + /// The response headers that were received. + /// + [Output(Description = "The response headers that were received.")] + public Output ResponseHeaders { get; set; } = default!; + /// protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { @@ -115,8 +127,13 @@ public abstract class SendHttpRequestBase : Activity { var response = await httpClient.SendAsync(request, cancellationToken); var parsedContent = await ParseContentAsync(context, response.Content); + var statusCode = (int)response.StatusCode; + var responseHeaders = new HttpHeaders(response.Headers); + context.Set(Result, response); context.Set(ParsedContent, parsedContent); + context.Set(StatusCode, statusCode); + context.Set(ResponseHeaders, responseHeaders); await HandleResponseAsync(context, response); } diff --git a/src/modules/Elsa.Http/Models/HttpHeaders.cs b/src/modules/Elsa.Http/Models/HttpHeaders.cs index 1cbe03b8c..e1f61b26d 100644 --- a/src/modules/Elsa.Http/Models/HttpHeaders.cs +++ b/src/modules/Elsa.Http/Models/HttpHeaders.cs @@ -1,3 +1,4 @@ +using System.Net.Http.Headers; using System.Text.Json.Serialization; using Elsa.Extensions; using Elsa.Http.Serialization; @@ -10,6 +11,25 @@ namespace Elsa.Http.Models; [JsonConverter(typeof(HttpHeadersConverter))] public class HttpHeaders : Dictionary { + /// + public HttpHeaders() + { + } + + /// + public HttpHeaders(IDictionary source) + { + foreach (var item in source) + Add(item.Key, item.Value); + } + + /// + public HttpHeaders(HttpResponseHeaders source) + { + foreach (var item in source) + Add(item.Key, item.Value.ToArray()); + } + /// /// Gets the content type. /// diff --git a/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs b/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs index 6299094b0..c0ccee005 100644 --- a/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs +++ b/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs @@ -46,6 +46,7 @@ public class MassTransitWorkflowDispatcher : IWorkflowDispatcher ActivityInstanceId = request.ActivityInstanceId, ActivityHash = request.ActivityHash, Input = request.Input, + Properties = request.Properties, CorrelationId = request.CorrelationId }, cancellationToken); return new(); diff --git a/src/modules/Elsa.WorkflowContexts/Middleware/WorkflowContextActivityExecutionMiddleware.cs b/src/modules/Elsa.WorkflowContexts/Middleware/WorkflowContextActivityExecutionMiddleware.cs index ecc43655c..ccb7d4ba2 100644 --- a/src/modules/Elsa.WorkflowContexts/Middleware/WorkflowContextActivityExecutionMiddleware.cs +++ b/src/modules/Elsa.WorkflowContexts/Middleware/WorkflowContextActivityExecutionMiddleware.cs @@ -36,7 +36,7 @@ public class WorkflowContextActivityExecutionMiddleware : IActivityExecutionMidd } // Check if this is a background execution. - var isBackgroundExecution = context.TransientProperties.GetValueOrDefault(BackgroundActivityCollectorMiddleware.IsBackgroundExecution); + var isBackgroundExecution = context.GetIsBackgroundExecution(); // Is the activity configured to load the context? foreach (var providerType in providerTypes) diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs index 2fd7df1b5..b53b7958a 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs @@ -230,6 +230,29 @@ public class ActivityExecutionContext : IExecutionContext /// The options used to schedule the activity. public async ValueTask ScheduleActivityAsync(ActivityNode? activityNode, ActivityExecutionContext? owner = default, ScheduleWorkOptions? options = default) { + if (this.GetIsBackgroundExecution()) + { + var scheduledActivity = new ScheduledActivity + { + ActivityNodeId = activityNode?.NodeId, + OwnerActivityInstanceId = owner?.Id, + Options = options != null ? new ScheduledActivityOptions + { + CompletionCallback = options?.CompletionCallback?.Method.Name, + Tag = options?.Tag, + ExistingActivityInstanceId = options?.ExistingActivityExecutionContext?.Id, + PreventDuplicateScheduling = options?.PreventDuplicateScheduling ?? false, + Variables = options?.Variables?.ToList(), + Input = options?.Input + } : default + }; + + var scheduledActivities = this.GetBackgroundScheduledActivities().ToList(); + scheduledActivities.Add(scheduledActivity); + this.SetBackgroundScheduledActivities(scheduledActivities); + return; + } + var completionCallback = options?.CompletionCallback; owner ??= this; diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs index 773597845..937934592 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs @@ -456,6 +456,16 @@ public static class ActivityExecutionContextExtensions /// public static async ValueTask CompleteActivityAsync(this ActivityExecutionContext context, object? result = default) { + var outcomes = result as Outcomes; + + // If the activity is executing in the background, simply capture the result and return. + if (context.GetIsBackgroundExecution()) + { + if (outcomes != null) + context.SetBackgroundOutcomes(outcomes.Names); + return; + } + // If the activity is not running, do nothing. if (context.Status != ActivityStatus.Running) return; @@ -470,7 +480,7 @@ public static class ActivityExecutionContextExtensions context.Status = ActivityStatus.Completed; // Record the outcomes, if any. - if (result is Outcomes outcomes) + if (outcomes != null) context.JournalData["Outcomes"] = outcomes.Names; // Record the output, if any. diff --git a/src/modules/Elsa.Workflows.Core/Extensions/BackgroundActivityExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/BackgroundActivityExecutionContextExtensions.cs new file mode 100644 index 000000000..b427d6751 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Extensions/BackgroundActivityExecutionContextExtensions.cs @@ -0,0 +1,65 @@ +using Elsa.Workflows.Models; + +namespace Elsa.Workflows; + +/// +/// Adds extension methods to . +/// +public static class BackgroundActivityExecutionContextExtensions +{ + /// + /// A key into the activity execution context's transient properties that indicates whether the current activity is being executed in the background. + /// + public static readonly object IsBackgroundExecution = new(); + + /// + /// Configures the activity execution context to execute the current activity in the background. + /// + public static void SetIsBackgroundExecution(this ActivityExecutionContext activityExecutionContext, bool value = true) + { + activityExecutionContext.TransientProperties[IsBackgroundExecution] = value; + } + + /// + /// Gets a value indicating whether the current activity is being executed in the background. + /// + public static bool GetIsBackgroundExecution(this ActivityExecutionContext activityExecutionContext) + { + return activityExecutionContext.TransientProperties.ContainsKey(IsBackgroundExecution); + } + + /// + /// Sets the background outcomes. + /// + public static void SetBackgroundOutcomes(this ActivityExecutionContext activityExecutionContext, IEnumerable outcomes) + { + var outcomesList = outcomes.ToList(); + activityExecutionContext.SetProperty("BackgroundOutcomes", outcomesList); + } + + /// + /// Gets the background outcomes. + /// + public static IEnumerable GetBackgroundOutcomes(this ActivityExecutionContext activityExecutionContext) + { + return activityExecutionContext.GetProperty>("BackgroundOutcomes") ?? Enumerable.Empty(); + } + + /// + /// Sets the background scheduled activities. + /// + public static void SetBackgroundScheduledActivities(this ActivityExecutionContext activityExecutionContext, IEnumerable scheduledActivities) + { + var scheduledActivitiesList = scheduledActivities.ToList(); + activityExecutionContext.SetProperty("BackgroundScheduledActivities", scheduledActivitiesList); + } + + /// + /// Gets the background scheduled activities. + /// + /// + public static IEnumerable GetBackgroundScheduledActivities(this ActivityExecutionContext activityExecutionContext) + { + return activityExecutionContext.GetProperty>("BackgroundScheduledActivities") ?? Enumerable.Empty(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ModuleExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/ModuleExtensions.cs index 70f989e5b..a591ccba7 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ModuleExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ModuleExtensions.cs @@ -16,6 +16,8 @@ public static class ModuleExtensions public static IServiceCollection AddStorageDriver(this IServiceCollection services) where T : class, IStorageDriver { - return services.AddSingleton(); + return services + .AddSingleton() + .AddSingleton(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Models/BackgroundExecutionOutcome.cs b/src/modules/Elsa.Workflows.Core/Models/BackgroundExecutionOutcome.cs new file mode 100644 index 000000000..8d462bb74 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Models/BackgroundExecutionOutcome.cs @@ -0,0 +1,3 @@ +namespace Elsa.Workflows.Models; + +public record BackgroundExecutionOutcome(string Name, object? Payload); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Models/BackgroundExecutionResult.cs b/src/modules/Elsa.Workflows.Core/Models/BackgroundExecutionResult.cs new file mode 100644 index 000000000..133d3e6b6 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Models/BackgroundExecutionResult.cs @@ -0,0 +1,8 @@ +namespace Elsa.Workflows.Models; + +public class BackgroundExecutionResult +{ + public ICollection Outcomes { get; set; } = new List(); + public ICollection ExecutionLog { get; set; } = new List(); + public IDictionary JournalData { get; } = new Dictionary(); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Models/ScheduledActivity.cs b/src/modules/Elsa.Workflows.Core/Models/ScheduledActivity.cs new file mode 100644 index 000000000..768bbb112 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Models/ScheduledActivity.cs @@ -0,0 +1,8 @@ +namespace Elsa.Workflows.Models; + +public class ScheduledActivity +{ + public string? ActivityNodeId { get; set; } + public string? OwnerActivityInstanceId { get; set; } + public ScheduledActivityOptions? Options { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Models/ScheduledActivityOptions.cs b/src/modules/Elsa.Workflows.Core/Models/ScheduledActivityOptions.cs new file mode 100644 index 000000000..403c91e65 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Models/ScheduledActivityOptions.cs @@ -0,0 +1,13 @@ +using Elsa.Workflows.Memory; + +namespace Elsa.Workflows.Models; + +public class ScheduledActivityOptions +{ + public string? CompletionCallback { get; set; } + public object? Tag { get; set; } + public ICollection? Variables { get; set; } + public string? ExistingActivityInstanceId { get; set; } + public bool PreventDuplicateScheduling { get; set; } + public IDictionary? Input { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs b/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs index 5615ac0e9..aa8ef62da 100644 --- a/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs +++ b/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs @@ -77,7 +77,9 @@ public class WorkflowStateExtractor : IWorkflowStateExtractor private void ApplyProperties(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) { - workflowExecutionContext.Properties = state.Properties; + // Merge properties. + foreach (var property in state.Properties) + workflowExecutionContext.Properties[property.Key] = property.Value; } private static void ApplyActivityExecutionContexts(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) @@ -248,7 +250,7 @@ public class WorkflowStateExtractor : IWorkflowStateExtractor // // If there are any faulted contexts, keep everything so that the user can fix the issue and potentially reschedule existing instances. // if (contexts.Any(x => x.Status == ActivityStatus.Faulted)) - return contexts; + return contexts; // return contexts // .Where(x => !x.IsCompleted) diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowInbox.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowInbox.cs index 949c7cf62..52245a1fe 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowInbox.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowInbox.cs @@ -32,13 +32,14 @@ public interface IWorkflowInbox /// The message to deliver. /// An optional cancellation token. ValueTask DeliverAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default); - + /// /// Broadcasts the specified message, which may trigger new workflows and resume existing ones. /// /// The message to broadcast. + /// An optional set of delivery options. /// An optional cancellation token. - ValueTask BroadcastAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default); + ValueTask BroadcastAsync(WorkflowInboxMessage message, BroadcastWorkflowInboxMessageOptions? options, CancellationToken cancellationToken = default); /// /// Finds all messages matching the specified filter. diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs b/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs index 0781caf60..e471f2815 100644 --- a/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs @@ -11,7 +11,7 @@ namespace Elsa.Extensions; public static class ActivityExecutionPipelineBuilderExtensions { /// - /// Installs the . + /// Installs the . /// - public static IActivityExecutionPipelineBuilder UseBackgroundActivityInvoker(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware(); + public static IActivityExecutionPipelineBuilder UseBackgroundActivityInvoker(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware(); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/CancelBackgroundActivities.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/CancelBackgroundActivities.cs index 7801a84e7..2b09aee67 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/CancelBackgroundActivities.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/CancelBackgroundActivities.cs @@ -31,7 +31,7 @@ public class CancelBackgroundActivities : INotificationHandler public async Task HandleAsync(WorkflowBookmarksIndexed notification, CancellationToken cancellationToken) { - var removedBookmarks = notification.IndexedWorkflowBookmarks.RemovedBookmarks.Where(x => x.Name == BackgroundActivityCollectorMiddleware.BackgroundActivityBookmarkName); + var removedBookmarks = notification.IndexedWorkflowBookmarks.RemovedBookmarks.Where(x => x.Name == BackgroundActivityInvokerMiddleware.BackgroundActivityBookmarkName); foreach (var removedBookmark in removedBookmarks) { diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ReadWorkflowInboxMessage.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ReadWorkflowInboxMessage.cs index 59e115967..ecf6cb080 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ReadWorkflowInboxMessage.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ReadWorkflowInboxMessage.cs @@ -1,5 +1,7 @@ +using Elsa.Extensions; using Elsa.Mediator.Contracts; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Notifications; namespace Elsa.Workflows.Runtime.Handlers; @@ -19,11 +21,16 @@ public class ReadWorkflowInboxMessage : INotificationHandler public async Task HandleAsync(WorkflowInboxMessageReceived notification, CancellationToken cancellationToken) { var message = notification.InboxMessage; - await _workflowInbox.BroadcastAsync(message, cancellationToken); + var options = new BroadcastWorkflowInboxMessageOptions + { + DispatchAsynchronously = notification.Options.DispatchAsynchronously + }; + var result = await _workflowInbox.BroadcastAsync(message, options, cancellationToken); + notification.WorkflowExecutionResults.AddRange(result.WorkflowExecutionResults); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs index da4f05bf5..defca4956 100644 --- a/src/modules/Elsa.Workflows.Runtime/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs +++ b/src/modules/Elsa.Workflows.Runtime/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs @@ -1,6 +1,8 @@ +using System.Text.Json; using Elsa.Extensions; using Elsa.Workflows.Middleware.Activities; using Elsa.Workflows.Models; +using Elsa.Workflows.Options; using Elsa.Workflows.Pipelines.ActivityExecution; using Elsa.Workflows.Runtime.Bookmarks; using Elsa.Workflows.Runtime.Middleware.Workflows; @@ -12,19 +14,17 @@ namespace Elsa.Workflows.Runtime.Middleware.Activities; /// Collects the current activity for scheduling for execution from a background job if the activity is of kind or . /// The actual scheduling of the activity happens in . /// -public class BackgroundActivityCollectorMiddleware : DefaultActivityInvokerMiddleware +public class BackgroundActivityInvokerMiddleware : DefaultActivityInvokerMiddleware { - /// - /// A key into the activity execution context's transient properties that indicates whether the current activity is being executed in the background. - /// - public static readonly object IsBackgroundExecution = new(); - - internal static string GetBackgroundActivityOutputKey(string activityId) => $"__BackgroundActivityOutput:{activityId}"; + internal static string GetBackgroundActivityOutputKey(string activityNodeId) => $"__BackgroundActivityOutput:{activityNodeId}"; + internal static string GetBackgroundActivityOutcomesKey(string activityNodeId) => $"__BackgroundActivityOutcomes:{activityNodeId}"; + internal static string GetBackgroundActivityJournalDataKey(string activityNodeId) => $"__BackgroundActivityJournalData:{activityNodeId}"; + internal static string GetBackgroundActivityScheduledActivitiesKey(string activityNodeId) => $"__BackgroundActivityScheduledActivities:{activityNodeId}"; internal static readonly object BackgroundActivitySchedulesKey = new(); internal const string BackgroundActivityBookmarkName = "BackgroundActivity"; /// - public BackgroundActivityCollectorMiddleware(ActivityMiddlewareDelegate next) : base(next) + public BackgroundActivityInvokerMiddleware(ActivityMiddlewareDelegate next) : base(next) { } @@ -37,8 +37,17 @@ public class BackgroundActivityCollectorMiddleware : DefaultActivityInvokerMiddl ScheduleBackgroundActivity(context); else { - CaptureOutputIfAny(context); await base.ExecuteActivityAsync(context); + + // This part is either executed from the background, or in the foreground when the activity is resumed. + var isResuming = !GetIsBackgroundExecution(context) && context.ActivityDescriptor.Kind is ActivityKind.Task or ActivityKind.Job; + if (isResuming) + { + CaptureOutputIfAny(context); + CaptureJournalData(context); + await CompleteBackgroundActivityOutcomesAsync(context); + await CompleteBackgroundActivityScheduledActivitiesAsync(context); + } } } @@ -51,11 +60,13 @@ public class BackgroundActivityCollectorMiddleware : DefaultActivityInvokerMiddl var activityDescriptor = context.ActivityDescriptor; var kind = activityDescriptor.Kind; - return !context.TransientProperties.ContainsKey(IsBackgroundExecution) + return !GetIsBackgroundExecution(context) && context.WorkflowExecutionContext.ExecuteDelegate == null && (kind is ActivityKind.Job || (kind == ActivityKind.Task && activity.GetRunAsynchronously())); } + private static bool GetIsBackgroundExecution(ActivityExecutionContext context) => context.TransientProperties.ContainsKey(BackgroundActivityExecutionContextExtensions.IsBackgroundExecution); + /// /// Schedules the current activity for execution in the background. /// @@ -77,21 +88,79 @@ public class BackgroundActivityCollectorMiddleware : DefaultActivityInvokerMiddl private static void CaptureOutputIfAny(ActivityExecutionContext context) { var activity = context.Activity; - var inputKey = GetBackgroundActivityOutputKey(activity.Id); + var inputKey = GetBackgroundActivityOutputKey(activity.NodeId); + var capturedOutput = context.WorkflowExecutionContext.GetProperty>(inputKey); - if (!context.WorkflowInput.TryGetValue(inputKey, out var capturedOutput)) + if (capturedOutput == null) return; - var input = (IDictionary)capturedOutput; - foreach (var inputEntry in input) + foreach (var outputEntry in capturedOutput) { - var outputDescriptor = context.ActivityDescriptor.Outputs.FirstOrDefault(x => x.Name == inputEntry.Key); + var outputDescriptor = context.ActivityDescriptor.Outputs.FirstOrDefault(x => x.Name == outputEntry.Key); if (outputDescriptor == null) continue; var output = (Output?)outputDescriptor.ValueGetter(activity); - context.Set(output, inputEntry.Value); + context.Set(output, outputEntry.Value); + } + } + + private void CaptureJournalData(ActivityExecutionContext context) + { + var activity = context.Activity; + var journalDataKey = GetBackgroundActivityJournalDataKey(activity.NodeId); + var journalData = context.WorkflowExecutionContext.GetProperty>(journalDataKey); + + if (journalData == null) + return; + + foreach (var journalEntry in journalData) + context.JournalData[journalEntry.Key] = journalEntry.Value; + } + + private async Task CompleteBackgroundActivityOutcomesAsync(ActivityExecutionContext context) + { + var outcomesKey = GetBackgroundActivityOutcomesKey(context.NodeId); + var outcomes = context.WorkflowExecutionContext.GetProperty>(outcomesKey); + + if (outcomes != null) + { + await context.CompleteActivityWithOutcomesAsync(outcomes.ToArray()); + + // Remove the outcomes from the workflow execution context. + context.WorkflowExecutionContext.Properties.Remove(outcomesKey); + } + } + + private async Task CompleteBackgroundActivityScheduledActivitiesAsync(ActivityExecutionContext context) + { + var scheduledActivitiesKey = GetBackgroundActivityScheduledActivitiesKey(context.NodeId); + var scheduledActivitiesJson = context.WorkflowExecutionContext.GetProperty(scheduledActivitiesKey); + var scheduledActivities = scheduledActivitiesJson != null ? JsonSerializer.Deserialize>(scheduledActivitiesJson) : null; + + if (scheduledActivities != null) + { + foreach (var scheduledActivity in scheduledActivities) + { + var activityNode = scheduledActivity.ActivityNodeId != null ? context.WorkflowExecutionContext.FindActivityByNodeId(scheduledActivity.ActivityNodeId) : null; + var owner = scheduledActivity.OwnerActivityInstanceId != null ? context.WorkflowExecutionContext.ActivityExecutionContexts.FirstOrDefault(x => x.Id == scheduledActivity.OwnerActivityInstanceId) : null; + var options = scheduledActivity.Options != null + ? new ScheduleWorkOptions + { + ExistingActivityExecutionContext = scheduledActivity.Options.ExistingActivityInstanceId != null ? context.WorkflowExecutionContext.ActivityExecutionContexts.FirstOrDefault(x => x.Id == scheduledActivity.Options.ExistingActivityInstanceId) : null, + Variables = scheduledActivity.Options?.Variables, + CompletionCallback = !string.IsNullOrEmpty(scheduledActivity.Options?.CompletionCallback) && owner != null ? owner.Activity.GetActivityCompletionCallback(scheduledActivity.Options.CompletionCallback) : default, + PreventDuplicateScheduling = scheduledActivity.Options?.PreventDuplicateScheduling ?? false, + Input = scheduledActivity.Options?.Input, + Tag = scheduledActivity.Options?.Tag + } + : default; + await context.ScheduleActivityAsync(activityNode, owner, options); + } + + // Remove the scheduled activities from the workflow execution context. + context.WorkflowExecutionContext.Properties.Remove(scheduledActivitiesKey); } } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/ScheduleBackgroundActivitiesMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/ScheduleBackgroundActivitiesMiddleware.cs index f5b83f1a8..c9d32723f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/ScheduleBackgroundActivitiesMiddleware.cs +++ b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/ScheduleBackgroundActivitiesMiddleware.cs @@ -43,7 +43,7 @@ public class ScheduleBackgroundActivitiesMiddleware : WorkflowExecutionMiddlewar var scheduledBackgroundActivities = workflowExecutionContext .TransientProperties - .GetOrAdd(BackgroundActivityCollectorMiddleware.BackgroundActivitySchedulesKey, () => new List()); + .GetOrAdd(BackgroundActivityInvokerMiddleware.BackgroundActivitySchedulesKey, () => new List()); if (scheduledBackgroundActivities.Any()) { diff --git a/src/modules/Elsa.Workflows.Runtime/Models/BroadcastWorkflowInboxMessageOptions.cs b/src/modules/Elsa.Workflows.Runtime/Models/BroadcastWorkflowInboxMessageOptions.cs new file mode 100644 index 000000000..e43cb2e98 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Models/BroadcastWorkflowInboxMessageOptions.cs @@ -0,0 +1,15 @@ +namespace Elsa.Workflows.Runtime.Models; + +/// +/// Represents the options for broadcasting a workflow inbox message. +/// +public class BroadcastWorkflowInboxMessageOptions +{ + /// + /// Gets or sets a value indicating whether the dispatch should be executed asynchronously. + /// + /// + /// true if the dispatch should be executed asynchronously; otherwise, false. + /// + public bool DispatchAsynchronously { get; set; } = true; +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowInboxMessageReceived.cs b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowInboxMessageReceived.cs index 075a960e2..823fc7cd4 100644 --- a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowInboxMessageReceived.cs +++ b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowInboxMessageReceived.cs @@ -1,5 +1,7 @@ using Elsa.Mediator.Contracts; using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.Runtime.Results; namespace Elsa.Workflows.Runtime.Notifications; @@ -7,4 +9,7 @@ namespace Elsa.Workflows.Runtime.Notifications; /// A notification that is sent when a workflow inbox message is received. /// /// The inbox message that was received. -public record WorkflowInboxMessageReceived(WorkflowInboxMessage InboxMessage) : INotification; \ No newline at end of file +public record WorkflowInboxMessageReceived( + WorkflowInboxMessage InboxMessage, + WorkflowInboxMessageDeliveryOptions Options, + ICollection WorkflowExecutionResults) : INotification; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Options/WorkflowInboxMessageDeliveryOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowInboxMessageDeliveryOptions.cs index 815de54dc..4b874e210 100644 --- a/src/modules/Elsa.Workflows.Runtime/Options/WorkflowInboxMessageDeliveryOptions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowInboxMessageDeliveryOptions.cs @@ -1,7 +1,3 @@ -using Elsa.Mediator; -using Elsa.Mediator.Contracts; -using Elsa.Workflows.Runtime.Notifications; - namespace Elsa.Workflows.Runtime.Options; /// @@ -10,7 +6,7 @@ namespace Elsa.Workflows.Runtime.Options; public class WorkflowInboxMessageDeliveryOptions { /// - /// The strategy to use when publishing the notification. + /// Whether to dispatch the message to the workflow dispatcher or send immediately. /// - public IEventPublishingStrategy EventPublishingStrategy { get; set; } = NotificationStrategy.Background; + public bool DispatchAsynchronously { get; set; } = true; } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Results/DeliverWorkflowInboxMessageResult.cs b/src/modules/Elsa.Workflows.Runtime/Results/DeliverWorkflowInboxMessageResult.cs index a05749e6a..5a786261f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Results/DeliverWorkflowInboxMessageResult.cs +++ b/src/modules/Elsa.Workflows.Runtime/Results/DeliverWorkflowInboxMessageResult.cs @@ -3,4 +3,4 @@ namespace Elsa.Workflows.Runtime.Results; /// /// Result of delivering a workflow inbox message. /// -public record DeliverWorkflowInboxMessageResult; \ No newline at end of file +public record DeliverWorkflowInboxMessageResult(ICollection WorkflowExecutionResults); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs index bc40b4473..372271f22 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs @@ -1,3 +1,4 @@ +using System.Text.Json; using Elsa.Common.Models; using Elsa.Workflows.Contracts; using Elsa.Workflows.Helpers; @@ -57,6 +58,7 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker public async Task ExecuteAsync(ScheduledBackgroundActivity scheduledBackgroundActivity, CancellationToken cancellationToken = default) { var workflowInstanceId = scheduledBackgroundActivity.WorkflowInstanceId; + var workflowState = await _workflowRuntime.ExportWorkflowStateAsync(workflowInstanceId, cancellationToken); if (workflowState == null) @@ -69,7 +71,6 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker var workflow = await _workflowDefinitionService.MaterializeWorkflowAsync(workflowDefinition, cancellationToken); var workflowExecutionContext = await WorkflowExecutionContext.CreateAsync(_serviceProvider, workflow, workflowState, cancellationTokens: cancellationToken); - var originalBookmarks = workflowExecutionContext.Bookmarks.ToList(); var activityNodeId = scheduledBackgroundActivity.ActivityNodeId; var activityExecutionContext = workflowExecutionContext.ActivityExecutionContexts.First(x => x.NodeId == activityNodeId); @@ -77,7 +78,7 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker await _variablePersistenceManager.LoadVariablesAsync(workflowExecutionContext); // Mark the activity as being invoked from a background worker. - activityExecutionContext.TransientProperties[BackgroundActivityCollectorMiddleware.IsBackgroundExecution] = true; + activityExecutionContext.SetIsBackgroundExecution(); // Invoke the activity. await _activityInvoker.InvokeAsync(activityExecutionContext); @@ -111,32 +112,25 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker outputValues[outputDescriptor.Name] = outputValue; } - // TODO: Instead of importing the entire workflow state, we should only import the following: - // - Variables - // - Activity state - // - Activity output - // - Bookmarks - workflowState = _workflowStateExtractor.Extract(workflowExecutionContext); - await _variablePersistenceManager.SaveVariablesAsync(workflowExecutionContext); - await _workflowRuntime.ImportWorkflowStateAsync(workflowState, cancellationToken); - - // Process bookmarks. - var newBookmarks = workflowExecutionContext.Bookmarks.ToList(); - var diff = Diff.For(originalBookmarks, newBookmarks); - await _bookmarksPersister.PersistBookmarksAsync(workflowExecutionContext, diff); - - // Resume the workflow, passing along the activity output. - // TODO: This approach will fail if the output is non-serializable. We need to find a way to pass the output to the workflow without serializing it. + // Resume the workflow, passing along activity output, outcomes and scheduled activities. var bookmarkId = scheduledBackgroundActivity.BookmarkId; - var inputKey = BackgroundActivityCollectorMiddleware.GetBackgroundActivityOutputKey(activityNodeId); + var inputKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityOutputKey(activityNodeId); + var outcomesKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityOutcomesKey(activityNodeId); + var journalDataKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityJournalDataKey(activityNodeId); + var scheduledActivitiesKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityScheduledActivitiesKey(activityNodeId); + var outcomes = activityExecutionContext.GetBackgroundOutcomes().ToList(); + var scheduledActivities = activityExecutionContext.GetBackgroundScheduledActivities().ToList(); var dispatchRequest = new DispatchWorkflowInstanceRequest { InstanceId = workflowInstanceId, BookmarkId = bookmarkId, - Input = new Dictionary + Properties = new Dictionary { - [inputKey] = outputValues + [outcomesKey] = outcomes, + [scheduledActivitiesKey] = JsonSerializer.Serialize(scheduledActivities), + [inputKey] = outputValues, + [journalDataKey] = activityExecutionContext.JournalData } }; diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs index 7f2fb4dda..1b9259f69 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs @@ -1,4 +1,5 @@ using Elsa.Common.Contracts; +using Elsa.Mediator; using Elsa.Mediator.Contracts; using Elsa.Workflows.Contracts; using Elsa.Workflows.Runtime.Contracts; @@ -16,6 +17,7 @@ namespace Elsa.Workflows.Runtime.Services; public class DefaultWorkflowInbox : IWorkflowInbox { private readonly IWorkflowDispatcher _workflowDispatcher; + private readonly IWorkflowRuntime _workflowRuntime; private readonly IWorkflowInboxMessageStore _messageStore; private readonly INotificationSender _notificationSender; private readonly ISystemClock _systemClock; @@ -27,6 +29,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox /// public DefaultWorkflowInbox( IWorkflowDispatcher workflowDispatcher, + IWorkflowRuntime workflowRuntime, IWorkflowInboxMessageStore messageStore, INotificationSender notificationSender, ISystemClock systemClock, @@ -34,6 +37,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox IBookmarkHasher bookmarkHasher) { _workflowDispatcher = workflowDispatcher; + _workflowRuntime = workflowRuntime; _messageStore = messageStore; _notificationSender = notificationSender; _systemClock = systemClock; @@ -72,10 +76,9 @@ public class DefaultWorkflowInbox : IWorkflowInbox await _messageStore.SaveAsync(message, cancellationToken); // Send a notification. - var strategy = options.EventPublishingStrategy; var workflowExecutionResults = new List(); - var notification = new WorkflowInboxMessageReceived(message); - await _notificationSender.SendAsync(notification, strategy, cancellationToken); + var notification = new WorkflowInboxMessageReceived(message, options, workflowExecutionResults); + await _notificationSender.SendAsync(notification, NotificationStrategy.Sequential, cancellationToken); // Return the result. return new SubmitWorkflowInboxMessageResult(message, workflowExecutionResults); @@ -84,19 +87,12 @@ public class DefaultWorkflowInbox : IWorkflowInbox /// public async ValueTask DeliverAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) { - await ResumeWorkflowsAsync(message, cancellationToken); - return new DeliverWorkflowInboxMessageResult(); + await ResumeWorkflowsAsynchronouslyAsync(message, cancellationToken); + return new DeliverWorkflowInboxMessageResult(new List()); } /// - public async ValueTask BroadcastAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) - { - await TriggerWorkflowsAsync(message, cancellationToken); - - return new DeliverWorkflowInboxMessageResult(); - } - - private async Task TriggerWorkflowsAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) + public async ValueTask BroadcastAsync(WorkflowInboxMessage message, BroadcastWorkflowInboxMessageOptions? options, CancellationToken cancellationToken = default) { var activityTypeName = message.ActivityTypeName; var correlationId = message.CorrelationId; @@ -107,8 +103,28 @@ public class DefaultWorkflowInbox : IWorkflowInbox if (workflowInstanceId != null) { - await ResumeWorkflowsAsync(message, cancellationToken); - return; + if (options?.DispatchAsynchronously == true) + { + await ResumeWorkflowsAsynchronouslyAsync(message, cancellationToken); + return new DeliverWorkflowInboxMessageResult(new List()); + } + + var results = await ResumeWorkflowsSynchronouslyAsync(message, cancellationToken); + return new DeliverWorkflowInboxMessageResult(results.ToList()); + } + + if (options?.DispatchAsynchronously == false) + { + var results = await _workflowRuntime.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, new TriggerWorkflowsOptions + { + CorrelationId = correlationId, + WorkflowInstanceId = workflowInstanceId, + ActivityInstanceId = activityInstanceId, + Input = input, + CancellationTokens = cancellationToken + }); + + return new DeliverWorkflowInboxMessageResult(results.TriggeredWorkflows); } await _workflowDispatcher.DispatchAsync(new DispatchTriggerWorkflowsRequest(activityTypeName, bookmarkPayload) @@ -118,9 +134,11 @@ public class DefaultWorkflowInbox : IWorkflowInbox ActivityInstanceId = activityInstanceId, Input = input }, cancellationToken); + + return new DeliverWorkflowInboxMessageResult(new List()); } - private async Task ResumeWorkflowsAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) + private async Task ResumeWorkflowsAsynchronouslyAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) { var activityTypeName = message.ActivityTypeName; var correlationId = message.CorrelationId; @@ -138,6 +156,25 @@ public class DefaultWorkflowInbox : IWorkflowInbox }, cancellationToken); } + private async Task> ResumeWorkflowsSynchronouslyAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) + { + var activityTypeName = message.ActivityTypeName; + var correlationId = message.CorrelationId; + var workflowInstanceId = message.WorkflowInstanceId; + var activityInstanceId = message.ActivityInstanceId; + var bookmarkPayload = message.BookmarkPayload; + var input = message.Input; + + return await _workflowRuntime.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, new TriggerWorkflowsOptions + { + CorrelationId = correlationId, + WorkflowInstanceId = workflowInstanceId, + ActivityInstanceId = activityInstanceId, + Input = input, + CancellationTokens = cancellationToken + }); + } + /// public async ValueTask> FindManyAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default) { diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs index f97926a0c..11730d0af 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs @@ -423,11 +423,5 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime private async Task AcquireLockAsync(string resource, CancellationToken cancellationToken) { return await _distributedLockProvider.AcquireLockAsync(resource, TimeSpan.FromMinutes(2), cancellationToken); - // if (AcquiredLock.Value?.Key == resource) - // return AcquiredLock.Value.Lock; - // - // var distributedLock = await _distributedLockProvider.AcquireLockAsync(resource, TimeSpan.FromMinutes(2), cancellationToken); - // AcquiredLock.Value = new AcquiredLock { Lock = distributedLock, Key = resource }; - // return distributedLock; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs b/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs index ff6d59e0c..1acd1159f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs @@ -31,7 +31,7 @@ public class EventPublisher : IEventPublisher IDictionary? input = default, CancellationToken cancellationToken = default) { - return await PublishInternalAsync(eventName, NotificationStrategy.Sequential, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken); + return await PublishInternalAsync(eventName, false, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken); } /// @@ -43,12 +43,12 @@ public class EventPublisher : IEventPublisher IDictionary? input = default, CancellationToken cancellationToken = default) { - await PublishInternalAsync(eventName, NotificationStrategy.FireAndForget, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken); + await PublishInternalAsync(eventName, true, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken); } private async Task> PublishInternalAsync( string eventName, - IEventPublishingStrategy publishingStrategy, + bool dispatchAsynchronously, string? correlationId = default, string? workflowInstanceId = default, string? activityInstanceId = default, @@ -59,7 +59,7 @@ public class EventPublisher : IEventPublisher var message = NewWorkflowInboxMessage.For(eventBookmark, workflowInstanceId, correlationId, activityInstanceId, input); var options = new WorkflowInboxMessageDeliveryOptions { - EventPublishingStrategy = publishingStrategy, + DispatchAsynchronously = dispatchAsynchronously }; var result = await _workflowInbox.SubmitAsync(message, options, cancellationToken);