SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Fix Alteration Runner (#5142)

Browse source 
* Refactor alteration runner logic and simplify CancelActivityHandler

The DefaultAlterationRunner.cs file was refactored to include a workflow middleware pipeline, replacing the previous method of updating workflow state. Comments were added to discuss potential solutions for reusing the same pipeline in the workflow runtime. The CancelActivityHandler was simplified by removing multiple functions and replacing them with a single CancelAsync method.

* Add RunAlterationsMiddleware to handle workflow alterations

The commit introduces a new middleware, RunAlterationsMiddleware, that is designed to process workflow alterations. The middleware is in charge of executing alteration handlers and taking care of any required commit actions. The original code for handling alterations in DefaultAlterationRunner has been significantly reduced as it now delegates most of its responsibility to this new middleware.

* Add comments to IAlterationPlanManager interface methods

In the IAlterationPlanManager interface, explanatory comments were added to each method. These include methods for getting a plan by ID, checking if all jobs in the plan have been completed, and completing an alteration plan. The changes made will greatly aid in understanding the purpose and functionality of each method.

* Refactor pipeline execution methods and replace middleware

The workflow execution pipeline has been refactored to allow for dynamic configuration. The middleware components are now retrievable properties and can be replaced individually. This change provides more extensibility with modifying the pipeline execution and replacing the DefaultActivitySchedulerMiddleware with desired middleware.

* Refactor pipeline alteration and addition methods

Simplified the pipeline alteration process in Elsa.Alterations. Instead of manually handling middleware delegates, added a new extension method, ReplaceTerminal, in WorkflowExecutionMiddlewareExtensions.cs to replace the terminal middleware component. This approach improves code readability and maintenance.

* Refactor RunAlterationsMiddleware class

Removed an unnecessary extension class and updated the RunAlterationsMiddleware class to streamline its structure. The handlers are now initialized directly in the constructor, eliminating the need for an additional field. Renamed local variable for better code clarity.

* Refactor variable name in RunAlterationsMiddleware

The variable name 'handlers1' has been renamed to 'supportedHandlers' in the RunAlterationsMiddleware.cs file. This change improves code readability and makes it clear that the list contains only the handlers that can handle the given alteration.

* Change RunAsync method to return void

The RunAsync function in the RunAlterationsMiddleware class no longer returns a boolean value. The returned 'false' has been replaced with a 'return' statement, and the 'return true' statement has been completely removed. This refactor simplifies the control flow when executing alterations.
This commit is contained in:
Sipke Schoorstra 2024-03-29 19:58:35 +01:00 committed by GitHub
parent f0eb19ac22
commit 0993b38b7a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 186 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
src/modules/Elsa.Alterations.Core/Contracts/IAlterationPlanManager.cs
View file

@ -2,9 +2,23 @@ using Elsa.Alterations.Core.Entities;
namespace Elsa.Alterations.Core.Contracts;
/// <summary>
/// Represents a manager for alteration plans.
/// </summary>
public interface IAlterationPlanManager
{
/// <summary>
/// Gets an alteration plan by ID.
/// </summary>
Task<AlterationPlan?> GetPlanAsync(string planId, CancellationToken cancellationToken = default);
/// <summary>
/// Gets a value indicating whether all jobs in the plan have been completed.
/// </summary>
Task<bool> GetIsAllJobsCompletedAsync(string planId, CancellationToken cancellationToken = default);
/// <summary>
/// Completes an alteration plan.
/// </summary>
Task CompletePlanAsync(AlterationPlan plan, CancellationToken cancellationToken = default);
}

42
src/modules/Elsa.Alterations/AlterationHandlers/CancelActivityHandler.cs
View file

@ -1,9 +1,8 @@
using Elsa.Alterations.AlterationTypes;
using Elsa.Alterations.Core.Abstractions;
using Elsa.Alterations.Core.Contexts;
using Elsa.Extensions;
using Elsa.Workflows;
using Elsa.Workflows.Runtime.Contracts;
using Elsa.Workflows.Runtime.Filters;
using JetBrains.Annotations;
namespace Elsa.Alterations.AlterationHandlers;
@ -14,20 +13,6 @@ namespace Elsa.Alterations.AlterationHandlers;
[UsedImplicitly]
public class CancelActivityHandler : AlterationHandlerBase<CancelActivity>
{
private readonly IBookmarkManager _bookmarkManager;
private readonly IActivityExecutionManager _activityExecutionManager;
private readonly IActivityExecutionMapper _activityExecutionMapper;
/// <summary>
/// Initializes a new instance of the <see cref="CancelActivityHandler"/> class.
/// </summary>
public CancelActivityHandler(IBookmarkManager bookmarkManager, IActivityExecutionManager activityExecutionManager, IActivityExecutionMapper activityExecutionMapper)
{
_bookmarkManager = bookmarkManager;
_activityExecutionManager = activityExecutionManager;
_activityExecutionMapper = activityExecutionMapper;
}
/// <inheritdoc />
protected override ValueTask HandleAsync(AlterationContext context, CancelActivity alteration)
{
@ -49,32 +34,19 @@ public class CancelActivityHandler : AlterationHandlerBase<CancelActivity>
return ValueTask.CompletedTask;
}
context.Succeed(() => CleanupAsync(activityExecutionContexts));
context.Succeed(() => CancelAsync(activityExecutionContexts));
return ValueTask.CompletedTask;
}
private async Task CleanupAsync(IEnumerable<ActivityExecutionContext> activityExecutionContexts)
private async Task CancelAsync(IEnumerable<ActivityExecutionContext> activityExecutionContexts)
{
foreach (var activityExecutionContext in activityExecutionContexts)
await CleanupAsync(activityExecutionContext);
}
private async Task CleanupAsync(ActivityExecutionContext activityExecutionContext)
{
await RemoveBookmarksAsync(activityExecutionContext);
await SaveActivityExecutionRecordAsync(activityExecutionContext);
foreach (var activityExecutionContext in activityExecutionContexts)
await CancelAsync(activityExecutionContext);
}
private async Task RemoveBookmarksAsync(ActivityExecutionContext activityExecutionContext)
private async Task CancelAsync(ActivityExecutionContext activityExecutionContext)
{
var filter = new BookmarkFilter { ActivityInstanceId = activityExecutionContext.Id };
await _bookmarkManager.DeleteManyAsync(filter, activityExecutionContext.CancellationToken);
}
private async Task SaveActivityExecutionRecordAsync(ActivityExecutionContext activityExecutionContext)
{
var activityExecutionRecord = _activityExecutionMapper.Map(activityExecutionContext);
await _activityExecutionManager.SaveAsync(activityExecutionRecord, CancellationToken.None);
await activityExecutionContext.CancelActivityAsync();
}
private static IEnumerable<ActivityExecutionContext> GetActivityExecutionContexts(AlterationContext context, CancelActivity alteration)

54
src/modules/Elsa.Alterations/Middleware/Workflows/RunAlterationsMiddleware.cs Normal file
View file

@ -0,0 +1,54 @@
using Elsa.Alterations.Core.Contexts;
using Elsa.Alterations.Core.Contracts;
using Elsa.Alterations.Core.Models;
using Elsa.Extensions;
using Elsa.Workflows;
using Elsa.Workflows.Pipelines.WorkflowExecution;
namespace Elsa.Alterations.Middleware.Workflows;
/// <summary>
/// Middleware that runs alterations.
/// </summary>
internal class RunAlterationsMiddleware(WorkflowMiddlewareDelegate next, IEnumerable<IAlterationHandler> handlers) : WorkflowExecutionMiddleware(next)
{
public static readonly object AlterationsPropertyKey = new();
public static readonly object AlterationsLogPropertyKey = new();
public override async ValueTask InvokeAsync(WorkflowExecutionContext context)
{
var alterations = (IEnumerable<IAlteration>)(context.TransientProperties.GetValue(AlterationsPropertyKey) ?? throw new InvalidOperationException("No alterations found in the transient properties."));
var log = (AlterationLog)(context.TransientProperties.GetValue(AlterationsLogPropertyKey) ?? throw new InvalidOperationException("No alteration log found in the transient properties."));
await RunAsync(context, alterations, log, context.CancellationTokens.ApplicationCancellationToken);
}
private async Task RunAsync(WorkflowExecutionContext workflowExecutionContext, IEnumerable<IAlteration> alterations, AlterationLog log, CancellationToken cancellationToken = default)
{
var commitActions = new List<Func<Task>>();
foreach (var alteration in alterations)
{
// Find handlers.
var supportedHandlers = handlers.Where(x => x.CanHandle(alteration)).ToList();
foreach (var handler in supportedHandlers)
{
// Execute handler.
var alterationContext = new AlterationContext(alteration, workflowExecutionContext, log, cancellationToken);
await handler.HandleAsync(alterationContext);
// If the handler has failed, exit.
if (alterationContext.HasFailed)
return;
// Collect the commit handler, if any.
if (alterationContext.CommitAction != null)
commitActions.Add(alterationContext.CommitAction);
}
}
// Execute commit handlers.
foreach (var commitAction in commitActions)
await commitAction();
}
}

63
src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs
View file

@ -2,10 +2,13 @@ using Elsa.Alterations.Core.Contexts;
using Elsa.Alterations.Core.Contracts;
using Elsa.Alterations.Core.Models;
using Elsa.Alterations.Core.Results;
using Elsa.Alterations.Middleware.Workflows;
using Elsa.Common.Contracts;
using Elsa.Extensions;
using Elsa.Workflows;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Management.Contracts;
using Elsa.Workflows.Pipelines.WorkflowExecution;
using Elsa.Workflows.Runtime.Contracts;
using Microsoft.Extensions.Logging;
@ -14,8 +17,8 @@ namespace Elsa.Alterations.Services;
/// <inheritdoc />
public class DefaultAlterationRunner : IAlterationRunner
{
private readonly IEnumerable<IAlterationHandler> _handlers;
private readonly IWorkflowRuntime _workflowRuntime;
private readonly IWorkflowExecutionPipeline _workflowExecutionPipeline;
private readonly IWorkflowDefinitionService _workflowDefinitionService;
private readonly IWorkflowStateExtractor _workflowStateExtractor;
private readonly ISystemClock _systemClock;
@ -25,15 +28,15 @@ public class DefaultAlterationRunner : IAlterationRunner
/// Initializes a new instance of the <see cref="DefaultAlterationRunner"/> class.
/// </summary>
public DefaultAlterationRunner(
IEnumerable<IAlterationHandler> handlers,
IWorkflowRuntime workflowRuntime,
IWorkflowExecutionPipeline workflowExecutionPipeline,
IWorkflowDefinitionService workflowDefinitionService,
IWorkflowStateExtractor workflowStateExtractor,
ISystemClock systemClock,
IServiceProvider serviceProvider)
{
_handlers = handlers;
_workflowRuntime = workflowRuntime;
_workflowExecutionPipeline = workflowExecutionPipeline;
_workflowDefinitionService = workflowDefinitionService;
_workflowStateExtractor = workflowStateExtractor;
_systemClock = systemClock;
@ -86,15 +89,23 @@ public class DefaultAlterationRunner : IAlterationRunner
// Create workflow execution context.
var workflowExecutionContext = await WorkflowExecutionContext.CreateAsync(_serviceProvider, workflow, workflowState, cancellationTokens: cancellationToken);
workflowExecutionContext.TransientProperties.Add(RunAlterationsMiddleware.AlterationsPropertyKey, alterations);
workflowExecutionContext.TransientProperties.Add(RunAlterationsMiddleware.AlterationsLogPropertyKey, log);
// Execute alterations.
var success = await RunAsync(workflowExecutionContext, alterations, log, cancellationToken);
// If the alterations have failed, exit.
if (!success)
return result;
// Build a new workflow execution pipeline.
var pipelineBuilder = new WorkflowExecutionPipelineBuilder(_serviceProvider);
_workflowExecutionPipeline.ConfigurePipelineBuilder(pipelineBuilder);
// Update workflow state.
// Replace the terminal DefaultActivitySchedulerMiddleware with the RunAlterationsMiddleware terminal.
pipelineBuilder.ReplaceTerminal<RunAlterationsMiddleware>();
// Build modified pipeline.
var pipeline = pipelineBuilder.Build();
// Execute the pipeline.
await pipeline(workflowExecutionContext);
// Extract workflow state.
workflowState = _workflowStateExtractor.Extract(workflowExecutionContext);
// Apply updated workflow state.
@ -105,36 +116,4 @@ public class DefaultAlterationRunner : IAlterationRunner
return result;
}
private async Task<bool> RunAsync(WorkflowExecutionContext workflowExecutionContext, IEnumerable<IAlteration> alterations, AlterationLog log, CancellationToken cancellationToken = default)
{
var commitActions = new List<Func<Task>>();
foreach (var alteration in alterations)
{
// Find handlers.
var handlers = _handlers.Where(x => x.CanHandle(alteration)).ToList();
foreach (var handler in handlers)
{
// Execute handler.
var alterationContext = new AlterationContext(alteration, workflowExecutionContext, log, cancellationToken);
await handler.HandleAsync(alterationContext);
// If the handler has failed, exit.
if (alterationContext.HasFailed)
return false;
// Collect the commit handler, if any.
if (alterationContext.CommitAction != null)
commitActions.Add(alterationContext.CommitAction);
}
}
// Execute commit handlers.
foreach (var commitAction in commitActions)
await commitAction();
return true;
}
}

21
src/modules/Elsa.Workflows.Core/Contracts/IWorkflowExecutionPipeline.cs
View file

@ -2,9 +2,30 @@ using Elsa.Workflows.Pipelines.WorkflowExecution;
namespace Elsa.Workflows.Contracts;
/// <summary>
/// Represents a workflow execution pipeline.
/// </summary>
public interface IWorkflowExecutionPipeline
{
/// <summary>
/// The pipeline builder factory.
/// </summary>
Action<IWorkflowExecutionPipelineBuilder> ConfigurePipelineBuilder { get; }
/// <summary>
/// Sets up the pipeline using the specified pipeline builder factory.
/// </summary>
WorkflowMiddlewareDelegate Setup(Action<IWorkflowExecutionPipelineBuilder> setup);
/// <summary>
/// The constructed pipeline delegate.
/// </summary>
WorkflowMiddlewareDelegate Pipeline { get; }
/// <summary>
/// Executes the pipeline with the specified workflow execution context.
/// </summary>
/// <param name="context"></param>
/// <returns></returns>
Task ExecuteAsync(WorkflowExecutionContext context);
}

13
src/modules/Elsa.Workflows.Core/Contracts/IWorkflowExecutionPipelineBuilder.cs
View file

@ -12,6 +12,11 @@ public interface IWorkflowExecutionPipelineBuilder
/// </summary>
public IDictionary<object, object?> Properties { get; }
/// <summary>
/// The middleware components that have been installed.
/// </summary>
public IEnumerable<Func<WorkflowMiddlewareDelegate, WorkflowMiddlewareDelegate>> Components { get; }
/// <summary>
/// The current service provider to resolve services from.
/// </summary>
@ -31,4 +36,12 @@ public interface IWorkflowExecutionPipelineBuilder
/// Clears the current pipeline.
/// </summary>
IWorkflowExecutionPipelineBuilder Reset();
/// <summary>
/// Replaces the middleware component at the specified index with the specified delegate.
/// </summary>
/// <param name="index">The index of the middleware component to replace.</param>
/// <param name="middleware">The delegate to use as the new middleware component.</param>
/// <returns>A new instance of <see cref="IWorkflowExecutionPipelineBuilder"/> with the middleware component replaced.</returns>
IWorkflowExecutionPipelineBuilder Replace(int index, Func<WorkflowMiddlewareDelegate, WorkflowMiddlewareDelegate> middleware);
}

