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

Fix and simplify composite activity + while/for/foreach handling

Browse source
This commit is contained in:
Sipke Schoorstra 2021-01-01 17:58:08 +01:00
parent 10fe049c10
commit dcbf14c144
30 changed files with 171 additions and 258 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
src/activities/Elsa.Activities.Timers.Quartz/Services/QuartzWorkflowScheduler.cs
View file

@ -63,8 +63,8 @@ namespace Elsa.Activities.Timers.Quartz.Services
var existingTrigger = await scheduler.GetTrigger(trigger, cancellationToken);
// if (existingTrigger != null)
// await scheduler.UnscheduleJob(existingTrigger.Key, cancellationToken);
if (existingTrigger != null)
await scheduler.UnscheduleJob(existingTrigger.Key, cancellationToken);
}
private async Task ScheduleJob(ITrigger trigger, CancellationToken cancellationToken)
@ -76,8 +76,8 @@ namespace Elsa.Activities.Timers.Quartz.Services
var scheduler = await _schedulerFactory.GetScheduler(cancellationToken);
var existingTrigger = await scheduler.GetTrigger(trigger.Key, cancellationToken);
// if (existingTrigger != null)
// await scheduler.UnscheduleJob(existingTrigger.Key, cancellationToken);
if (existingTrigger != null)
await scheduler.UnscheduleJob(existingTrigger.Key, cancellationToken);
await scheduler.ScheduleJob(trigger, cancellationToken);
}
@ -104,7 +104,7 @@ namespace Elsa.Activities.Timers.Quartz.Services
private TriggerKey CreateTriggerKey(string? tenantId, string workflowDefinitionId, string? workflowInstanceId, string activityId)
{
var groupName = $"tenant:{tenantId ?? "default"}-workflow-instance:{workflowInstanceId ?? workflowDefinitionId}";
return new TriggerKey($"activity:{activityId}:{Guid.NewGuid()}", groupName);
return new TriggerKey($"activity:{activityId}", groupName);
}
}
}

32
src/activities/Elsa.Activities.Timers.Quartz/Services/WorkflowRunnerQueue.cs
View file

@ -9,6 +9,7 @@ using Microsoft.Extensions.Logging;
namespace Elsa.Activities.Timers.Quartz.Services
{
// TODO: Consider turning this into a global service to allow background, sequential execution of a given workflow instance (but allow for parallel execution of workflow instances with different definitions). Executing the same workflow instances sequentially prevents loss of data during update concurrency conflicts.
public class WorkflowRunnerQueue
{
private readonly IBackgroundWorker _backgroundWorker;
@ -21,7 +22,7 @@ namespace Elsa.Activities.Timers.Quartz.Services
_serviceProvider = serviceProvider;
_logger = logger;
}
public async Task Enqueue(string workflowInstanceId, string activityId, CancellationToken cancellationToken = default)
{
await _backgroundWorker.ScheduleTask(async () => await RunWorkflowAsync(workflowInstanceId, activityId, cancellationToken), cancellationToken);
@ -33,19 +34,34 @@ namespace Elsa.Activities.Timers.Quartz.Services
var store = scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
var workflowInstance = await store.FindByIdAsync(workflowInstanceId, cancellationToken);
if (workflowInstance == null)
{
_logger.LogError("Could not run Workflow instance with ID {WorkflowInstanceId} because it is not in the database", workflowInstanceId);
if (!ValidatePreconditions(workflowInstanceId, workflowInstance))
return;
}
//await context.Scheduler.UnscheduleJob(context.Trigger.Key, cancellationToken);
var workflowDefinitionId = workflowInstance.DefinitionId;
_logger.LogDebug("Running {WorkflowInstanceId} with status {WorkflowStatus}.", workflowInstance!.WorkflowStatus);
var workflowDefinitionId = workflowInstance!.DefinitionId;
var tenantId = workflowInstance.TenantId;
var workflowRegistry = scope.ServiceProvider.GetRequiredService<IWorkflowRegistry>();
var workflowBlueprint = (await workflowRegistry.GetWorkflowAsync(workflowDefinitionId, tenantId, VersionOptions.SpecificVersion(workflowInstance.Version), cancellationToken))!;
var workflowRunner = scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
await workflowRunner.RunWorkflowAsync(workflowBlueprint, workflowInstance!, activityId, cancellationToken: cancellationToken);
}
private bool ValidatePreconditions(string? workflowInstanceId, WorkflowInstance? workflowInstance)
{
if (workflowInstance == null)
{
_logger.LogError("Could not run workflow instance with ID {WorkflowInstanceId} because it does not exist.", workflowInstanceId);
return false;
}
if (workflowInstance.WorkflowStatus != WorkflowStatus.Suspended)
{
_logger.LogWarning("Could not run workflow instance with ID {WorkflowInstanceId} because it has a status other than Suspended. Its actual status is {WorkflowStatus}", workflowInstanceId, workflowInstance.WorkflowStatus);
return false;
}
return true;
}
}
}

