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

Merge branch 'elsa-workflows:main' into main

Browse source
This commit is contained in:
Sergei Khlebnikov 2024-12-16 22:58:04 +03:00 committed by GitHub
parent 311b792f9b 5d92c5760f
commit 8583da9b5f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 65 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
src/modules/Elsa.Retention/CleanupStrategies/DeleteWorkflowInstanceStrategy.cs Normal file
View file

@ -0,0 +1,20 @@
using Elsa.Retention.Contracts;
using Elsa.Workflows.Management;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Filters;
namespace Elsa.Retention.CleanupStrategies;
/// <summary>
/// Deletes the workflow instance.
/// </summary>
public class DeleteWorkflowInstanceStrategy(IWorkflowInstanceStore store) : IDeletionCleanupStrategy<WorkflowInstance>
{
public async Task Cleanup(ICollection<WorkflowInstance> collection)
{
await store.DeleteAsync(new WorkflowInstanceFilter
{
Ids = collection.Select(x => x.Id).ToArray()
});
}
}

2
src/modules/Elsa.Retention/Feature/RetentionFeature.cs
View file

@ -7,6 +7,7 @@ using Elsa.Retention.Contracts;
using Elsa.Retention.Extensions;
using Elsa.Retention.Jobs;
using Elsa.Retention.Options;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Runtime.Entities;
using Microsoft.Extensions.DependencyInjection;
@ -44,6 +45,7 @@ public class RetentionFeature : FeatureBase
Services.AddScoped<IDeletionCleanupStrategy<StoredBookmark>, DeleteBookmarkStrategy>();
Services.AddScoped<IDeletionCleanupStrategy<ActivityExecutionRecord>, DeleteActivityExecutionRecordStrategy>();
Services.AddScoped<IDeletionCleanupStrategy<WorkflowExecutionLogRecord>, DeleteWorkflowExecutionRecordStrategy>();
Services.AddScoped<IDeletionCleanupStrategy<WorkflowInstance>, DeleteWorkflowInstanceStrategy>();
Services.AddScoped<IRelatedEntityCollector, BookmarkCollector>();
Services.AddScoped<IRelatedEntityCollector, ActivityExecutionRecordCollector>();

17
src/modules/Elsa.Retention/Jobs/CleanupJob.cs
View file

