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

Fix Join behavior

Browse source
This commit is contained in:
Sipke Schoorstra 2020-10-22 22:38:05 +02:00
parent 9663040c31
commit 7c807c4e71
19 changed files with 128 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
src/core/Elsa.Abstractions/Builders/IActivityBuilder.cs
View file

@ -20,6 +20,7 @@ namespace Elsa.Builders
IOutcomeBuilder When(string outcome);
IActivityBuilder Then(IActivityBuilder targetActivity);
IActivityBuilder WithId(string id);
IActivityBuilder WithName(string name);
Func<ActivityExecutionContext, CancellationToken, ValueTask<IActivity>> BuildActivityAsync();
IWorkflowBlueprint Build();
}

6
src/core/Elsa.Abstractions/Builders/IConnectionBuilder.cs
View file

@ -1,10 +1,12 @@
using System;
namespace Elsa.Builders
{
public interface IConnectionBuilder
{
IWorkflowBuilder WorkflowBuilder { get; }
IActivityBuilder Source { get; }
IActivityBuilder Target{ get; }
Func<IActivityBuilder> Source { get; }
Func<IActivityBuilder> Target{ get; }
string Outcome { get; }
}
}

1
src/core/Elsa.Abstractions/Builders/IOutcomeBuilder.cs
View file

@ -7,6 +7,7 @@ namespace Elsa.Builders
IWorkflowBuilder WorkflowBuilder { get; }
IActivityBuilder Source { get; }
string? Outcome { get; }
IConnectionBuilder Then(string activityName);
IWorkflowBlueprint Build();
}
}

6
src/core/Elsa.Abstractions/Builders/IWorkflowBuilder.cs
View file

@ -16,6 +16,7 @@ namespace Elsa.Builders
bool DeleteCompletedInstances { get; }
bool IsEnabled { get; }
IServiceProvider ServiceProvider { get; }
IReadOnlyCollection<IActivityBuilder> Activities { get; }
IWorkflowBuilder WithId(string value);
IWorkflowBuilder WithName(string value);
IWorkflowBuilder WithDescription(string value);
@ -55,6 +56,11 @@ namespace Elsa.Builders
IActivityBuilder source,
IActivityBuilder target,
string outcome = OutcomeNames.Done);
IConnectionBuilder Connect(
Func<IActivityBuilder> source,
Func<IActivityBuilder> target,
string outcome = OutcomeNames.Done);
IWorkflowBlueprint Build();
IWorkflowBlueprint Build(IWorkflow workflow);

2
src/core/Elsa.Abstractions/Services/Activity.cs
View file

@ -19,7 +19,7 @@ namespace Elsa.Services
public string? DisplayName { get; set; }
public string? Description { get; set; }
public bool PersistWorkflow { get; set; }
public JObject Data { get; set; } = default!;
public JObject Data { get; set; } = new JObject();
public ValueTask<bool> CanExecuteAsync(ActivityExecutionContext context, CancellationToken cancellationToken) =>
OnCanExecuteAsync(context, cancellationToken);

1
src/core/Elsa.Abstractions/Services/Models/IActivityBlueprint.cs
View file

@ -10,7 +10,6 @@ namespace Elsa.Services.Models
string? Name { get; }
public string Type { get; }
public bool PersistWorkflow { get; }
//public JObject Data { get; }
Func<ActivityExecutionContext, CancellationToken, ValueTask<IActivity>> CreateActivityAsync { get; }
}
}

1
src/core/Elsa.Core/Activities/ControlFlow/Complete/FinishBuilderExtensions.cs
View file

@ -1,5 +1,4 @@
using System;
using System.Threading.Tasks;
using Elsa.Builders;
using Elsa.Services.Models;

6
src/core/Elsa.Core/Activities/ControlFlow/Fork/Fork.cs
View file

@ -13,11 +13,9 @@ namespace Elsa.Activities.ControlFlow
Outcomes = new[] { "x => x.state.branches" })]
public class Fork : Activity
{
[ActivityProperty(
Hint = "Enter one or more names representing branches, separated with a comma. Example: Branch 1, Branch 2"
)]
[ActivityProperty(Hint = "Enter one or more names representing branches, separated with a comma. Example: Branch 1, Branch 2")]
public HashSet<string> Branches { get; set; } = new HashSet<string>();
protected override IActivityExecutionResult OnExecute() => Done(Branches);
protected override IActivityExecutionResult OnExecute() => Combine(Done(), Outcomes(Branches));
}
}

7
src/core/Elsa.Core/Activities/ControlFlow/Fork/ForkBuilder.cs
View file