1
src/activities/Elsa.Activities.Timers/ActivityResults/ScheduleWorkflowResult.cs
View file

@ -1,4 +1,3 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.Timers.Services;

3
src/clients/Elsa.Client/Models/WorkflowInstance.cs
View file

@ -15,7 +15,6 @@ namespace Elsa.Client.Models
Variables = new Variables();
Activities = new List<ActivityInstance>();
ScheduledActivities = new Stack<ScheduledActivity>();
PostScheduledActivities = new Stack<ScheduledActivity>();
}
[DataMember(Order = 1)] public string Id { get; set; } = default!;
@ -42,6 +41,6 @@ namespace Elsa.Client.Models
[DataMember(Order = 16)] public WorkflowFault? Fault { get; set; }
[DataMember(Order = 17)] public Stack<ScheduledActivity> ScheduledActivities { get; set; }
[DataMember(Order = 18)] public Stack<ScheduledActivity> PostScheduledActivities { get; set; }
[DataMember(Order = 18)]public Stack<string> ParentActivities { get; set; }
}
}

20
src/core/Elsa.Abstractions/ActivityResults/PostScheduleActivitiesResult.cs
View file

@ -1,20 +0,0 @@
using System.Collections.Generic;
using System.Linq;
using Elsa.Models;
using Elsa.Services.Models;
namespace Elsa.ActivityResults
{
public class PostScheduleActivitiesResult : ActivityExecutionResult
{
public PostScheduleActivitiesResult(IEnumerable<string> activityIds, object? input = default) =>
Activities = activityIds.Select(x => new ScheduledActivity(x, input));
public PostScheduleActivitiesResult(IEnumerable<ScheduledActivity> activities) => Activities = activities;
public IEnumerable<ScheduledActivity> Activities { get; }
protected override void Execute(ActivityExecutionContext activityExecutionContext) =>
activityExecutionContext.WorkflowExecutionContext.PostScheduleActivities(Activities);
}
}

4
src/core/Elsa.Abstractions/Extensions/CompositeActivityBlueprintExtensions.cs
View file

@ -11,8 +11,7 @@ namespace Elsa
{
public static IEnumerable<IActivityBlueprint> GetStartActivities(this ICompositeActivityBlueprint workflowBlueprint)
{
var targetActivityIds = workflowBlueprint.Connections.Select(x => x.Target.Activity.Id).Distinct()
.ToLookup(x => x);
var targetActivityIds = workflowBlueprint.Connections.Select(x => x.Target.Activity?.Id).Distinct().ToLookup(x => x);
var query =
from activity in workflowBlueprint.Activities
@ -25,6 +24,7 @@ namespace Elsa
public static IEnumerable<IActivityBlueprint> GetStartActivities(this ICompositeActivityBlueprint workflowBlueprint, string activityType) => workflowBlueprint.GetStartActivities().Where(x => x.Type == activityType);
public static IEnumerable<IActivityBlueprint> GetStartActivities(this ICompositeActivityBlueprint workflowBlueprint, Type activityType) => workflowBlueprint.GetStartActivities(activityType.Name);
public static IEnumerable<IActivityBlueprint> GetStartActivities<T>(this ICompositeActivityBlueprint workflowBlueprint) where T : IActivity => workflowBlueprint.GetStartActivities(typeof(T));
public static IEnumerable<IActivityBlueprint> GetEndActivities(this ICompositeActivityBlueprint workflowBlueprint) => workflowBlueprint.Activities.Where(x => !workflowBlueprint.GetOutboundConnections(x.Id).Any());
public static IActivityBlueprint? GetActivity(this ICompositeActivityBlueprint workflowBlueprint, string id) => workflowBlueprint.Activities.FirstOrDefault(x => x.Id == id);
public static IEnumerable<IActivityBlueprint> GetActivities(this ICompositeActivityBlueprint workflowBlueprint, IEnumerable<string> ids) => workflowBlueprint.Activities.Where(x => ids.Contains(x.Id));

2
src/core/Elsa.Abstractions/Models/WorkflowInstance.cs
View file

@ -13,7 +13,6 @@ namespace Elsa.Models
Variables = new Variables();
Activities = new List<ActivityInstance>();
ScheduledActivities = new Stack<ScheduledActivity>();
PostScheduledActivities = new Stack<ScheduledActivity>();
ParentActivities = new Stack<string>();
}
@ -43,7 +42,6 @@ namespace Elsa.Models
public WorkflowFault? Fault { get; set; }
public Stack<ScheduledActivity> ScheduledActivities { get; set; }
public Stack<ScheduledActivity> PostScheduledActivities { get; set; }
public Stack<string> ParentActivities { get; set; }
}
}

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