43
src/modules/Elsa.Workflows.Core/Pipelines/WorkflowExecution/WorkflowExecutionMiddlewareExtensions.cs
View file

@ -1,3 +1,4 @@
using System.Diagnostics.CodeAnalysis;
using Elsa.Workflows.Contracts;
using Microsoft.Extensions.DependencyInjection;
@ -11,16 +12,50 @@ public static class WorkflowExecutionMiddlewareExtensions
/// <summary>
/// Installs the specified middleware component into the pipeline being built.
/// </summary>
public static IWorkflowExecutionPipelineBuilder UseMiddleware<TMiddleware>(this IWorkflowExecutionPipelineBuilder pipelineBuilder, params object[] args) where TMiddleware: IWorkflowExecutionMiddleware
public static IWorkflowExecutionPipelineBuilder UseMiddleware<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] TMiddleware>(
this IWorkflowExecutionPipelineBuilder pipelineBuilder, params object[] args) where TMiddleware : IWorkflowExecutionMiddleware
{
var delegateFactory = CreateMiddlewareDelegateFactory<TMiddleware>(pipelineBuilder, args);
return pipelineBuilder.Use(delegateFactory);
}
/// <summary>
/// Replaces the terminal middleware component with the specified middleware component.
/// </summary>
public static IWorkflowExecutionPipelineBuilder ReplaceTerminal<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] TMiddleware>(
this IWorkflowExecutionPipelineBuilder pipelineBuilder, params object[] args) where TMiddleware : IWorkflowExecutionMiddleware
{
var index = pipelineBuilder.Components.Count() - 1;
return pipelineBuilder.Replace<TMiddleware>(index, args);
}
/// <summary>
/// Replaces the middleware component at the specified index with the specified middleware component.
/// </summary>
public static IWorkflowExecutionPipelineBuilder Replace<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] TMiddleware>(
this IWorkflowExecutionPipelineBuilder pipelineBuilder, int index, params object[] args) where TMiddleware : IWorkflowExecutionMiddleware
{
var delegateFactory = CreateMiddlewareDelegateFactory<TMiddleware>(pipelineBuilder, args);
return pipelineBuilder.Replace(index, delegateFactory);
}
/// <summary>
/// Creates a middleware delegate for the specified middleware component.
/// </summary>
public static Func<WorkflowMiddlewareDelegate, WorkflowMiddlewareDelegate> CreateMiddlewareDelegateFactory<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] TMiddleware>(
this IWorkflowExecutionPipelineBuilder pipelineBuilder, params object[] args) where TMiddleware : IWorkflowExecutionMiddleware
{
var middleware = typeof(TMiddleware);
return pipelineBuilder.Use(next =>
return next =>
{
var invokeMethod = MiddlewareHelpers.GetInvokeMethod(middleware);
var ctorParams = new[] { next }.Concat(args).Select(x => x).ToArray();
var ctorParams = new[]
{
next
}.Concat(args).Select(x => x).ToArray();
var instance = ActivatorUtilities.CreateInstance(pipelineBuilder.ServiceProvider, middleware, ctorParams);
return (WorkflowMiddlewareDelegate)invokeMethod.CreateDelegate(typeof(WorkflowMiddlewareDelegate), instance);
});
};
}
}

