diff --git a/src/modules/Elsa.Http/Activities/FlowSendHttpRequest.cs b/src/modules/Elsa.Http/Activities/FlowSendHttpRequest.cs index 2960b5887..cae3cbacf 100644 --- a/src/modules/Elsa.Http/Activities/FlowSendHttpRequest.cs +++ b/src/modules/Elsa.Http/Activities/FlowSendHttpRequest.cs @@ -12,7 +12,7 @@ namespace Elsa.Http; /// /// Send an HTTP request. /// -[Activity("Elsa", "HTTP", "Send an HTTP request.", DisplayName = "HTTP Request (flow)", Kind = ActivityKind.Task)] +[Activity("Elsa", "HTTP", "Send an HTTP request.", DisplayName = "HTTP Request (flow)", Kind = ActivityKind.Action)] public class FlowSendHttpRequest : SendHttpRequestBase, IActivityPropertyDefaultValueProvider { /// diff --git a/src/modules/Elsa.Http/Activities/SendHttpRequest.cs b/src/modules/Elsa.Http/Activities/SendHttpRequest.cs index b45e5558f..0fd76e72c 100644 --- a/src/modules/Elsa.Http/Activities/SendHttpRequest.cs +++ b/src/modules/Elsa.Http/Activities/SendHttpRequest.cs @@ -10,7 +10,7 @@ namespace Elsa.Http; /// /// Send an HTTP request. /// -[Activity("Elsa", "HTTP", "Send an HTTP request.", DisplayName = "HTTP Request", Kind = ActivityKind.Task)] +[Activity("Elsa", "HTTP", "Send an HTTP request.", DisplayName = "HTTP Request", Kind = ActivityKind.Action)] public class SendHttpRequest : SendHttpRequestBase { /// diff --git a/src/modules/Elsa.Http/Activities/SendHttpRequestTask.cs b/src/modules/Elsa.Http/Activities/SendHttpRequestTask.cs new file mode 100644 index 000000000..3c2c114ba --- /dev/null +++ b/src/modules/Elsa.Http/Activities/SendHttpRequestTask.cs @@ -0,0 +1,220 @@ +using System.Net; +using System.Net.Http.Headers; +using System.Runtime.CompilerServices; +using Elsa.Extensions; +using Elsa.Http.ContentWriters; +using Elsa.Http.UIHints; +using Elsa.Workflows; +using Elsa.Workflows.Attributes; +using Elsa.Workflows.UIHints; +using Elsa.Workflows.Models; +using HttpHeaders = Elsa.Http.Models.HttpHeaders; + +namespace Elsa.Http; + +/// +/// Sends HTTP requests from a background task. +/// +[Activity("Elsa", "HTTP", "Send an HTTP request from a background task.", DisplayName = "HTTP Request Task", Kind = ActivityKind.Job)] +public class SendHttpRequestTask : Activity +{ + /// + public SendHttpRequestTask([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) + { + } + + /// + /// The URL to send the request to. + /// + [Input(Description = "The URL to send the request to.")] + public Input Url { get; set; } = default!; + + /// + /// The HTTP method to use when sending the request. + /// + [Input( + Description = "The HTTP method to use when sending the request.", + Options = new[] { "GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD" }, + DefaultValue = "GET", + UIHint = InputUIHints.DropDown + )] + public Input Method { get; set; } = new("GET"); + + /// + /// The content to send with the request. Can be a string, an object, a byte array or a stream. + /// + [Input(Description = "The content to send with the request. Can be a string, an object, a byte array or a stream.")] + public Input Content { get; set; } = default!; + + /// + /// The content type to use when sending the request. + /// + [Input( + Description = "The content type to use when sending the request.", + UIHandler = typeof(HttpContentTypeOptionsProvider), + UIHint = InputUIHints.DropDown + )] + public Input ContentType { get; set; } = default!; + + /// + /// The Authorization header value to send with the request. + /// + /// Bearer {some-access-token} + [Input(Description = "The Authorization header value to send with the request. For example: Bearer {some-access-token}", Category = "Security")] + public Input Authorization { get; set; } = default!; + + /// + /// A list of expected status codes to handle. + /// + [Input( + Description = "A list of expected status codes to handle.", + UIHint = InputUIHints.MultiText, + DefaultValueProvider = typeof(FlowSendHttpRequest) + )] + public Input> ExpectedStatusCodes { get; set; } = default!; + + /// + /// A value that allows to add the Authorization header without validation. + /// + [Input(Description = "A value that allows to add the Authorization header without validation.", Category = "Security")] + public Input DisableAuthorizationHeaderValidation { get; set; } = default!; + + /// + /// The headers to send along with the request. + /// + [Input( + Description = "The headers to send along with the request.", + UIHint = InputUIHints.JsonEditor, + Category = "Advanced" + )] + public Input RequestHeaders { get; set; } = new(new HttpHeaders()); + + /// + /// The parsed content, if any. + /// + [Output(Description = "The parsed content, if any.")] + public Output ParsedContent { get; set; } = default!; + + /// + protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) + { + await TrySendAsync(context); + } + + private async Task TrySendAsync(ActivityExecutionContext context) + { + var request = PrepareRequest(context); + var httpClientFactory = context.GetRequiredService(); + var httpClient = httpClientFactory.CreateClient(nameof(SendHttpRequestBase)); + var cancellationToken = context.CancellationToken; + + try + { + var response = await httpClient.SendAsync(request, cancellationToken); + var parsedContent = await ParseContentAsync(context, response.Content); + context.SetResult(response.StatusCode); + context.Set(ParsedContent, parsedContent); + + HandleResponse(context, response); + } + catch (HttpRequestException e) + { + context.AddExecutionLogEntry("Error", e.Message, payload: new { StackTrace = e.StackTrace }); + context.JournalData.Add("Error", e.Message); + HandleRequestException(context, e); + } + catch (TaskCanceledException e) + { + context.AddExecutionLogEntry("Error", e.Message, payload: new { StackTrace = e.StackTrace }); + context.JournalData.Add("Cancelled", true); + HandleTaskCanceledException(context, e); + } + } + + private void HandleResponse(ActivityExecutionContext context, HttpResponseMessage response) + { + var expectedStatusCodes = ExpectedStatusCodes.GetOrDefault(context) ?? new List(0); + var statusCode = (int)response.StatusCode; + var hasMatchingStatusCode = expectedStatusCodes.Contains(statusCode); + var outcome = expectedStatusCodes.Any() ? hasMatchingStatusCode ? statusCode.ToString() : "Unmatched status code" : default; + var outcomes = new List(); + + if (outcome != null) + outcomes.Add(outcome); + + outcomes.Add("Done"); + context.JournalData["StatusCode"] = statusCode; + context.SetBackgroundOutcomes(outcomes); + } + + private void HandleRequestException(ActivityExecutionContext context, HttpRequestException exception) + { + context.SetBackgroundOutcomes(new[] { "Failed to connect" }); + } + + private void HandleTaskCanceledException(ActivityExecutionContext context, TaskCanceledException exception) + { + context.SetBackgroundOutcomes(new[] { "Timeout" }); + } + + private async Task ParseContentAsync(ActivityExecutionContext context, HttpContent httpContent) + { + if (!HasContent(httpContent)) + return null; + + var cancellationToken = context.CancellationToken; + var targetType = ParsedContent.GetTargetType(context); + var contentStream = await httpContent.ReadAsStreamAsync(cancellationToken); + var contentType = httpContent.Headers.ContentType?.MediaType!; + + targetType ??= contentType switch + { + "application/json" => typeof(object), + _ => typeof(string) + }; + + return await context.ParseContentAsync(contentStream, contentType, targetType, cancellationToken); + } + + private static bool HasContent(HttpContent httpContent) => httpContent.Headers.ContentLength > 0; + + private HttpRequestMessage PrepareRequest(ActivityExecutionContext context) + { + var method = Method.GetOrDefault(context) ?? "GET"; + var url = Url.Get(context); + var request = new HttpRequestMessage(new HttpMethod(method), url); + var headers = context.GetHeaders(RequestHeaders); + var authorization = Authorization.GetOrDefault(context); + var addAuthorizationWithoutValidation = DisableAuthorizationHeaderValidation.GetOrDefault(context); + + if (!string.IsNullOrWhiteSpace(authorization)) + if (addAuthorizationWithoutValidation) + request.Headers.TryAddWithoutValidation("Authorization", authorization); + else + request.Headers.Authorization = AuthenticationHeaderValue.Parse(authorization); + + foreach (var header in headers) + request.Headers.Add(header.Key, header.Value.AsEnumerable()); + + var contentType = ContentType.GetOrDefault(context); + var content = Content.GetOrDefault(context); + + if (contentType != null && content != null) + { + var factories = context.GetServices(); + var factory = SelectContentWriter(contentType, factories); + request.Content = factory.CreateHttpContent(content, contentType); + } + + return request; + } + + private IHttpContentFactory SelectContentWriter(string? contentType, IEnumerable factories) + { + if (string.IsNullOrWhiteSpace(contentType)) + return new JsonContentFactory(); + + var parsedContentType = new System.Net.Mime.ContentType(contentType); + return factories.FirstOrDefault(httpContentFactory => httpContentFactory.SupportedContentTypes.Any(c => c == parsedContentType.MediaType)) ?? new JsonContentFactory(); + } +} \ No newline at end of file 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..d587de713 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.TransientProperties.GetValueOrDefault(BackgroundActivityInvokerMiddleware.IsBackgroundExecution); // Is the activity configured to load the context? foreach (var providerType in providerTypes) 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..c4cc0de38 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Extensions/BackgroundActivityExecutionContextExtensions.cs @@ -0,0 +1,24 @@ +namespace Elsa.Workflows; + +/// +/// Adds extension methods to . +/// +public static class BackgroundActivityExecutionContextExtensions +{ + /// + /// 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(); + } +} \ 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/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/IWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs index 3421c4cb6..746481379 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs @@ -88,9 +88,4 @@ public interface IWorkflowRuntime /// Counts the number of workflow instances based on the provided query args. /// Task CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default); - - /// - /// Merges the specified workflow state into the workflow runtime. - /// - Task MergeWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default); } \ No newline at end of file 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/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs index da4f05bf5..c10cf90d1 100644 --- a/src/modules/Elsa.Workflows.Runtime/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs +++ b/src/modules/Elsa.Workflows.Runtime/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs @@ -12,19 +12,21 @@ 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 readonly object BackgroundActivitySchedulesKey = new(); internal const string BackgroundActivityBookmarkName = "BackgroundActivity"; /// - public BackgroundActivityCollectorMiddleware(ActivityMiddlewareDelegate next) : base(next) + public BackgroundActivityInvokerMiddleware(ActivityMiddlewareDelegate next) : base(next) { } @@ -37,8 +39,16 @@ 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 CompleteBackgroundActivityAsync(context); + } } } @@ -51,11 +61,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(IsBackgroundExecution); + /// /// Schedules the current activity for execution in the background. /// @@ -77,21 +89,48 @@ public class BackgroundActivityCollectorMiddleware : DefaultActivityInvokerMiddl private static void CaptureOutputIfAny(ActivityExecutionContext context) { var activity = context.Activity; - var inputKey = GetBackgroundActivityOutputKey(activity.Id); - - if (!context.WorkflowInput.TryGetValue(inputKey, out var capturedOutput)) + var inputKey = GetBackgroundActivityOutputKey(activity.NodeId); + var capturedOutput = context.WorkflowExecutionContext.GetProperty>(inputKey); + + 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 CompleteBackgroundActivityAsync(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); } } } \ 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/Services/DefaultBackgroundActivityInvoker.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs index 55d0f8697..379a10e24 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs @@ -70,7 +70,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); @@ -78,7 +77,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.TransientProperties[BackgroundActivityInvokerMiddleware.IsBackgroundExecution] = true; // Invoke the activity. await _activityInvoker.InvokeAsync(activityExecutionContext); @@ -86,6 +85,7 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker // Capture any activity output produced by the activity (but only if the associated memory block is stored in the workflow itself). var outputDescriptors = activityExecutionContext.ActivityDescriptor.Outputs; var outputValues = new Dictionary(); + var outcomes = activityExecutionContext.GetBackgroundOutcomes().ToList(); foreach (var outputDescriptor in outputDescriptors) { @@ -112,33 +112,21 @@ 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.MergeWorkflowStateAsync(workflowState, cancellationToken); - //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. 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 dispatchRequest = new DispatchWorkflowInstanceRequest { - InstanceId = workflowInstanceId, - BookmarkId = bookmarkId, - Input = new Dictionary + InstanceId = workflowInstanceId, + BookmarkId = bookmarkId, + Properties = new Dictionary { - [inputKey] = outputValues + [outcomesKey] = outcomes, + [inputKey] = outputValues, + [journalDataKey] = activityExecutionContext.JournalData } }; diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs index 4c00e1e67..11730d0af 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs @@ -292,19 +292,6 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime return await _workflowInstanceStore.CountAsync(filter, cancellationToken); } - public async Task MergeWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default) - { - var existingWorkflowInstance = (await _workflowInstanceStore.FindAsync(workflowState.Id, cancellationToken))!; - var workflowInstance = _workflowStateMapper.Map(workflowState)!; - - foreach (var bookmark in workflowState.Bookmarks) - { - existingWorkflowInstance.WorkflowState.Bookmarks.RemoveWhere(x => x.Id == bookmark.Id); - existingWorkflowInstance.WorkflowState.Bookmarks.Add(bookmark); - } - await _workflowInstanceManager.SaveAsync(workflowInstance, cancellationToken); - } - private async Task StartWorkflowAsync(IWorkflowHost workflowHost, StartWorkflowRuntimeOptions options) { var workflowInstanceId = string.IsNullOrEmpty(options.InstanceId) ? _identityGenerator.GenerateId() : options.InstanceId;