@ -45,7 +45,6 @@ namespace Elsa.Services
protected ScheduleActivitiesResult Schedule(IEnumerable<string> activityIds, object? input) => new(activityIds, input);
protected ScheduleActivitiesResult Schedule(string activityId, object? input) => Schedule(new[] { activityId }, input);
protected ScheduleActivitiesResult Schedule(IEnumerable<ScheduledActivity> activities) => new(activities);
protected PostScheduleActivitiesResult PostSchedule(params string[] activityIds) => new(activityIds);
protected CombinedResult Combine(IEnumerable<IActivityExecutionResult> results) => new(results);
protected CombinedResult Combine(params IActivityExecutionResult[] results) => new(results);
protected FaultResult Fault(LocalizedString message) => new(message);

35
src/core/Elsa.Abstractions/Services/CompositeActivity.cs
View file

@ -1,5 +1,4 @@
using System.Linq;
using Elsa.ActivityResults;
using Elsa.ActivityResults;
using Elsa.Builders;
using Elsa.Services.Models;
@ -7,6 +6,8 @@ namespace Elsa.Services
{
public class CompositeActivity : Activity
{
internal const string Enter = "Enter";
public virtual void Build(ICompositeActivityBuilder activity)
{
}
@ -19,34 +20,14 @@ namespace Elsa.Services
protected override IActivityExecutionResult OnExecute(ActivityExecutionContext context)
{
if (IsScheduled)
if (!IsScheduled)
{
if (HasPendingChildren(context))
return PostSchedule(Id);
context.WorkflowExecutionContext.WorkflowInstance.ParentActivities.Pop();
IsScheduled = false;
return Complete(context);
IsScheduled = true;
return Outcome(Enter);
}
var compositeActivityBlueprint = (ICompositeActivityBlueprint) context.ActivityBlueprint;
var startActivities = compositeActivityBlueprint.GetStartActivities().Select(x => x.Id).ToList();
context.WorkflowExecutionContext.WorkflowInstance.ParentActivities.Push(Id);
context.WorkflowExecutionContext.PostScheduleActivity(Id);
IsScheduled = true;
return Schedule(startActivities, context.Input);
}
protected virtual IActivityExecutionResult Complete(ActivityExecutionContext context) => Done();
private static bool HasPendingChildren(ActivityExecutionContext context)
{
var children = ((CompositeActivityBlueprint) context.ActivityBlueprint).Activities.Select(x => x.Id).ToList();
var workflowInstance = context.WorkflowExecutionContext.WorkflowInstance;
//var hasPendingPostScheduledChildren = workflowInstance.PostScheduledActivities.Any(x => children.Contains(x.ActivityId));
var hasPendingScheduledChildren = workflowInstance.ScheduledActivities.Any(x => children.Contains(x.ActivityId));
//return hasPendingPostScheduledChildren || hasPendingScheduledChildren;
return hasPendingScheduledChildren;
IsScheduled = false;
return Done();
}
}
}

15
src/core/Elsa.Abstractions/Services/Models/WorkflowExecutionContext.cs
View file