12
src/modules/Elsa.Workflows.Core/Pipelines/WorkflowExecution/WorkflowExecutionPipeline.cs
View file

@ -10,17 +10,21 @@ public class WorkflowExecutionPipeline : IWorkflowExecutionPipeline
private WorkflowMiddlewareDelegate? _pipeline;
/// <summary>
/// Constructor.
/// Initializes a new instance of the <see cref="WorkflowExecutionPipeline"/> class.
/// </summary>
public WorkflowExecutionPipeline(IServiceProvider serviceProvider, Action<IWorkflowExecutionPipelineBuilder> pipelineBuilder)
public WorkflowExecutionPipeline(IServiceProvider serviceProvider, Action<IWorkflowExecutionPipelineBuilder> configurePipelineBuilder)
{
_serviceProvider = serviceProvider;
Setup(pipelineBuilder);
ConfigurePipelineBuilder = configurePipelineBuilder;
Setup(configurePipelineBuilder);
}
/// <inheritdoc />
public WorkflowMiddlewareDelegate Pipeline => _pipeline ??= CreateDefaultPipeline();
/// <inheritdoc />
public Action<IWorkflowExecutionPipelineBuilder> ConfigurePipelineBuilder { get; }
/// <inheritdoc />
public WorkflowMiddlewareDelegate Setup(Action<IWorkflowExecutionPipelineBuilder> setup)
{

9
src/modules/Elsa.Workflows.Core/Pipelines/WorkflowExecution/WorkflowExecutionPipelineBuilder.cs
View file

@ -19,6 +19,9 @@ public class WorkflowExecutionPipelineBuilder : IWorkflowExecutionPipelineBuilde
/// <inheritdoc />
public IDictionary<object, object?> Properties { get; } = new Dictionary<object, object?>();
/// <inheritdoc />
public IEnumerable<Func<WorkflowMiddlewareDelegate, WorkflowMiddlewareDelegate>> Components => _components.ToList();
/// <inheritdoc />
public IServiceProvider ServiceProvider
{
@ -51,6 +54,12 @@ public class WorkflowExecutionPipelineBuilder : IWorkflowExecutionPipelineBuilde
return this;
}
public IWorkflowExecutionPipelineBuilder Replace(int index, Func<WorkflowMiddlewareDelegate, WorkflowMiddlewareDelegate> middleware)
{
_components[index] = middleware;
return this;
}
private T? GetProperty<T>(string key) => Properties.TryGetValue(key, out var value) ? (T?)value : default(T);
private void SetProperty<T>(string key, T value) => Properties[key] = value;
}