@ -31,6 +31,7 @@ public class CleanupJob(
/// <param name="cancellationToken"></param>
public async Task ExecuteAsync(CancellationToken cancellationToken = default)
{
Console.WriteLine(DateTime.Now.ToLongTimeString());
var collectors = GetServices(typeof(IRelatedEntityCollector), typeof(IRelatedEntityCollector<>));
var deletedWorkflowInstances = 0L;
@ -51,6 +52,10 @@ public class CleanupJob(
foreach (var collectorService in collectors)
{
var cleanupStrategyConcreteType = policy.CleanupStrategy.MakeGenericType(collectorService.Key);
if(cleanupStrategyConcreteType == typeof(WorkflowInstance))
continue;
var collector = collectorService.Value as IRelatedEntityCollector;
var cleanupService = serviceProvider.GetService(cleanupStrategyConcreteType) as ICleanupStrategy;
@ -71,11 +76,15 @@ public class CleanupJob(
await cleanupService.Cleanup(entities);
}
}
var cleanupWorkflowInstances = policy.CleanupStrategy.MakeGenericType(typeof(WorkflowInstance));
var workflowInstanceCleaner = serviceProvider.GetService(cleanupWorkflowInstances) as ICleanupStrategy<WorkflowInstance>;
deletedWorkflowInstances += await workflowInstanceStore.DeleteAsync(new WorkflowInstanceFilter
{
Ids = page.Items.Select(x => x.Id).ToArray()
}, cancellationToken);
if (workflowInstanceCleaner == null)
throw new Exception($"{policy.CleanupStrategy} has no strategy to clean WorkflowInstances");
await workflowInstanceCleaner.Cleanup(page.Items);
deletedWorkflowInstances += page.Items.Count;
if (page.TotalCount <= page.Items.Count + pageArgs.Offset)
{

33
src/modules/Elsa.Workflows.Core/Activities/Fork.cs
View file

@ -1,7 +1,6 @@
using System.Collections.Immutable;
using System.ComponentModel;
using System.Runtime.CompilerServices;
using Elsa.Extensions;
using Elsa.Workflows.Attributes;
using Elsa.Workflows.Signals;
using Elsa.Workflows.UIHints;
@ -18,7 +17,7 @@ namespace Elsa.Workflows.Activities;
public class Fork : Activity
{
/// <inheritdoc />
public Fork([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
public Fork([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)
{
// Handle break signals directly instead of using the BreakBehavior. The behavior stops propagation of the signal, which is not what we want.
OnSignalReceived<BreakSignal>(OnBreakSignalReceived);
@ -48,13 +47,7 @@ public class Fork : Activity
if (isBreaking)
{
// Remove all bookmarks from other branches.
RemoveBookmarks(targetContext);
// Signal activity completion.
await CompleteAsync(targetContext);
// Exit.
return;
}
@ -75,37 +68,15 @@ public class Fork : Activity
switch (joinMode)
{
case ForkJoinMode.WaitAny:
{
// Remove all bookmarks from other branches.
RemoveBookmarks(targetContext);
// Signal activity completion.
await CompleteAsync(targetContext);
}
break;
case ForkJoinMode.WaitAll:
{
var allSet = allChildActivityIds.All(x => completedActivityIds.Contains(x));
if (allSet)
// Signal activity completion.
await CompleteAsync(targetContext);
}
if (allSet) await CompleteAsync(targetContext);
break;
}
}
private void RemoveBookmarks(ActivityExecutionContext context)
{
// Find all descendants for each branch and remove them as well as any associated bookmarks.
var workflowExecutionContext = context.WorkflowExecutionContext;
var forkNode = context.ActivityNode;
var branchNodes = forkNode.Children;
var branchDescendantActivityIds = branchNodes.SelectMany(x => x.Flatten()).Select(x => x.Activity.Id).ToHashSet();
workflowExecutionContext.Bookmarks.RemoveWhere(x => branchDescendantActivityIds.Contains(x.ActivityId));
}
private void OnBreakSignalReceived(BreakSignal signal, SignalContext signalContext)
{
signalContext.ReceiverActivityExecutionContext.SetIsBreaking();

12
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs
View file

@ -23,13 +23,19 @@ public partial class ActivityExecutionContext
ClearBookmarks();
ClearCompletionCallbacks();
WorkflowExecutionContext.Bookmarks.RemoveWhere(x => x.ActivityNodeId == NodeId);
// Add an execution log entry.
AddExecutionLogEntry("Canceled", payload: JournalData);
await this.SendSignalAsync(new CancelSignal());
await CancelChildActivitiesAsync();
// ReSharper disable once MethodSupportsCancellation
await _publisher.SendAsync(new ActivityCancelled(this));
}
private async Task CancelChildActivitiesAsync()
{
var childContexts = WorkflowExecutionContext.ActivityExecutionContexts.Where(x => x.ParentActivityExecutionContext == this && x.CanCancelActivity()).ToList();
foreach (var childContext in childContexts)
await childContext.CancelActivityAsync();
}
}

28
test/component/Elsa.Workflows.ComponentTests/Scenarios/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs
View file

@ -21,20 +21,20 @@ public class BulkDispatchWorkflowsTests : AppComponentTest
_signalManager = Scope.ServiceProvider.GetRequiredService<SignalManager>();
}
/// <summary>
/// Dispatches and waits for child workflows to complete.
/// </summary>
[Fact]
public async Task DispatchAndWaitWorkflow_ShouldWaitForChildWorkflowToComplete()
{
var workflowClient = await _workflowRuntime.CreateClientAsync();
await workflowClient.CreateInstanceAsync(new CreateWorkflowInstanceRequest
{
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(GreetEmployeesWorkflow.DefinitionId, VersionOptions.Published)
});
await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty);
await _signalManager.WaitAsync<string>("Completed");
}
// /// <summary>
// /// Dispatches and waits for child workflows to complete.
// /// </summary>
// [Fact(Skip = "This test is flaky and needs to be fixed.")]
// public async Task DispatchAndWaitWorkflow_ShouldWaitForChildWorkflowToComplete()
// {
// var workflowClient = await _workflowRuntime.CreateClientAsync();
// await workflowClient.CreateInstanceAsync(new CreateWorkflowInstanceRequest
// {
// WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(GreetEmployeesWorkflow.DefinitionId, VersionOptions.Published)
// });
// await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty);
// await _signalManager.WaitAsync<string>("Completed");
// }
/// <summary>
/// Individual items are sent as input to child workflows.

31
test/component/Elsa.Workflows.ComponentTests/Scenarios/DispatchWorkflows/DispatchWorkflowsTests.cs
View file

@ -13,22 +13,16 @@ namespace Elsa.Workflows.ComponentTests.Scenarios.DispatchWorkflows;
public class DispatchWorkflowsTests : AppComponentTest
{
private readonly WorkflowEvents _workflowEvents;
private readonly SignalManager _signalManager;
private readonly IWorkflowRuntime _workflowRuntime;
private readonly object _childWorkflowCompletedSignal = new();
private readonly object _parentWorkflowCompletedSignal = new();
public DispatchWorkflowsTests(App app) : base(app)
{
_workflowRuntime = Scope.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
_workflowEvents = Scope.ServiceProvider.GetRequiredService<WorkflowEvents>();
_signalManager = Scope.ServiceProvider.GetRequiredService<SignalManager>();
_workflowEvents.WorkflowInstanceSaved += OnWorkflowInstanceSaved;
}
[Fact]
[Fact (Skip = "This test is flaky and needs to be fixed.")]
public async Task DispatchAndWaitWorkflow_ShouldWaitForChildWorkflowToComplete()
{
var workflowClient = await _workflowRuntime.CreateClientAsync();
@ -37,27 +31,6 @@ public class DispatchWorkflowsTests : AppComponentTest
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(DispatchAndWaitWorkflow.DefinitionId, VersionOptions.Published)
});
await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty);
var childWorkflowInstanceArgs = await _signalManager.WaitAsync<WorkflowInstanceSavedEventArgs>(_childWorkflowCompletedSignal);
var parentWorkflowInstanceArgs = await _signalManager.WaitAsync<WorkflowInstanceSavedEventArgs>(_parentWorkflowCompletedSignal);
Assert.Equal(WorkflowStatus.Finished, childWorkflowInstanceArgs.WorkflowInstance.Status);
Assert.Equal(WorkflowStatus.Finished, parentWorkflowInstanceArgs.WorkflowInstance.Status);
}
private void OnWorkflowInstanceSaved(object? sender, WorkflowInstanceSavedEventArgs e)
{
if (e.WorkflowInstance.Status != WorkflowStatus.Finished)
return;
if (e.WorkflowInstance.DefinitionId == ChildWorkflow.DefinitionId)
_signalManager.Trigger(_childWorkflowCompletedSignal, e);
if (e.WorkflowInstance.DefinitionId == DispatchAndWaitWorkflow.DefinitionId)
_signalManager.Trigger(_parentWorkflowCompletedSignal, e);
}
protected override void OnDispose()
{
_workflowEvents.WorkflowInstanceSaved -= OnWorkflowInstanceSaved;
await _signalManager.WaitAsync<string>("Completed");
}
}

4
test/component/Elsa.Workflows.ComponentTests/Scenarios/DispatchWorkflows/Workflows/DispatchAndWaitWorkflow.cs
View file

@ -1,3 +1,4 @@
using Elsa.Testing.Shared.Activities;
using Elsa.Workflows.Activities;
using Elsa.Workflows.Runtime.Activities;
@ -17,7 +18,8 @@ public class DispatchAndWaitWorkflow : WorkflowBase
{
WorkflowDefinitionId = new(ChildWorkflow.DefinitionId),
WaitForCompletion = new (true)
}
},
new TriggerSignal("Completed")
}
};
}