@ -33,7 +33,6 @@ namespace Elsa.Services.Models
public JsonSerializer Serializer { get; }
public object? Input { get; }
public bool HasScheduledActivities => WorkflowInstance.ScheduledActivities.Any();
public bool HasPostScheduledActivities => WorkflowInstance.PostScheduledActivities.Any();
public IWorkflowFault? WorkflowFault { get; private set; }
public bool IsFirstPass { get; private set; }
public bool ContextHasChanged { get; set; }
@ -55,16 +54,8 @@ namespace Elsa.Services.Models
ScheduleActivity(activity);
}
public void PostScheduleActivities(IEnumerable<ScheduledActivity> activities)
{
foreach (var activity in activities)
PostScheduleActivity(activity);
}
public void ScheduleActivity(string activityId, object? input = default) => ScheduleActivity(new ScheduledActivity(activityId, input));
public void ScheduleActivity(ScheduledActivity activity) => WorkflowInstance.ScheduledActivities.Push(activity);
public void PostScheduleActivity(string activityId, object? input = default) => PostScheduleActivity(new ScheduledActivity(activityId, input));
public void PostScheduleActivity(ScheduledActivity activity) => WorkflowInstance.PostScheduledActivities.Push(activity);
public ScheduledActivity PopScheduledActivity() => WorkflowInstance.ScheduledActivities.Pop();
public ScheduledActivity PeekScheduledActivity() => WorkflowInstance.ScheduledActivities.Peek();
@ -102,12 +93,6 @@ namespace Elsa.Services.Models
public IActivityBlueprint? GetActivityBlueprintById(string id) => WorkflowBlueprint.Activities.FirstOrDefault(x => x.Id == id);
public IActivityBlueprint? GetActivityBlueprintByName(string name) => WorkflowBlueprint.Activities.FirstOrDefault(x => x.Name == name);
public void SchedulePostActivity()
{
var activity = WorkflowInstance.PostScheduledActivities.Pop();
ScheduleActivity(activity);
}
public object? GetOutputFrom(string activityName)
{
var activityBlueprint = GetActivityBlueprintByName(activityName)!;

1
src/core/Elsa.Core/Activities/ControlFlow/IfThen/Switch.cs
View file

@ -1,4 +1,3 @@
using System;
using System.Collections.Generic;
using System.Linq;
using Elsa.ActivityResults;

42
src/core/Elsa.Core/Builders/CompositeActivityBuilder.cs
View file

@ -10,7 +10,7 @@ namespace Elsa.Builders
{
public class CompositeActivityBuilder : ActivityBuilder, ICompositeActivityBuilder
{
private readonly Func<ICompositeActivityBuilder> _workflowBuilderFactory;
private readonly Func<ICompositeActivityBuilder> _compositeActivityBuilderFactory;
public CompositeActivityBuilder(IServiceProvider serviceProvider)
{
@ -18,7 +18,7 @@ namespace Elsa.Builders
ActivityBuilders = new List<IActivityBuilder>();
ConnectionBuilders = new List<IConnectionBuilder>();
_workflowBuilderFactory = () =>
_compositeActivityBuilderFactory = () =>
{
var builder = serviceProvider.GetRequiredService<ICompositeActivityBuilder>();
builder.WorkflowBuilder = WorkflowBuilder;
@ -138,14 +138,13 @@ namespace Elsa.Builders
activityBuilder.ActivityId = $"{activityIdPrefix}-{++index}";
activityBlueprints.AddRange(activityBuilders.Select(x => BuildActivityBlueprint(x, compositeActivityBlueprint)));
var activityBlueprintDictionary = activityBlueprints.ToDictionary(x => x.Id);
connections.AddRange(ConnectionBuilders.Select(x => new Connection(activityBlueprintDictionary[x.Source().ActivityId], activityBlueprintDictionary[x.Target().ActivityId], x.Outcome)));
// Build composite activities.
var compositeActivityBuilders = activityBuilders.Where(x => typeof(CompositeActivity).IsAssignableFrom(x.ActivityType));
BuildCompositeActivities(compositeActivityBuilders, activityBlueprints, connections, activityPropertyProviders);
var activityBlueprintDictionary = activityBlueprints.ToDictionary(x => x.Id);
connections.AddRange(ConnectionBuilders.Select(x => new Connection(activityBlueprintDictionary[x.Source().ActivityId], activityBlueprintDictionary[x.Target().ActivityId], x.Outcome)));
activityPropertyProviders.AddRange(
activityBuilders
.Select(x => (x.ActivityId, x.PropertyValueProviders))
@ -168,21 +167,24 @@ namespace Elsa.Builders
foreach (var activityBuilder in compositeActivityBuilders)
{
var compositeActivity = (CompositeActivity) ActivatorUtilities.CreateInstance(scope.ServiceProvider, activityBuilder.ActivityType);
var workflowBuilder = _workflowBuilderFactory();
workflowBuilder.ActivityId = activityBuilder.ActivityId;
compositeActivity.Build(workflowBuilder);
var compositeActivityBuilder = _compositeActivityBuilderFactory();
compositeActivityBuilder.ActivityId = activityBuilder.ActivityId;
compositeActivity.Build(compositeActivityBuilder);
var workflow = workflowBuilder.Build($"{activityBuilder.ActivityId}:activity");
var activityDictionary = workflow.Activities.ToDictionary(x => x.Id);
var compositeActivityBlueprint = compositeActivityBuilder.Build($"{activityBuilder.ActivityId}:activity");
var activityDictionary = compositeActivityBlueprint.Activities.ToDictionary(x => x.Id);
activityBlueprints.AddRange(workflow.Activities);
connections.AddRange(workflow.Connections.Select(x => new Connection(activityDictionary[x.Source.Activity.Id], activityDictionary[x.Target.Activity.Id], x.Source.Outcome)));
activityPropertyProviders.AddRange(workflow.ActivityPropertyProviders);
var compositeActivityBlueprint = (ICompositeActivityBlueprint) activityBlueprints.Single(x => x.Id == activityBuilder.ActivityId);
compositeActivityBlueprint.Activities = workflow.Activities;
compositeActivityBlueprint.Connections = workflow.Connections;
compositeActivityBlueprint.ActivityPropertyProviders = workflow.ActivityPropertyProviders;
activityBlueprints.AddRange(compositeActivityBlueprint.Activities);
connections.AddRange(compositeActivityBlueprint.Connections.Select(x => new Connection(activityDictionary[x.Source.Activity.Id], activityDictionary[x.Target.Activity.Id], x.Source.Outcome)));
activityPropertyProviders.AddRange(compositeActivityBlueprint.ActivityPropertyProviders);
compositeActivityBlueprint.Activities = compositeActivityBlueprint.Activities;
compositeActivityBlueprint.Connections = compositeActivityBlueprint.Connections;
compositeActivityBlueprint.ActivityPropertyProviders = compositeActivityBlueprint.ActivityPropertyProviders;
// Connect the composite activity to its starting activities.
var startActivities = compositeActivityBlueprint.GetStartActivities().ToList();
connections.AddRange(startActivities.Select(x => new Connection(compositeActivityBlueprint, x, CompositeActivity.Enter)));
}
}

9
src/core/Elsa.Core/ElsaOptions.cs
View file

@ -118,6 +118,15 @@ namespace Elsa
WorkflowFactory.Add(workflow.GetType(), workflow);
return this;
}
public ElsaOptions AddWorkflow<T>(Func<IServiceProvider, T> workflow) where T: class, IWorkflow
{
Services.AddSingleton<T>(workflow);
Services.AddSingleton<IWorkflow>(sp => sp.GetRequiredService<T>());
WorkflowFactory.Add(typeof(T), sp => sp.GetRequiredService<T>());
return this;
}
public ElsaOptions AddWorkflowsFrom<T>() => AddWorkflowsFrom(typeof(T).Assembly);

52
src/core/Elsa.Core/Services/WorkflowRunner.cs
View file

@ -1,9 +1,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.ControlFlow;
using Elsa.ActivityProviders;
using Elsa.ActivityResults;
using Elsa.Builders;
@ -301,59 +299,21 @@ namespace Elsa.Services
{
var parentActivityBlueprint = inboundConnection.Source.Activity;
if (parentActivityBlueprint.Type == nameof(While) && inboundConnection.Source.Outcome == "Iterate")
if (inboundConnection.Source.Outcome == "Iterate") // This covers While/For/ForEach/ParallelForEach based on the convention of having an unclosed "Iterate" branch. This should be refactored by allowing activities to explicitly opt-in to being rescheduled once there are no more scheduled activities.
{
workflowExecutionContext.ScheduleActivity(parentActivityBlueprint.Id);
break;
}
// if (parentActivityBlueprint is CompositeActivityBlueprint)
// {
// workflowExecutionContext.ScheduleActivity(parentActivityBlueprint.Id);
// break;
// }
}
// Schedule the parent activity, if any
}
if (!workflowExecutionContext.HasScheduledActivities && workflowExecutionContext.Status == WorkflowStatus.Running)
{
// Re-schedule the parent activity, if any
if (activityBlueprint.Parent != null && workflowBlueprint.GetActivity(activityBlueprint.Parent.Id) != null)
{
workflowExecutionContext.ScheduleActivity(activityBlueprint.Parent.Id);
}
// Find first parent that requires rescheduling for reevaluation.
// var parents = workflowBlueprint.GetInboundActivityPath(currentActivityId).ToList();
//
// foreach (var parentId in parents)
// {
// var parentActivityBlueprint = workflowBlueprint.GetActivity(parentId)!;
//
// if (parentActivityBlueprint.Type == nameof(While) || parentActivityBlueprint is ICompositeActivityBlueprint)
// {
// workflowExecutionContext.ScheduleActivity(parentId);
// break;
// }
// }
// if (scheduledActivity.ParentId != null)
// {
// workflowExecutionContext.ScheduleActivity(scheduledActivity.ParentId, null, null);
// }
//
// if (workflowExecutionContext.HasPostScheduledActivities)
// {
// workflowExecutionContext.SchedulePostActivity();
//
// // // Get children of current activity's parent.
// // var parentActivity = activityBlueprint.Parent;
// // var childActivities = workflowBlueprint.Activities.Where(x => x.Parent == parentActivity).ToList();
// // var childActivityIds = childActivities.Select(x => x.Id).ToList();
// // var scheduledPostActivities = workflowExecutionContext.WorkflowInstance.PostScheduledActivities.Where(x => childActivityIds.Contains(x.ActivityId)).ToList();
// //
// // foreach (var scheduledPostActivity in scheduledPostActivities)
// // workflowExecutionContext.ScheduleActivity(scheduledPostActivity);
// //
// // workflowExecutionContext.WorkflowInstance.PostScheduledActivities = new Stack<ScheduledActivity>(workflowExecutionContext.WorkflowInstance.PostScheduledActivities.Where(x => !childActivityIds.Contains(x.ActivityId)).Reverse().ToList());
// }
}
}

2
src/dashboards/blazor/ElsaDashboard.Application/Display/TimersDisplayProvider.cs
View file

@ -1,7 +1,5 @@
using System.Collections.Generic;
using ElsaDashboard.Application.Activities;
using ElsaDashboard.Application.Activities.Timers;
using ElsaDashboard.Extensions;
using ElsaDashboard.Models;
using ElsaDashboard.Services;

2
src/dashboards/blazor/ElsaDashboard.Application/Extensions/ServiceCollectionExtensions.cs
View file

@ -1,6 +1,4 @@
using Blazored.Modal;
using ElsaDashboard.Application.Activities;
using ElsaDashboard.Application.Activities.Console;
using ElsaDashboard.Application.Display;
using ElsaDashboard.Application.Services;
using ElsaDashboard.Extensions;

1
src/persistence/Elsa.Persistence.YesSql/Documents/WorkflowInstanceDocument.cs
View file

@ -36,7 +36,6 @@ namespace Elsa.Persistence.YesSql.Documents
public WorkflowFault? Fault { get; set; }
public Stack<ScheduledActivity> ScheduledActivities { get; set; } = new();
public Stack<ScheduledActivity> PostScheduledActivities { get; set; } = new();
public Stack<string> ParentActivities { get; set; } = new();
}
}

