| | | 1 | | using System.Reflection; |
| | | 2 | | using Elsa.Extensions; |
| | | 3 | | using Elsa.Mediator.Contracts; |
| | | 4 | | using Elsa.Workflows.Activities; |
| | | 5 | | using Elsa.Workflows.CommitStates; |
| | | 6 | | using Elsa.Workflows.Pipelines.ActivityExecution; |
| | | 7 | | using Microsoft.Extensions.Logging; |
| | | 8 | | using Microsoft.Extensions.Options; |
| | | 9 | | |
| | | 10 | | namespace Elsa.Workflows.Middleware.Activities; |
| | | 11 | | |
| | | 12 | | /// <summary> |
| | | 13 | | /// Provides extension methods to <see cref="IActivityExecutionPipelineBuilder"/>. |
| | | 14 | | /// </summary> |
| | | 15 | | public static class ActivityInvokerMiddlewareExtensions |
| | | 16 | | { |
| | | 17 | | /// <summary> |
| | | 18 | | /// Adds the <see cref="DefaultActivityInvokerMiddleware"/> component to the pipeline. |
| | | 19 | | /// </summary> |
| | 0 | 20 | | public static IActivityExecutionPipelineBuilder UseDefaultActivityInvoker(this IActivityExecutionPipelineBuilder pip |
| | | 21 | | } |
| | | 22 | | |
| | | 23 | | /// <summary> |
| | | 24 | | /// A default activity execution middleware component that evaluates the current activity's properties, executes the act |
| | | 25 | | /// </summary> |
| | | 26 | | public class DefaultActivityInvokerMiddleware(ActivityMiddlewareDelegate next, ICommitStrategyRegistry commitStrategyReg |
| | | 27 | | : IActivityExecutionMiddleware |
| | | 28 | | { |
| | | 29 | | private static readonly MethodInfo ExecuteAsyncMethodInfo = typeof(IActivity).GetMethod(nameof(IActivity.ExecuteAsyn |
| | | 30 | | |
| | | 31 | | /// <inheritdoc /> |
| | | 32 | | public async ValueTask InvokeAsync(ActivityExecutionContext context) |
| | | 33 | | { |
| | | 34 | | context.CancellationToken.ThrowIfCancellationRequested(); |
| | | 35 | | |
| | | 36 | | var workflowExecutionContext = context.WorkflowExecutionContext; |
| | | 37 | | |
| | | 38 | | // Evaluate input properties. |
| | | 39 | | await EvaluateInputPropertiesAsync(context); |
| | | 40 | | |
| | | 41 | | // Prevent the activity from being started if cancellation is requested. |
| | | 42 | | if (context.CancellationToken.IsCancellationRequested) |
| | | 43 | | { |
| | | 44 | | context.TransitionTo(ActivityStatus.Canceled); |
| | | 45 | | context.AddExecutionLogEntry("Activity cancelled"); |
| | | 46 | | return; |
| | | 47 | | } |
| | | 48 | | |
| | | 49 | | // Check if the activity can be executed. |
| | | 50 | | if (!await context.Activity.CanExecuteAsync(context)) |
| | | 51 | | { |
| | | 52 | | context.TransitionTo(ActivityStatus.Pending); |
| | | 53 | | context.AddExecutionLogEntry("Precondition Failed", "Cannot execute at this time"); |
| | | 54 | | return; |
| | | 55 | | } |
| | | 56 | | |
| | | 57 | | // Mark workflow and activity as executing. |
| | | 58 | | using var executionState = context.EnterExecution(); |
| | | 59 | | |
| | | 60 | | // Conditionally commit the workflow state. |
| | | 61 | | if (ShouldCommit(context, ActivityLifetimeEvent.ActivityExecuting)) |
| | | 62 | | await context.WorkflowExecutionContext.CommitAsync(); |
| | | 63 | | |
| | | 64 | | var previousActivityStatus = context.Status; |
| | | 65 | | context.TransitionTo(ActivityStatus.Running); |
| | | 66 | | |
| | | 67 | | // Execute activity. |
| | | 68 | | await ExecuteActivityAsync(context); |
| | | 69 | | |
| | | 70 | | var currentActivityStatus = context.Status; |
| | | 71 | | var activityDidComplete = previousActivityStatus != ActivityStatus.Completed && currentActivityStatus == Activit |
| | | 72 | | |
| | | 73 | | // Reset execute delegate. |
| | | 74 | | workflowExecutionContext.ExecuteDelegate = null; |
| | | 75 | | |
| | | 76 | | // If a bookmark was used to resume, burn it if not burnt already by the activity. |
| | | 77 | | var resumedBookmark = workflowExecutionContext.ResumedBookmarkContext?.Bookmark; |
| | | 78 | | |
| | | 79 | | if (resumedBookmark is { AutoBurn: true }) |
| | | 80 | | { |
| | | 81 | | logger.LogDebug("Auto-burning bookmark {BookmarkId}", resumedBookmark.Id); |
| | | 82 | | workflowExecutionContext.Bookmarks.Remove(resumedBookmark); |
| | | 83 | | } |
| | | 84 | | |
| | | 85 | | // Update execution count. |
| | | 86 | | context.IncrementExecutionCount(); |
| | | 87 | | |
| | | 88 | | // Invoke next middleware. |
| | | 89 | | await next(context); |
| | | 90 | | |
| | | 91 | | // If the activity completed, send a notification. |
| | | 92 | | if (activityDidComplete) |
| | | 93 | | { |
| | | 94 | | var mediator = context.GetRequiredService<INotificationSender>(); |
| | | 95 | | await mediator.SendAsync(new Notifications.ActivityCompleted(context), context.CancellationToken); |
| | | 96 | | } |
| | | 97 | | |
| | | 98 | | // Conditionally commit the workflow state. |
| | | 99 | | if (ShouldCommit(context, ActivityLifetimeEvent.ActivityExecuted)) |
| | | 100 | | await context.WorkflowExecutionContext.CommitAsync(); |
| | | 101 | | } |
| | | 102 | | |
| | | 103 | | /// <summary> |
| | | 104 | | /// Executes the activity using the specified context. |
| | | 105 | | /// This method is virtual so that modules might override this implementation to do things like e.g. asynchronous pr |
| | | 106 | | /// </summary> |
| | | 107 | | protected virtual async ValueTask ExecuteActivityAsync(ActivityExecutionContext context) |
| | | 108 | | { |
| | | 109 | | var executeDelegate = context.WorkflowExecutionContext.ExecuteDelegate |
| | | 110 | | ?? (ExecuteActivityDelegate)Delegate.CreateDelegate(typeof(ExecuteActivityDelegate), conte |
| | | 111 | | |
| | | 112 | | await executeDelegate(context); |
| | | 113 | | } |
| | | 114 | | |
| | | 115 | | private async Task EvaluateInputPropertiesAsync(ActivityExecutionContext context) |
| | | 116 | | { |
| | | 117 | | // Evaluate containing composite input properties, if any. |
| | | 118 | | var compositeContainerContexts = context.GetAncestors().Where(x => x.Activity is Composite).ToList(); |
| | | 119 | | |
| | | 120 | | foreach (var activityExecutionContext in compositeContainerContexts) |
| | | 121 | | { |
| | | 122 | | if (!activityExecutionContext.GetHasEvaluatedProperties()) |
| | | 123 | | await activityExecutionContext.EvaluateInputPropertiesAsync(); |
| | | 124 | | } |
| | | 125 | | |
| | | 126 | | // Evaluate input properties. |
| | | 127 | | await context.EvaluateInputPropertiesAsync(); |
| | | 128 | | } |
| | | 129 | | |
| | | 130 | | private bool ShouldCommit(ActivityExecutionContext context, ActivityLifetimeEvent lifetimeEvent) |
| | | 131 | | { |
| | | 132 | | var strategyName = context.Activity.GetCommitStrategy(); |
| | | 133 | | |
| | | 134 | | IActivityCommitStrategy? strategy = !string.IsNullOrWhiteSpace(strategyName) |
| | | 135 | | ? commitStrategyRegistry.FindActivityStrategy(strategyName) |
| | | 136 | | : commitStateOptions.Value.DefaultActivityCommitStrategy; |
| | | 137 | | |
| | | 138 | | var commitAction = CommitAction.Default; |
| | | 139 | | |
| | | 140 | | if (strategy != null) |
| | | 141 | | { |
| | | 142 | | var strategyContext = new ActivityCommitStateStrategyContext(context, lifetimeEvent); |
| | | 143 | | commitAction = strategy.ShouldCommit(strategyContext); |
| | | 144 | | } |
| | | 145 | | |
| | | 146 | | switch (commitAction) |
| | | 147 | | { |
| | | 148 | | case CommitAction.Skip: |
| | | 149 | | return false; |
| | | 150 | | case CommitAction.Commit: |
| | | 151 | | return true; |
| | | 152 | | case CommitAction.Default: |
| | | 153 | | { |
| | | 154 | | var workflowStrategyName = context.WorkflowExecutionContext.Workflow.Options.CommitStrategyName; |
| | | 155 | | |
| | | 156 | | IWorkflowCommitStrategy? workflowStrategy = !string.IsNullOrWhiteSpace(workflowStrategyName) |
| | | 157 | | ? commitStrategyRegistry.FindWorkflowStrategy(workflowStrategyName) |
| | | 158 | | : commitStateOptions.Value.DefaultWorkflowCommitStrategy; |
| | | 159 | | |
| | | 160 | | if (workflowStrategy == null) |
| | | 161 | | return false; |
| | | 162 | | |
| | | 163 | | var workflowLifetimeEvent = lifetimeEvent == ActivityLifetimeEvent.ActivityExecuting ? WorkflowLifet |
| | | 164 | | var workflowCommitStateStrategyContext = new WorkflowCommitStateStrategyContext(context.WorkflowExec |
| | | 165 | | commitAction = workflowStrategy.ShouldCommit(workflowCommitStateStrategyContext); |
| | | 166 | | |
| | | 167 | | return commitAction == CommitAction.Commit; |
| | | 168 | | } |
| | | 169 | | default: |
| | | 170 | | throw new ArgumentOutOfRangeException(nameof(commitAction), commitAction, "Unknown commit action"); |
| | | 171 | | } |
| | | 172 | | } |
| | | 173 | | } |