@ -6,17 +6,16 @@ namespace Elsa.Activities.ControlFlow
public class ForkBuilder
{
private readonly IActivityBuilder _activityBuilder;
private readonly Fork _fork;
public ForkBuilder(IActivityBuilder activityBuilder)
{
_activityBuilder = activityBuilder;
//_fork = (Fork)activityBuilder.Activity;
}
public IOutcomeBuilder When(string branch)
{
_fork.Branches.Add(branch);
//_activityBuilder.
//_fork.Branches.Add(branch);
return _activityBuilder.When(branch);
}
}

47
src/core/Elsa.Core/Activities/ControlFlow/Join/Join.cs
View file

@ -33,38 +33,34 @@ namespace Elsa.Activities.ControlFlow
WaitAny
}
[ActivityProperty(
Type = ActivityPropertyTypes.Select,
Hint = "Either 'WaitAll' or 'WaitAny'")
]
[ActivityProperty(Type = ActivityPropertyTypes.Select, Hint = "Either 'WaitAll' or 'WaitAny'")]
[SelectOptions("WaitAll", "WaitAny")]
public JoinMode Mode { get; set; }
public IReadOnlyCollection<string> InboundTransitions { get; set; }
public IReadOnlyCollection<string> InboundTransitions
{
get => GetState<IReadOnlyCollection<string>>(() => new List<string>());
set => SetState(value);
}
protected override IActivityExecutionResult OnExecute(ActivityExecutionContext context)
{
var workflowExecutionContext = context.WorkflowExecutionContext;
var recordedInboundTransitions = InboundTransitions;
var inboundConnections = workflowExecutionContext.GetInboundConnections(Id);
var done = false;
switch (Mode)
var done = Mode switch
{
case JoinMode.WaitAll:
done = inboundConnections.All(x => recordedInboundTransitions.Contains(GetTransitionKey(x)));
break;
case JoinMode.WaitAny:
done = inboundConnections.Any(x => recordedInboundTransitions.Contains(GetTransitionKey(x)));
break;
}
JoinMode.WaitAll => inboundConnections.All(x => recordedInboundTransitions.Contains(GetTransitionKey(x))),
JoinMode.WaitAny => inboundConnections.Any(x => recordedInboundTransitions.Contains(GetTransitionKey(x))),
_ => false
};
if (done)
{
// Remove any inbound blocking activities.
var ancestorActivityIds = workflowExecutionContext.GetInboundActivityPath(Id).ToList();
var blockingActivities =
workflowExecutionContext.WorkflowInstance.BlockingActivities.Where(x => ancestorActivityIds.Contains(x.ActivityId)).ToList();
var blockingActivities = workflowExecutionContext.WorkflowInstance.BlockingActivities.Where(x => ancestorActivityIds.Contains(x.ActivityId)).ToList();
foreach (var blockingActivity in blockingActivities)
workflowExecutionContext.WorkflowInstance.BlockingActivities.Remove(blockingActivity);
@ -78,7 +74,7 @@ namespace Elsa.Activities.ControlFlow
return Done();
}
private void RecordInboundTransitions(WorkflowExecutionContext workflowExecutionContext, IActivityBlueprint activity)
private async Task RecordInboundTransitionsAsync(WorkflowExecutionContext workflowExecutionContext, IActivityBlueprint activity, CancellationToken cancellationToken)
{
// Get outbound connections of the executing activity.
var outboundConnections = workflowExecutionContext.GetOutboundConnections(activity.Id);
@ -91,12 +87,15 @@ namespace Elsa.Activities.ControlFlow
select connection;
var inboundConnections = inboundTransitionsQuery.ToList();
var joinBlueprint = inboundConnections.FirstOrDefault()?.Target.Activity;
var joinActivity = joinBlueprint != null ? (Join)await joinBlueprint.CreateActivityAsync(new ActivityExecutionContext(workflowExecutionContext, workflowExecutionContext.ServiceProvider, joinBlueprint), cancellationToken) : default;
// For each inbound connection, record the transition.
foreach (var inboundConnection in inboundConnections)
{
var inboundTransitions = InboundTransitions;
InboundTransitions = inboundTransitions
var inboundTransitions = joinActivity!.InboundTransitions;
joinActivity!.InboundTransitions = inboundTransitions
.Union(new[] { GetTransitionKey(inboundConnection) })
.Distinct()
.ToList();
@ -111,11 +110,9 @@ namespace Elsa.Activities.ControlFlow
return $"@{sourceActivityId}_{sourceOutcomeName}";
}
public Task Handle(ActivityExecuted notification, CancellationToken cancellationToken)
public async Task Handle(ActivityExecuted notification, CancellationToken cancellationToken)
{
RecordInboundTransitions(notification.WorkflowExecutionContext, notification.Activity);
return Task.CompletedTask;
await RecordInboundTransitionsAsync(notification.WorkflowExecutionContext, notification.Activity, cancellationToken);
}
}
}