4
src/persistence/Elsa.Persistence.YesSql/Extensions/WorkflowDefinitionDocumentExtensions.cs
View file

@ -1,6 +1,4 @@
using System;
using System.Linq.Expressions;
using Elsa.Models;
using Elsa.Models;
using Elsa.Persistence.YesSql.Documents;
using Elsa.Persistence.YesSql.Indexes;
using YesSql;

1
src/persistence/Elsa.Persistence.YesSql/Stores/YesSqlWorkflowDefinitionStore.cs
View file

@ -1,4 +1,3 @@
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using AutoMapper;

12
src/samples/console/Elsa.Samples.ProgrammaticCompositeActivitiesConsole/Activities/CountDownActivity.cs
View file

@ -1,6 +1,6 @@
using Elsa.Activities.Console;
using System;
using Elsa.Activities.Console;
using Elsa.Activities.ControlFlow;
using Elsa.ActivityResults;
using Elsa.Attributes;
using Elsa.Builders;
using Elsa.Services;
@ -20,11 +20,13 @@ namespace Elsa.Samples.ProgrammaticCompositeActivitiesConsole.Activities
.StartWith(GetInstructions)
.WriteLine(context => (string)context.Input)
.ReadLine()
.Finish(context => (string) context.Input);
.IfElse(context => string.Equals(context.GetInput<string>(), "left", StringComparison.CurrentCultureIgnoreCase), ifElse =>
{
ifElse.When(IfElse.True).WriteLine("We're going left");
ifElse.When(IfElse.False).WriteLine("We're going right");
});
}
protected override IActivityExecutionResult Complete(ActivityExecutionContext context) => Outcome(((string) context.WorkflowExecutionContext.WorkflowInstance.Output)!);
private static void GetInstructions(ActivityExecutionContext context) => context.Output = "Turn left or right?";
}
}

