From cc24c5c759d46212a67ab6eed79a192dddbe050f Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 18 Dec 2023 16:05:51 +0100 Subject: [PATCH] Add bookmarks processing to DefaultBackgroundActivityInvoker The code addition includes two dependencies, IBookmarksPersister and IWorkflowStateExtractor, in the DefaultBackgroundActivityInvoker service. With these changes, during the workflow execution phase, bookmarks are captured and processed for any state changes. These modifications provide more control and efficiency over handling workflow states during the system's background operations. --- .../DefaultBackgroundActivityInvoker.cs | 43 +++++++++++++------ 1 file changed, 31 insertions(+), 12 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs index 5084146de..1c6e50ddf 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs @@ -1,6 +1,7 @@ using Elsa.Common.Models; using Elsa.Workflows.Core; using Elsa.Workflows.Core.Contracts; +using Elsa.Workflows.Core.Helpers; using Elsa.Workflows.Core.Memory; using Elsa.Workflows.Core.Models; using Elsa.Workflows.Core.Services; @@ -23,6 +24,8 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker private readonly IWorkflowDefinitionService _workflowDefinitionService; private readonly IVariablePersistenceManager _variablePersistenceManager; private readonly IActivityInvoker _activityInvoker; + private readonly IBookmarksPersister _bookmarksPersister; + private readonly IWorkflowStateExtractor _workflowStateExtractor; private readonly IServiceProvider _serviceProvider; private readonly ILogger _logger; @@ -35,6 +38,8 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker IWorkflowDefinitionService workflowDefinitionService, IVariablePersistenceManager variablePersistenceManager, IActivityInvoker activityInvoker, + IBookmarksPersister bookmarksPersister, + IWorkflowStateExtractor workflowStateExtractor, IServiceProvider serviceProvider, ILogger logger) { @@ -43,6 +48,8 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker _workflowDefinitionService = workflowDefinitionService; _variablePersistenceManager = variablePersistenceManager; _activityInvoker = activityInvoker; + _bookmarksPersister = bookmarksPersister; + _workflowStateExtractor = workflowStateExtractor; _serviceProvider = serviceProvider; _logger = logger; } @@ -63,6 +70,7 @@ 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); @@ -82,31 +90,42 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker foreach (var outputDescriptor in outputDescriptors) { var output = (Output?)outputDescriptor.ValueGetter(activityExecutionContext.Activity); - - if(output == null) + + if (output == null) continue; var memoryBlockReference = output.MemoryBlockReference(); - - if(!activityExecutionContext.ExpressionExecutionContext.TryGetBlock(memoryBlockReference, out var memoryBlock)) + + if (!activityExecutionContext.ExpressionExecutionContext.TryGetBlock(memoryBlockReference, out var memoryBlock)) continue; - + var variableMetadata = memoryBlock.Metadata as VariableBlockMetadata; var driver = variableMetadata?.StorageDriverType; - + // We only capture output written to the workflow itself. Other drivers like blob storage, etc. will be ignored since the foreground context will be loading those. - if(driver != typeof(WorkflowStorageDriver)) + if (driver != typeof(WorkflowStorageDriver)) continue; - + var outputValue = activityExecutionContext.Get(memoryBlockReference); if (outputValue != null) outputValues[outputDescriptor.Name] = outputValue; } - // Persist any variables that were written to by the activity. + // 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. var bookmarkId = scheduledBackgroundActivity.BookmarkId; @@ -122,12 +141,12 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker } }; - if(cancellationToken.IsCancellationRequested) + if (cancellationToken.IsCancellationRequested) { _logger.LogInformation("Background execution for activity {ActivityNodeId} was canceled", activityNodeId); return; } - + await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken); } } \ No newline at end of file