6
src/core/Elsa.Core/Builders/ActivityBuilder.cs
View file

@ -53,6 +53,12 @@ namespace Elsa.Builders
return this;
}
public IActivityBuilder WithName(string name)
{
Name = name;
return this;
}
public Func<ActivityExecutionContext, CancellationToken, ValueTask<IActivity>> BuildActivityAsync() =>
async (context, cancellationToken) =>
{

10
src/core/Elsa.Core/Builders/ConnectionBuilder.cs
View file

@ -1,13 +1,15 @@
using System;
namespace Elsa.Builders
{
public class ConnectionBuilder : IConnectionBuilder
{
public IWorkflowBuilder WorkflowBuilder { get; }
public IActivityBuilder Source { get; }
public IActivityBuilder Target{ get; }
public Func<IActivityBuilder> Source { get; }
public Func<IActivityBuilder> Target { get; }
public string Outcome { get; }
public ConnectionBuilder(IWorkflowBuilder workflowBuilder, IActivityBuilder source, IActivityBuilder target, string outcome = OutcomeNames.Done)
public ConnectionBuilder(IWorkflowBuilder workflowBuilder, Func<IActivityBuilder> source, Func<IActivityBuilder> target, string outcome = OutcomeNames.Done)
{
Source = source;
Target = target;

9
src/core/Elsa.Core/Builders/OutcomeBuilder.cs
View file

@ -1,4 +1,5 @@
using System;
using System.Linq;
using Elsa.Services;
using Elsa.Services.Models;
@ -25,6 +26,14 @@ namespace Elsa.Builders
public IActivityBuilder Then<T>(Action<IActivityBuilder>? branch = default)
where T : class, IActivity => Then(WorkflowBuilder.Add<T>(branch));
public IConnectionBuilder Then(string activityName)
{
return WorkflowBuilder.Connect(
() => Source,
() => WorkflowBuilder.Activities.First(x => x.Name == activityName),
Outcome);
}
private IActivityBuilder Then(IActivityBuilder activityBuilder, Action<IActivityBuilder>? branch = default)
{
branch?.Invoke(activityBuilder);

20
src/core/Elsa.Core/Builders/WorkflowBuilder.cs
View file

@ -36,6 +36,7 @@ namespace Elsa.Builders
public WorkflowPersistenceBehavior PersistenceBehavior { get; private set; }
public bool DeleteCompletedInstances { get; private set; }
public bool IsEnabled { get; private set; }
public IReadOnlyCollection<IActivityBuilder> Activities => _activityBuilders.ToList().AsReadOnly();
public IWorkflowBuilder WithId(string value)
{
@ -144,7 +145,7 @@ namespace Elsa.Builders
IDictionary<string, IActivityPropertyValueProvider>? propertyValueProviders = default)
where T : class, IActivity
{
var activityBuilder = new ActivityBuilder(typeof(T), this, propertyValueProviders);
var activityBuilder = new ActivityBuilder(typeof(T), this, propertyValueProviders);
return Add(activityBuilder);
}
@ -160,6 +161,13 @@ namespace Elsa.Builders
public IConnectionBuilder Connect(
IActivityBuilder source,
IActivityBuilder target,
string outcome = OutcomeNames.Done) =>
Connect(() => source, () => target, outcome);
public IConnectionBuilder Connect(
Func<IActivityBuilder> source,
Func<IActivityBuilder> target,
string outcome = OutcomeNames.Done)
{
var connectionBuilder = new ConnectionBuilder(this, source, target, outcome);
@ -170,10 +178,12 @@ namespace Elsa.Builders
public IActivityBuilder Then<T>(
Action<ISetupActivity<T>>? setup = default,
Action<IActivityBuilder>? branch = default)
where T : class, IActivity => StartWith(setup, branch);
where T : class, IActivity =>
StartWith(setup, branch);
public IActivityBuilder Then<T>(Action<IActivityBuilder>? branch = default)
where T : class, IActivity => StartWith<T>(branch);
where T : class, IActivity =>
StartWith<T>(branch);
public IWorkflowBlueprint Build(IWorkflow workflow)
{
@ -197,7 +207,7 @@ namespace Elsa.Builders
var activityBlueprintDictionary = activityBlueprints.ToDictionary(x => x.Id);
var connections = _connectionBuilders
.Select(x => new Connection(activityBlueprintDictionary[x.Source.ActivityId], activityBlueprintDictionary[x.Target.ActivityId], x.Outcome))
.Select(x => new Connection(activityBlueprintDictionary[x.Source().ActivityId], activityBlueprintDictionary[x.Target().ActivityId], x.Outcome))
.ToList();
var activityPropertyValueProviders = _activityBuilders
@ -218,7 +228,7 @@ namespace Elsa.Builders
activityBlueprints,
connections,
new ActivityPropertyProviders(activityPropertyValueProviders));
return workflow;
}

1
src/core/Elsa.Core/Services/WorkflowRegistry.cs
View file

@ -5,7 +5,6 @@ using System.Threading;
using System.Threading.Tasks;
using Elsa.Models;
using Elsa.Services.Models;
using Open.Linq.AsyncExtensions;
namespace Elsa.Services
{

4
test/integration/Elsa.Core.IntegrationTests/BasicWorkflowUnitTests.cs → test/integration/Elsa.Core.IntegrationTests/BasicWorkflowTests.cs
View file

@ -7,9 +7,9 @@ using Xunit.Abstractions;
namespace Elsa.Core.IntegrationTests
{
public class BasicWorkflowUnitTests : WorkflowsUnitTestBase
public class BasicWorkflowTests : WorkflowsUnitTestBase
{
public BasicWorkflowUnitTests(ITestOutputHelper testOutputHelper) : base(testOutputHelper)
public BasicWorkflowTests(ITestOutputHelper testOutputHelper) : base(testOutputHelper)
{
}

4
test/integration/Elsa.Core.IntegrationTests/ForLoopWorkflowUnitTests.cs → test/integration/Elsa.Core.IntegrationTests/ForLoopWorkflowTests.cs
View file

@ -7,9 +7,9 @@ using Xunit.Abstractions;
namespace Elsa.Core.IntegrationTests
{
public class ForLoopWorkflowUnitTests : WorkflowsUnitTestBase
public class ForLoopWorkflowTests : WorkflowsUnitTestBase
{
public ForLoopWorkflowUnitTests(ITestOutputHelper testOutputHelper) : base(testOutputHelper)
public ForLoopWorkflowTests(ITestOutputHelper testOutputHelper) : base(testOutputHelper)
{
}

24
test/integration/Elsa.Core.IntegrationTests/ForkJoinWorkflowTests.cs Normal file
View file

@ -0,0 +1,24 @@
using System.Threading.Tasks;
using Elsa.Core.IntegrationTests.Workflows;
using Elsa.Models;
using Elsa.Testing.Shared.Helpers;
using Xunit;
using Xunit.Abstractions;
namespace Elsa.Core.IntegrationTests
{
public class ForkJoinWorkflowTests : WorkflowsUnitTestBase
{
public ForkJoinWorkflowTests(ITestOutputHelper testOutputHelper) : base(testOutputHelper)
{
}
[Fact(DisplayName = "Runs fork and join workflow.")]
public async Task Test01()
{
var workflowInstance = await WorkflowHost.RunWorkflowAsync<ForkJoinWaitAllWorkflow>();
Assert.Equal(WorkflowStatus.Completed, workflowInstance.Status);
}
}
}

24
test/integration/Elsa.Core.IntegrationTests/Workflows/ForkJoinWaitAllWorkflow.cs Normal file
View file

@ -0,0 +1,24 @@
using System.Collections.Generic;
using Elsa.Activities.Console;
using Elsa.Activities.ControlFlow;
using Elsa.Builders;
namespace Elsa.Core.IntegrationTests.Workflows
{
public class ForkJoinWaitAllWorkflow : IWorkflow
{
public void Build(IWorkflowBuilder workflow)
{
workflow.StartWith<Fork>(
activity => activity.Set(x => x.Branches, new HashSet<string>(new[] { "Branch 1", "Branch 2", "Branch 3" })),
fork =>
{
fork.When("Branch 1").WriteLine("Branch 1 executed").Then("Join");
fork.When("Branch 2").WriteLine("Branch 2 executed").Then("Join");
fork.When("Branch 3").WriteLine("Branch 3 executed").Then("Join");
})
.Add<Join>(join => join.Set(x => x.Mode, Join.JoinMode.WaitAll)).WithName("Join")
.WriteLine("Finished");
}
}
}