8
src/samples/console/Elsa.Samples.ProgrammaticCompositeActivitiesConsole/Workflows/CompositionWorkflow.cs
View file

@ -13,10 +13,8 @@ namespace Elsa.Samples.ProgrammaticCompositeActivitiesConsole.Workflows
.WriteLine("Welcome to the Composite Activities demo workflow!")
// A custom, composite activity
.Then<CountDownActivity>(countDown =>
{
countDown.When("Left").WriteLine("We're going left.");
countDown.When("Right").WriteLine("We're going right.");
});
.Then<CountDownActivity>()
.WriteLine("Done")
;
}
}

2
src/samples/worker/Elsa.Samples.Timers/MyContainer1.cs → src/samples/worker/Elsa.Samples.Timers/Activities/MyContainer1.cs
View file

@ -5,7 +5,7 @@ using Elsa.Builders;
using Elsa.Services;
using NodaTime;
namespace Elsa.Samples.Timers
namespace Elsa.Samples.Timers.Activities
{
public class MyContainer1 : CompositeActivity
{

41
src/samples/worker/Elsa.Samples.Timers/Activities/MyContainer2.cs Normal file
View file

@ -0,0 +1,41 @@
using Elsa.Activities.Console;
using Elsa.Activities.ControlFlow;
using Elsa.Activities.Timers;
using Elsa.Builders;
using Elsa.Services;
using NodaTime;
namespace Elsa.Samples.Timers.Activities
{
public class MyContainer2 : CompositeActivity
{
public override void Build(ICompositeActivityBuilder activity)
{
activity
.WriteLine("In 2 seconds...")
.Timer(Duration.FromSeconds(2))
.WriteLine("The time is ripe.")
.Then<Fork>(fork => fork.WithBranches("C", "D", "E"), fork =>
{
fork.When("C")
.Then<MyContainer1>()
.Then("Join2");
fork
.When("D")
.While(true, @while => @while
.Timer(Duration.FromSeconds(1))
.WriteLine("Timer D went off"))
.Then("Join2");
fork
.When("E")
.Timer(Duration.FromSeconds(15))
.WriteLine("Timer E went off. Exiting fork.")
.Then("Join2");
})
.Add<Join>(join => join.WithMode(Join.JoinMode.WaitAny)).WithName("Join2").WriteLine("Container 2 Joined!")
;
}
}
}

35
src/samples/worker/Elsa.Samples.Timers/MyContainer2.cs
View file

@ -1,35 +0,0 @@
using Elsa.Activities.Console;
using Elsa.Activities.ControlFlow;
using Elsa.Activities.Timers;
using Elsa.Builders;
using Elsa.Services;
using NodaTime;
namespace Elsa.Samples.Timers
{
public class MyContainer2 : CompositeActivity
{
public override void Build(ICompositeActivityBuilder activity)
{
activity
.StartIn(Duration.FromSeconds(5))
.Then<Fork>(fork => fork.WithBranches("A", "B"), fork =>
{
fork
.When("A")
.While(true, @while => @while
.Timer(Duration.FromSeconds(5))
.WriteLine("Timer C went off"))
.Then("Join2");
fork
.When("B")
.While(true, @while => @while
.Timer(Duration.FromSeconds(5))
.WriteLine("Timer D went off"))
.Then("Join2");
})
.Add<Join>(join => join.WithMode(Join.JoinMode.WaitAny)).WithName("Join2").WriteLine("Container 2 Joined!");
}
}
}

11
src/samples/worker/Elsa.Samples.Timers/Program.cs
View file

@ -1,5 +1,7 @@
using System.Threading.Tasks;
using Elsa.Persistence.YesSql.Extensions;
using Elsa.Samples.Timers.Activities;
using Elsa.Samples.Timers.Workflows;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using NodaTime;
@ -19,14 +21,15 @@ namespace Elsa.Samples.Timers
.AddElsa(options => options.UseYesSqlPersistence()
.AddConsoleActivities()
.AddQuartzTimerActivities()
.AddWorkflow<RecurringTaskWorkflow>()
//.AddWorkflow<RecurringTaskWorkflow>()
.AddActivity<MyContainer1>()
.AddActivity<MyContainer2>()
//.AddWorkflow<CancelTimerWorkflow>()
.AddWorkflow<CancelTimerWorkflow>()
//.AddWorkflow<CronTaskWorkflow>()
//.AddWorkflow(new OneOffWorkflow(SystemClock.Instance.GetCurrentInstant().Plus(Duration.FromSeconds(5))))
//.AddWorkflow(sp => ActivatorUtilities.CreateInstance<OneOffWorkflow>(sp, sp.GetRequiredService<IClock>().GetCurrentInstant().Plus(Duration.FromSeconds(5)), sp.GetRequiredService<IClock>()))
)
.StartWorkflow<RecurringTaskWorkflow>();
//.StartWorkflow<RecurringTaskWorkflow>()
;
});
}
}

41
src/samples/worker/Elsa.Samples.Timers/RecurringTaskWorkflow.cs
View file

@ -1,41 +0,0 @@
using System;
using System.Threading.Tasks;
using Elsa.Activities.Console;
using Elsa.Activities.ControlFlow;
using Elsa.Activities.Timers;
using Elsa.Builders;
using NodaTime;
namespace Elsa.Samples.Timers
{
public class RecurringTaskWorkflow : IWorkflow
{
private readonly IClock _clock;
public RecurringTaskWorkflow(IClock clock)
{
_clock = clock;
}
public void Build(IWorkflowBuilder workflow)
{
workflow
.WriteLine("Started")
.Then<Fork>(fork => fork.WithBranches("A", "B"), fork =>
{
fork
.When("A")
.Then<MyContainer1>()
.Then("Join3");
fork
.When("B")
.Then<MyContainer2>()
.Then("Join3");
})
.Add<Join>(join => join.WithMode(Join.JoinMode.WaitAny)).WithName("Join3")
.WriteLine("Workflow Joined!")
.WriteLine("Finished");
}
}
}

5
src/samples/worker/Elsa.Samples.Timers/CancelTimerWorkflow.cs → src/samples/worker/Elsa.Samples.Timers/Workflows/CancelTimerWorkflow.cs
View file

@ -1,12 +1,11 @@
using System.Collections.Generic;
using Elsa.Activities.Console;
using Elsa.Activities.ControlFlow;
using Elsa.Activities.Timers;
using Elsa.Builders;
using NodaTime;
namespace Elsa.Samples.Timers
namespace Elsa.Samples.Timers.Workflows
{
public class CancelTimerWorkflow : IWorkflow
{
@ -14,7 +13,7 @@ namespace Elsa.Samples.Timers
{
workflow
.StartAt(SystemClock.Instance.GetCurrentInstant().Plus(Duration.FromSeconds(5)))
.WriteLine("CancelTimerWorkflow is executed")
.WriteLine("CancelTimerWorkflow is executing")
.Then<Fork>(
activity => activity.Set(x => x.Branches, new HashSet<string>(new[] { "Branch 1", "Branch 2" })),
fork =>

2
src/samples/worker/Elsa.Samples.Timers/CronTaskWorkflow.cs → src/samples/worker/Elsa.Samples.Timers/Workflows/CronTaskWorkflow.cs
View file

@ -3,7 +3,7 @@ using Elsa.Activities.Console;
using Elsa.Activities.Timers;
using Elsa.Builders;
namespace Elsa.Samples.Timers
namespace Elsa.Samples.Timers.Workflows
{
public class CronTaskWorkflow : IWorkflow
{

10
src/samples/worker/Elsa.Samples.Timers/OneOffWorkflow.cs → src/samples/worker/Elsa.Samples.Timers/Workflows/OneOffWorkflow.cs
View file

@ -3,7 +3,7 @@ using Elsa.Activities.Timers;
using Elsa.Builders;
using NodaTime;
namespace Elsa.Samples.Timers
namespace Elsa.Samples.Timers.Workflows
{
/// <summary>
/// A workflow that executes only once in the near future.
@ -11,19 +11,21 @@ namespace Elsa.Samples.Timers
public class OneOffWorkflow : IWorkflow
{
private readonly Instant _executeAt;
private readonly IClock _clock;
public OneOffWorkflow(Instant executeAt)
public OneOffWorkflow(Instant executeAt, IClock clock)
{
_executeAt = executeAt;
_clock = clock;
}
public void Build(IWorkflowBuilder workflow)
{
workflow
.StartAt(_executeAt)
.WriteLine(context => $"Started at {context.GetService<IClock>().GetCurrentInstant()}. Next event happens 3 seconds from now.")
.WriteLine(() => $"Started at {_clock.GetCurrentInstant()}. Next event happens 3 seconds from now.")
.StartIn(Duration.FromSeconds(3))
.WriteLine(context => $"Follow-up occurred at {context.GetService<IClock>().GetCurrentInstant()}.");
.WriteLine(() => $"Follow-up occurred at {_clock.GetCurrentInstant()}.");
}
}
}

25
src/samples/worker/Elsa.Samples.Timers/Workflows/RecurringTaskWorkflow.cs Normal file
View file

@ -0,0 +1,25 @@
using Elsa.Activities.Console;
using Elsa.Builders;
using Elsa.Samples.Timers.Activities;
using NodaTime;
namespace Elsa.Samples.Timers.Workflows
{
public class RecurringTaskWorkflow : IWorkflow
{
private readonly IClock _clock;
public RecurringTaskWorkflow(IClock clock)
{
_clock = clock;
}
public void Build(IWorkflowBuilder workflow)
{
workflow
.WriteLine("Started")
.Then<MyContainer2>()
.WriteLine("Finished");
}
}
}