Refactor rename to updated workflow terminology
This commit is contained in:
parent
b7103188fd
commit
dce24b4f3b
|
|
@ -14,16 +14,16 @@ namespace Elsa.Activities.Http.Controllers
|
|||
public class HttpWorkflowsController : ControllerBase
|
||||
{
|
||||
private readonly ITokenService tokenService;
|
||||
private readonly IWorkflowInvoker workflowInvoker;
|
||||
private readonly IWorkflowRunner workflowRunner;
|
||||
private readonly IWorkflowInstanceStore workflowInstanceStore;
|
||||
|
||||
public HttpWorkflowsController(
|
||||
ITokenService tokenService,
|
||||
IWorkflowInvoker workflowInvoker,
|
||||
IWorkflowRunner workflowRunner,
|
||||
IWorkflowInstanceStore workflowInstanceStore)
|
||||
{
|
||||
this.tokenService = tokenService;
|
||||
this.workflowInvoker = workflowInvoker;
|
||||
this.workflowRunner = workflowRunner;
|
||||
this.workflowInstanceStore = workflowInstanceStore;
|
||||
}
|
||||
|
||||
|
|
@ -42,7 +42,7 @@ namespace Elsa.Activities.Http.Controllers
|
|||
var input = new Variables();
|
||||
input.SetVariable("Signal", signal.Name);
|
||||
|
||||
await workflowInvoker.ResumeAsync(workflowInstance, input, cancellationToken: cancellationToken);
|
||||
await workflowRunner.ResumeAsync(workflowInstance, input, cancellationToken: cancellationToken);
|
||||
|
||||
return HttpContext.Items.ContainsKey(WorkflowHttpResult.Instance)
|
||||
? (IActionResult)new EmptyResult()
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
{
|
||||
private readonly HttpContext httpContext;
|
||||
private readonly ITokenService tokenService;
|
||||
private readonly IWorkflowInvoker workflowInvoker;
|
||||
private readonly IWorkflowRunner workflowRunner;
|
||||
private readonly IWorkflowRegistry workflowRegistry;
|
||||
private readonly IWorkflowFactory workflowFactory;
|
||||
private readonly IWorkflowInstanceStore workflowInstanceStore;
|
||||
|
|
@ -24,14 +24,14 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
public SignalRequestHandler(
|
||||
IHttpContextAccessor httpContextAccessor,
|
||||
ITokenService tokenService,
|
||||
IWorkflowInvoker workflowInvoker,
|
||||
IWorkflowRunner workflowRunner,
|
||||
IWorkflowRegistry workflowRegistry,
|
||||
IWorkflowFactory workflowFactory,
|
||||
IWorkflowInstanceStore workflowInstanceStore)
|
||||
{
|
||||
httpContext = httpContextAccessor.HttpContext;
|
||||
this.tokenService = tokenService;
|
||||
this.workflowInvoker = workflowInvoker;
|
||||
this.workflowRunner = workflowRunner;
|
||||
this.workflowRegistry = workflowRegistry;
|
||||
this.workflowFactory = workflowFactory;
|
||||
this.workflowInstanceStore = workflowInstanceStore;
|
||||
|
|
@ -51,7 +51,7 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
return new NotFoundResult();
|
||||
|
||||
if (!CheckIfExecuting(workflowInstance))
|
||||
return new BadRequestResult($"Cannot signal a workflow with status other than {WorkflowStatus.Executing}. Actual workflow status: {workflowInstance.Status}.");
|
||||
return new BadRequestResult($"Cannot signal a workflow with status other than {WorkflowStatus.Running}. Actual workflow status: {workflowInstance.Status}.");
|
||||
|
||||
await ResumeWorkflowAsync(workflowInstance, signal);
|
||||
|
||||
|
|
@ -69,7 +69,7 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
await workflowInstanceStore.GetByIdAsync(signal.WorkflowInstanceId, cancellationToken);
|
||||
|
||||
private bool CheckIfExecuting(WorkflowInstance workflowInstance) =>
|
||||
workflowInstance.Status == WorkflowStatus.Executing;
|
||||
workflowInstance.Status == WorkflowStatus.Running;
|
||||
|
||||
private async Task ResumeWorkflowAsync(WorkflowInstance workflowInstance, Signal signal)
|
||||
{
|
||||
|
|
@ -83,7 +83,7 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
|
||||
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, workflowInstance);
|
||||
var blockingSignalActivities = workflow.BlockingActivities.ToList();
|
||||
await workflowInvoker.ResumeAsync(workflow, blockingSignalActivities, cancellationToken);
|
||||
await workflowRunner.ResumeAsync(workflow, blockingSignalActivities, cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -18,19 +18,19 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
public class TriggerRequestHandler : IRequestHandler
|
||||
{
|
||||
private readonly HttpContext httpContext;
|
||||
private readonly IWorkflowInvoker workflowInvoker;
|
||||
private readonly IWorkflowRunner workflowRunner;
|
||||
private readonly IWorkflowRegistry registry;
|
||||
private readonly IWorkflowInstanceStore workflowInstanceStore;
|
||||
private readonly CancellationToken cancellationToken;
|
||||
|
||||
public TriggerRequestHandler(
|
||||
IHttpContextAccessor httpContext,
|
||||
IWorkflowInvoker workflowInvoker,
|
||||
IWorkflowRunner workflowRunner,
|
||||
IWorkflowRegistry registry,
|
||||
IWorkflowInstanceStore workflowInstanceStore)
|
||||
{
|
||||
this.httpContext = httpContext.HttpContext;
|
||||
this.workflowInvoker = workflowInvoker;
|
||||
this.workflowRunner = workflowRunner;
|
||||
this.registry = registry;
|
||||
this.workflowInstanceStore = workflowInstanceStore;
|
||||
cancellationToken = httpContext.HttpContext.RequestAborted;
|
||||
|
|
@ -87,7 +87,7 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
{
|
||||
foreach (var item in items)
|
||||
{
|
||||
await workflowInvoker.StartAsync(
|
||||
await workflowRunner.RunAsync(
|
||||
item.Item1,
|
||||
Variables.Empty,
|
||||
new[] { item.Item2.Id },
|
||||
|
|
@ -99,7 +99,7 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
{
|
||||
foreach (var (workflowInstance, activity) in items)
|
||||
{
|
||||
await workflowInvoker.ResumeAsync(
|
||||
await workflowRunner.ResumeAsync(
|
||||
workflowInstance,
|
||||
Variables.Empty,
|
||||
new[] { activity.Id },
|
||||
|
|
|
|||
|
|
@ -19,11 +19,11 @@ namespace Elsa.Activities.MassTransit.Consumers
|
|||
new PropertyCorrelationIdSelector<T>("CommandId")
|
||||
};
|
||||
|
||||
private readonly IWorkflowInvoker workflowInvoker;
|
||||
private readonly IWorkflowRunner workflowRunner;
|
||||
|
||||
public WorkflowConsumer(IWorkflowInvoker workflowInvoker)
|
||||
public WorkflowConsumer(IWorkflowRunner workflowRunner)
|
||||
{
|
||||
this.workflowInvoker = workflowInvoker;
|
||||
this.workflowRunner = workflowRunner;
|
||||
}
|
||||
|
||||
public async Task Consume(ConsumeContext<T> context)
|
||||
|
|
@ -42,7 +42,7 @@ namespace Elsa.Activities.MassTransit.Consumers
|
|||
break;
|
||||
}
|
||||
|
||||
await workflowInvoker.TriggerAsync(
|
||||
await workflowRunner.TriggerAsync(
|
||||
activityType,
|
||||
input,
|
||||
correlationId.ToString(),
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ namespace Elsa.Activities.Timers.HostedServices
|
|||
{
|
||||
using (var scope = serviceProvider.CreateScope())
|
||||
{
|
||||
var workflowInvoker = scope.ServiceProvider.GetRequiredService<IWorkflowInvoker>();
|
||||
var workflowInvoker = scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
|
||||
await workflowInvoker.TriggerAsync(nameof(TimerEvent), Variables.Empty, stoppingToken);
|
||||
await workflowInvoker.TriggerAsync(nameof(CronEvent), Variables.Empty, stoppingToken);
|
||||
await workflowInvoker.TriggerAsync(nameof(InstantEvent), Variables.Empty, stoppingToken);
|
||||
|
|
|
|||
|
|
@ -19,11 +19,11 @@ namespace Elsa.Activities.Workflows.Activities
|
|||
)]
|
||||
public class TriggerSignal : Activity
|
||||
{
|
||||
private readonly IWorkflowInvoker workflowInvoker;
|
||||
private readonly IWorkflowRunner workflowRunner;
|
||||
|
||||
public TriggerSignal(IWorkflowInvoker workflowInvoker)
|
||||
public TriggerSignal(IWorkflowRunner workflowRunner)
|
||||
{
|
||||
this.workflowInvoker = workflowInvoker;
|
||||
this.workflowRunner = workflowRunner;
|
||||
}
|
||||
|
||||
[ActivityProperty(Hint = "An expression that evaluates to the name of the signal to trigger.")]
|
||||
|
|
@ -57,7 +57,7 @@ namespace Elsa.Activities.Workflows.Activities
|
|||
|
||||
input.SetVariable("Signal", signal);
|
||||
|
||||
await workflowInvoker.TriggerAsync(
|
||||
await workflowRunner.TriggerAsync(
|
||||
nameof(Signaled),
|
||||
input,
|
||||
correlationId,
|
||||
|
|
|
|||
|
|
@ -17,11 +17,11 @@ namespace Elsa.Activities.Workflows.Activities
|
|||
)]
|
||||
public class TriggerWorkflow : Activity
|
||||
{
|
||||
private readonly IWorkflowInvoker workflowInvoker;
|
||||
private readonly IWorkflowRunner workflowRunner;
|
||||
|
||||
public TriggerWorkflow(IWorkflowInvoker workflowInvoker)
|
||||
public TriggerWorkflow(IWorkflowRunner workflowRunner)
|
||||
{
|
||||
this.workflowInvoker = workflowInvoker;
|
||||
this.workflowRunner = workflowRunner;
|
||||
}
|
||||
|
||||
[ActivityProperty(Hint = "An expression that evaluates to the activity type to use when triggering workflows.")]
|
||||
|
|
@ -53,7 +53,7 @@ namespace Elsa.Activities.Workflows.Activities
|
|||
var input = await context.EvaluateAsync(Input, cancellationToken);
|
||||
var correlationId = await context.EvaluateAsync(CorrelationId, cancellationToken);
|
||||
|
||||
await workflowInvoker.TriggerAsync(
|
||||
await workflowRunner.TriggerAsync(
|
||||
activityType,
|
||||
input,
|
||||
correlationId,
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ namespace Elsa.Activities.Workflows.Extensions
|
|||
public static class WorkflowInvokerExtensions
|
||||
{
|
||||
public static async Task TriggerSignalAsync(
|
||||
this IWorkflowInvoker workflowInvoker,
|
||||
this IWorkflowRunner workflowRunner,
|
||||
string signalName,
|
||||
Variables input = default,
|
||||
Func<JObject, bool> activityStatePredicate = null,
|
||||
|
|
@ -24,7 +24,7 @@ namespace Elsa.Activities.Workflows.Extensions
|
|||
if (input != null)
|
||||
combinedInput.SetVariables(input);
|
||||
|
||||
await workflowInvoker.TriggerAsync(nameof(Signaled), combinedInput, correlationId, activityStatePredicate, cancellationToken);
|
||||
await workflowRunner.TriggerAsync(nameof(Signaled), combinedInput, correlationId, activityStatePredicate, cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -2,10 +2,11 @@
|
|||
{
|
||||
public enum WorkflowStatus
|
||||
{
|
||||
Idle,
|
||||
Executing,
|
||||
Finished,
|
||||
Pending,
|
||||
Running,
|
||||
Completed,
|
||||
Suspended,
|
||||
Faulted,
|
||||
Aborted
|
||||
Cancelled
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,13 +7,13 @@ namespace Elsa.Results
|
|||
{
|
||||
public abstract class ActivityExecutionResult : IActivityExecutionResult
|
||||
{
|
||||
public virtual Task ExecuteAsync(IWorkflowInvoker invoker, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken)
|
||||
public virtual Task ExecuteAsync(IWorkflowRunner runner, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken)
|
||||
{
|
||||
Execute(invoker, workflowContext);
|
||||
Execute(runner, workflowContext);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
protected virtual void Execute(IWorkflowInvoker invoker, WorkflowExecutionContext workflowContext)
|
||||
protected virtual void Execute(IWorkflowRunner runner, WorkflowExecutionContext workflowContext)
|
||||
{
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ namespace Elsa.Results
|
|||
/// </summary>
|
||||
public class NoopResult : ActivityExecutionResult
|
||||
{
|
||||
protected override void Execute(IWorkflowInvoker invoker, WorkflowExecutionContext workflowContext)
|
||||
protected override void Execute(IWorkflowRunner runner, WorkflowExecutionContext workflowContext)
|
||||
{
|
||||
// Noop.
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,13 +8,13 @@ namespace Elsa.Services.Extensions
|
|||
{
|
||||
public static class WorkflowExtensions
|
||||
{
|
||||
public static bool IsExecuting(this Workflow workflow) => workflow.Status == WorkflowStatus.Executing;
|
||||
public static bool IsExecuting(this Workflow workflow) => workflow.Status == WorkflowStatus.Running;
|
||||
|
||||
public static bool IsFaultedOrAborted(this Workflow workflow) =>
|
||||
workflow.Status == WorkflowStatus.Aborted ||
|
||||
workflow.Status == WorkflowStatus.Cancelled ||
|
||||
workflow.Status == WorkflowStatus.Faulted;
|
||||
|
||||
public static bool IsFinished(this Workflow workflow) => workflow.Status == WorkflowStatus.Finished;
|
||||
public static bool IsFinished(this Workflow workflow) => workflow.Status == WorkflowStatus.Completed;
|
||||
|
||||
public static IEnumerable<IActivity> GetStartActivities(this Workflow workflow)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -10,12 +10,12 @@ namespace Elsa.Services.Extensions
|
|||
/// Starts new workflows that start with the specified activity name and resumes halted workflows that are blocked on activities with the specified activity name.
|
||||
/// </summary>
|
||||
public static Task TriggerAsync(
|
||||
this IWorkflowInvoker workflowInvoker,
|
||||
this IWorkflowRunner workflowRunner,
|
||||
string activityType,
|
||||
Variables input,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return workflowInvoker.TriggerAsync(activityType, input, cancellationToken: cancellationToken);
|
||||
return workflowRunner.TriggerAsync(activityType, input, cancellationToken: cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,63 +1,63 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Models;
|
||||
using Elsa.Services.Models;
|
||||
using Newtonsoft.Json.Linq;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
public interface IWorkflowInvoker
|
||||
{
|
||||
Task<WorkflowExecutionContext> StartAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
|
||||
Task<WorkflowExecutionContext> StartAsync<T>(
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default) where T : IWorkflow, new();
|
||||
|
||||
Task<WorkflowExecutionContext> StartAsync(
|
||||
WorkflowDefinitionVersion workflowDefinition,
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
|
||||
Task<WorkflowExecutionContext> ResumeAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
|
||||
Task<WorkflowExecutionContext> ResumeAsync<T>(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default
|
||||
) where T : IWorkflow, new();
|
||||
|
||||
Task<WorkflowExecutionContext> ResumeAsync(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
|
||||
/// <summary>
|
||||
/// Starts new workflows that start with the specified activity name and resumes halted workflows that are blocked on activities with the specified activity name.
|
||||
/// </summary>
|
||||
Task<IEnumerable<WorkflowExecutionContext>> TriggerAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default);
|
||||
}
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Models;
|
||||
using Elsa.Services.Models;
|
||||
using Newtonsoft.Json.Linq;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
public interface IWorkflowRunner
|
||||
{
|
||||
Task<WorkflowExecutionContext> RunAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
|
||||
Task<WorkflowExecutionContext> RunAsync<T>(
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default) where T : IWorkflow, new();
|
||||
|
||||
Task<WorkflowExecutionContext> RunAsync(
|
||||
WorkflowDefinitionVersion workflowDefinition,
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
|
||||
Task<WorkflowExecutionContext> ResumeAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
|
||||
Task<WorkflowExecutionContext> ResumeAsync<T>(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default
|
||||
) where T : IWorkflow, new();
|
||||
|
||||
Task<WorkflowExecutionContext> ResumeAsync(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
|
||||
/// <summary>
|
||||
/// Starts new workflows that start with the specified activity name and resumes halted workflows that are blocked on activities with the specified activity name.
|
||||
/// </summary>
|
||||
Task<IEnumerable<WorkflowExecutionContext>> TriggerAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default);
|
||||
}
|
||||
}
|
||||
|
|
@ -5,6 +5,6 @@ namespace Elsa.Services.Models
|
|||
{
|
||||
public interface IActivityExecutionResult
|
||||
{
|
||||
Task ExecuteAsync(IWorkflowInvoker invoker, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken);
|
||||
Task ExecuteAsync(IWorkflowRunner runner, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -84,7 +84,7 @@ namespace Elsa.Services.Models
|
|||
public void Start()
|
||||
{
|
||||
Workflow.StartedAt = clock.GetCurrentInstant();
|
||||
Workflow.Status = WorkflowStatus.Executing;
|
||||
Workflow.Status = WorkflowStatus.Running;
|
||||
}
|
||||
|
||||
public void Fault(IActivity activity, Exception exception) => Fault(activity, exception.Message);
|
||||
|
|
@ -109,14 +109,14 @@ namespace Elsa.Services.Models
|
|||
public void Finish()
|
||||
{
|
||||
Workflow.FinishedAt = clock.GetCurrentInstant();
|
||||
Workflow.Status = WorkflowStatus.Finished;
|
||||
Workflow.Status = WorkflowStatus.Completed;
|
||||
Workflow.BlockingActivities.Clear();
|
||||
}
|
||||
|
||||
public void Abort()
|
||||
{
|
||||
Workflow.AbortedAt = clock.GetCurrentInstant();
|
||||
Workflow.Status = WorkflowStatus.Aborted;
|
||||
Workflow.Status = WorkflowStatus.Cancelled;
|
||||
}
|
||||
|
||||
public Variables GetVariables() => Workflow.Scopes
|
||||
|
|
|
|||
|
|
@ -80,7 +80,7 @@ namespace Microsoft.Extensions.DependencyInjection
|
|||
.AddSingleton<IWorkflowSerializerProvider, WorkflowSerializerProvider>()
|
||||
.AddTransient<IWorkflowRegistry, WorkflowRegistry>()
|
||||
.AddScoped<IWorkflowEventHandler, PersistenceWorkflowEventHandler>()
|
||||
.AddScoped<IWorkflowInvoker, WorkflowInvoker>()
|
||||
.AddScoped<IWorkflowRunner, WorkflowRunner>()
|
||||
.AddScoped<IActivityResolver, ActivityResolver>()
|
||||
.AddScoped<IWorkflowEventHandler, ActivityLoggingWorkflowEventHandler>()
|
||||
.AddTransient<IWorkflowProvider, StoreWorkflowProvider>()
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ namespace Elsa.Persistence.Memory
|
|||
{
|
||||
var query = workflowInstances.Values.AsQueryable();
|
||||
|
||||
query = query.Where(x => x.Status == WorkflowStatus.Executing);
|
||||
query = query.Where(x => x.Status == WorkflowStatus.Running);
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(correlationId))
|
||||
query = query.Where(x => x.CorrelationId == correlationId);
|
||||
|
|
|
|||
|
|
@ -21,11 +21,11 @@ namespace Elsa.Results
|
|||
|
||||
public IReadOnlyCollection<IActivityExecutionResult> Results { get; }
|
||||
|
||||
public override async Task ExecuteAsync(IWorkflowInvoker invoker, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken)
|
||||
public override async Task ExecuteAsync(IWorkflowRunner runner, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken)
|
||||
{
|
||||
foreach (var result in Results)
|
||||
{
|
||||
await result.ExecuteAsync(invoker, workflowContext, cancellationToken);
|
||||
await result.ExecuteAsync(runner, workflowContext, cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ namespace Elsa.Results
|
|||
}
|
||||
|
||||
public override async Task ExecuteAsync(
|
||||
IWorkflowInvoker invoker,
|
||||
IWorkflowRunner runner,
|
||||
WorkflowExecutionContext workflowContext,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ namespace Elsa.Results
|
|||
public class FinishWorkflowResult : ActivityExecutionResult
|
||||
{
|
||||
public override async Task ExecuteAsync(
|
||||
IWorkflowInvoker invoker,
|
||||
IWorkflowRunner runner,
|
||||
WorkflowExecutionContext workflowContext,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ namespace Elsa.Results
|
|||
|
||||
public bool ContinueOnFirstPass { get; }
|
||||
|
||||
public override async Task ExecuteAsync(IWorkflowInvoker invoker, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken)
|
||||
public override async Task ExecuteAsync(IWorkflowRunner runner, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken)
|
||||
{
|
||||
var activity = workflowContext.CurrentActivity;
|
||||
|
||||
|
|
@ -29,7 +29,7 @@ namespace Elsa.Results
|
|||
|
||||
workflowContext.IsFirstPass = false;
|
||||
|
||||
await result.ExecuteAsync(invoker, workflowContext, cancellationToken);
|
||||
await result.ExecuteAsync(runner, workflowContext, cancellationToken);
|
||||
}
|
||||
else
|
||||
{
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ namespace Elsa.Results
|
|||
|
||||
public IReadOnlyList<string> EndpointNames { get; }
|
||||
|
||||
public override async Task ExecuteAsync(IWorkflowInvoker invoker, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken)
|
||||
public override async Task ExecuteAsync(IWorkflowRunner runner, WorkflowExecutionContext workflowContext, CancellationToken cancellationToken)
|
||||
{
|
||||
var currentActivity = workflowContext.CurrentActivity;
|
||||
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ namespace Elsa.Results
|
|||
|
||||
public Variable Value { get; }
|
||||
|
||||
protected override void Execute(IWorkflowInvoker invoker, WorkflowExecutionContext workflowContext)
|
||||
protected override void Execute(IWorkflowRunner runner, WorkflowExecutionContext workflowContext)
|
||||
{
|
||||
workflowContext.CurrentActivity.Output = Value;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ namespace Elsa.Results
|
|||
this.activity = activity;
|
||||
}
|
||||
|
||||
protected override void Execute(IWorkflowInvoker invoker, WorkflowExecutionContext workflowContext)
|
||||
protected override void Execute(IWorkflowRunner runner, WorkflowExecutionContext workflowContext)
|
||||
{
|
||||
workflowContext.ScheduleActivity(activity);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,456 +1,456 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Models;
|
||||
using Elsa.Persistence;
|
||||
using Elsa.Services.Extensions;
|
||||
using Elsa.Services.Models;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Newtonsoft.Json.Linq;
|
||||
using NodaTime;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
internal class WorkflowInvoker : IWorkflowInvoker
|
||||
{
|
||||
private readonly IActivityInvoker activityInvoker;
|
||||
private readonly IWorkflowFactory workflowFactory;
|
||||
private readonly IWorkflowRegistry workflowRegistry;
|
||||
private readonly IWorkflowInstanceStore workflowInstanceStore;
|
||||
private readonly IEnumerable<IWorkflowEventHandler> workflowEventHandlers;
|
||||
private readonly IClock clock;
|
||||
private readonly IServiceProvider serviceProvider;
|
||||
private readonly ILogger logger;
|
||||
|
||||
public WorkflowInvoker(
|
||||
IActivityInvoker activityInvoker,
|
||||
IWorkflowFactory workflowFactory,
|
||||
IWorkflowRegistry workflowRegistry,
|
||||
IWorkflowInstanceStore workflowInstanceStore,
|
||||
IEnumerable<IWorkflowEventHandler> workflowEventHandlers,
|
||||
IClock clock,
|
||||
IServiceProvider serviceProvider,
|
||||
ILogger<WorkflowInvoker> logger)
|
||||
{
|
||||
this.activityInvoker = activityInvoker;
|
||||
this.workflowFactory = workflowFactory;
|
||||
this.workflowRegistry = workflowRegistry;
|
||||
this.workflowInstanceStore = workflowInstanceStore;
|
||||
this.workflowEventHandlers = workflowEventHandlers;
|
||||
this.clock = clock;
|
||||
this.serviceProvider = serviceProvider;
|
||||
this.logger = logger;
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> StartAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return ExecuteAsync(workflow, false, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> StartAsync(
|
||||
WorkflowDefinitionVersion workflowDefinition,
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, correlationId: correlationId);
|
||||
var startActivities = workflow.Activities.Find(startActivityIds);
|
||||
|
||||
return ExecuteAsync(workflow, false, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> StartAsync<T>(
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default) where T : IWorkflow, new()
|
||||
{
|
||||
var workflow = workflowFactory.CreateWorkflow<T>(input, correlationId: correlationId);
|
||||
var startActivities = workflow.Activities.Find(startActivityIds);
|
||||
|
||||
return ExecuteAsync(workflow, false, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> ResumeAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return ExecuteAsync(workflow, true, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> ResumeAsync<T>(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = null,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default) where T : IWorkflow, new()
|
||||
{
|
||||
var workflow = workflowFactory.CreateWorkflow<T>(input, workflowInstance);
|
||||
var startActivities = workflow.Activities.Find(startActivityIds);
|
||||
return ExecuteAsync(workflow, true, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<WorkflowExecutionContext> ResumeAsync(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = null,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var definition = await workflowRegistry.GetWorkflowDefinitionAsync(
|
||||
workflowInstance.DefinitionId,
|
||||
VersionOptions.SpecificVersion(workflowInstance.Version),
|
||||
cancellationToken);
|
||||
var workflow = workflowFactory.CreateWorkflow(definition, input, workflowInstance);
|
||||
return await ExecuteAsync(workflow, true, startActivityIds, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<WorkflowExecutionContext>> TriggerAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var startedExecutionContexts = await StartManyAsync(
|
||||
activityType,
|
||||
input,
|
||||
correlationId,
|
||||
activityStatePredicate,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
var resumedExecutionContexts = await ResumeManyAsync(
|
||||
activityType,
|
||||
input,
|
||||
correlationId,
|
||||
activityStatePredicate,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
return startedExecutionContexts.Concat(resumedExecutionContexts);
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<WorkflowExecutionContext>> ResumeManyAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowInstances = await workflowInstanceStore
|
||||
.ListByBlockingActivityAsync(activityType, correlationId, cancellationToken)
|
||||
.ToListAsync();
|
||||
|
||||
if (activityStatePredicate != null)
|
||||
workflowInstances = workflowInstances.Where(x => activityStatePredicate(x.Item2.State)).ToList();
|
||||
|
||||
return await ResumeManyAsync(
|
||||
workflowInstances,
|
||||
input,
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<WorkflowExecutionContext>> StartManyAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowDefinitions = await workflowRegistry.ListByStartActivityAsync(activityType, cancellationToken);
|
||||
|
||||
if (activityStatePredicate != null)
|
||||
workflowDefinitions = workflowDefinitions.Where(x => activityStatePredicate(x.Item2.State));
|
||||
|
||||
workflowDefinitions = await FilterRunningSingletonsAsync(
|
||||
workflowDefinitions,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
return await StartManyAsync(workflowDefinitions, input, correlationId, cancellationToken);
|
||||
}
|
||||
|
||||
private Task<WorkflowExecutionContext> ExecuteAsync(
|
||||
Workflow workflow,
|
||||
bool resume,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var startActivities = startActivityIds != null
|
||||
? workflow.Activities.Find(startActivityIds)
|
||||
: Enumerable.Empty<IActivity>();
|
||||
|
||||
return ExecuteAsync(workflow, resume, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<WorkflowExecutionContext> ExecuteAsync(
|
||||
Workflow workflow,
|
||||
bool resume,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowExecutionContext = await CreateWorkflowExecutionContextAsync(
|
||||
workflow,
|
||||
startActivities,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
var start = !resume;
|
||||
|
||||
while (workflowExecutionContext.HasScheduledActivities)
|
||||
{
|
||||
var currentActivity = workflowExecutionContext.PopScheduledActivity();
|
||||
|
||||
var result = start
|
||||
? await ExecuteActivityAsync(workflowExecutionContext, currentActivity, cancellationToken)
|
||||
: await ResumeActivityAsync(workflowExecutionContext, currentActivity, cancellationToken);
|
||||
|
||||
if (result == null)
|
||||
break;
|
||||
|
||||
await result.ExecuteAsync(this, workflowExecutionContext, cancellationToken);
|
||||
|
||||
workflowExecutionContext.IsFirstPass = false;
|
||||
start = true;
|
||||
}
|
||||
|
||||
await FinalizeWorkflowExecutionAsync(workflowExecutionContext, cancellationToken);
|
||||
|
||||
return workflowExecutionContext;
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<WorkflowExecutionContext>> StartManyAsync(
|
||||
IEnumerable<(WorkflowDefinitionVersion, ActivityDefinition)> workflowDefinitions,
|
||||
Variables input,
|
||||
string correlationId,
|
||||
CancellationToken cancellationToken1)
|
||||
{
|
||||
var executionContexts = new List<WorkflowExecutionContext>();
|
||||
|
||||
foreach (var (workflowDefinition, activityDefinition) in workflowDefinitions)
|
||||
{
|
||||
var startActivityIds = workflowDefinition.Activities
|
||||
.Where(x => x.Id == activityDefinition.Id)
|
||||
.Select(x => x.Id);
|
||||
|
||||
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, correlationId: correlationId);
|
||||
|
||||
var executionContext = await ExecuteAsync(
|
||||
workflow,
|
||||
false,
|
||||
startActivityIds,
|
||||
cancellationToken1
|
||||
);
|
||||
executionContexts.Add(executionContext);
|
||||
}
|
||||
|
||||
return executionContexts;
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<WorkflowExecutionContext>> ResumeManyAsync(
|
||||
IEnumerable<(WorkflowInstance, ActivityInstance)> workflowInstances,
|
||||
Variables input,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var executionContexts = new List<WorkflowExecutionContext>();
|
||||
var workflowInstanceGroups = workflowInstances.GroupBy(x => x.Item1);
|
||||
|
||||
foreach (var workflowInstanceGroup in workflowInstanceGroups)
|
||||
{
|
||||
var workflowInstance = workflowInstanceGroup.Key;
|
||||
|
||||
var workflowDefinition = await workflowRegistry.GetWorkflowDefinitionAsync(
|
||||
workflowInstance.DefinitionId,
|
||||
VersionOptions.SpecificVersion(workflowInstance.Version),
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, workflowInstance);
|
||||
|
||||
foreach (var activity in workflowInstanceGroup)
|
||||
{
|
||||
var executionContext = await ExecuteAsync(
|
||||
workflow,
|
||||
true,
|
||||
new[] { activity.Item2.Id },
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
executionContexts.Add(executionContext);
|
||||
}
|
||||
}
|
||||
|
||||
return executionContexts;
|
||||
}
|
||||
|
||||
private async Task FinalizeWorkflowExecutionAsync(
|
||||
WorkflowExecutionContext workflowExecutionContext,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (!workflowExecutionContext.Workflow.BlockingActivities.Any() &&
|
||||
workflowExecutionContext.Workflow.IsExecuting())
|
||||
{
|
||||
workflowExecutionContext.Finish();
|
||||
}
|
||||
else
|
||||
{
|
||||
// Notify event handlers that halting activities are about to be executed.
|
||||
await workflowEventHandlers.InvokeAsync(
|
||||
async x => await x.InvokingHaltedActivitiesAsync(workflowExecutionContext, cancellationToken),
|
||||
logger
|
||||
);
|
||||
|
||||
// Invoke Halted event on activity drivers that halted the workflow.
|
||||
while (workflowExecutionContext.HasScheduledHaltingActivities)
|
||||
{
|
||||
var currentActivity = workflowExecutionContext.PopScheduledHaltingActivity();
|
||||
var result = await ExecuteActivityHaltedAsync(
|
||||
workflowExecutionContext,
|
||||
currentActivity,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
await result.ExecuteAsync(this, workflowExecutionContext, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
// Notify event handlers that workflow execution has ended.
|
||||
await workflowEventHandlers.InvokeAsync(
|
||||
async x => await x.WorkflowInvokedAsync(workflowExecutionContext, cancellationToken),
|
||||
logger
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<IActivityExecutionResult> ExecuteActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
async () => await activityInvoker.ExecuteAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<IActivityExecutionResult> ResumeActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
async () => await activityInvoker.ResumeAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<IActivityExecutionResult> InvokeActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
Func<Task<IActivityExecutionResult>> executeAction,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
workflowContext.Workflow.Status = WorkflowStatus.Aborted;
|
||||
workflowContext.Workflow.FinishedAt = clock.GetCurrentInstant();
|
||||
return null;
|
||||
}
|
||||
|
||||
return await executeAction();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
FaultWorkflow(workflowContext, activity, ex);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private async Task<IActivityExecutionResult> ExecuteActivityHaltedAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
async () => await activityInvoker.HaltedAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private void FaultWorkflow(WorkflowExecutionContext workflowContext, IActivity activity, Exception ex)
|
||||
{
|
||||
logger.LogError(
|
||||
ex,
|
||||
"An unhandled error occurred while executing an activity. Putting the workflow in the faulted state."
|
||||
);
|
||||
workflowContext.Fault(activity, ex);
|
||||
}
|
||||
|
||||
private async Task<WorkflowExecutionContext> CreateWorkflowExecutionContextAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var workflowExecutionContext = new WorkflowExecutionContext(workflow, clock, serviceProvider);
|
||||
var startActivityList = startActivities?.ToList() ?? workflow.GetStartActivities().Take(1).ToList();
|
||||
|
||||
foreach (var startActivity in startActivityList)
|
||||
{
|
||||
if (await startActivity.CanExecuteAsync(workflowExecutionContext, cancellationToken))
|
||||
workflowExecutionContext.ScheduleActivity(startActivity);
|
||||
}
|
||||
|
||||
if (workflowExecutionContext.HasScheduledActivities)
|
||||
{
|
||||
workflow.BlockingActivities.RemoveWhere(startActivityList.Contains);
|
||||
|
||||
if (workflowExecutionContext.Workflow.Status == WorkflowStatus.Idle)
|
||||
workflowExecutionContext.Start();
|
||||
}
|
||||
|
||||
return workflowExecutionContext;
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<(WorkflowDefinitionVersion, ActivityDefinition)>> FilterRunningSingletonsAsync(
|
||||
IEnumerable<(WorkflowDefinitionVersion, ActivityDefinition)> workflowDefinitions,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var definitions = workflowDefinitions.ToList();
|
||||
var transients = definitions.Where(x => !x.Item1.IsSingleton).ToList();
|
||||
var singletons = definitions.Where(x => x.Item1.IsSingleton).ToList();
|
||||
var result = transients.ToList();
|
||||
|
||||
foreach (var definition in singletons)
|
||||
{
|
||||
var instances = await workflowInstanceStore.ListByStatusAsync(
|
||||
definition.Item1.DefinitionId,
|
||||
WorkflowStatus.Executing,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
if (!instances.Any())
|
||||
{
|
||||
result.Add(definition);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
}
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Models;
|
||||
using Elsa.Persistence;
|
||||
using Elsa.Services.Extensions;
|
||||
using Elsa.Services.Models;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Newtonsoft.Json.Linq;
|
||||
using NodaTime;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
internal class WorkflowRunner : IWorkflowRunner
|
||||
{
|
||||
private readonly IActivityInvoker activityInvoker;
|
||||
private readonly IWorkflowFactory workflowFactory;
|
||||
private readonly IWorkflowRegistry workflowRegistry;
|
||||
private readonly IWorkflowInstanceStore workflowInstanceStore;
|
||||
private readonly IEnumerable<IWorkflowEventHandler> workflowEventHandlers;
|
||||
private readonly IClock clock;
|
||||
private readonly IServiceProvider serviceProvider;
|
||||
private readonly ILogger logger;
|
||||
|
||||
public WorkflowRunner(
|
||||
IActivityInvoker activityInvoker,
|
||||
IWorkflowFactory workflowFactory,
|
||||
IWorkflowRegistry workflowRegistry,
|
||||
IWorkflowInstanceStore workflowInstanceStore,
|
||||
IEnumerable<IWorkflowEventHandler> workflowEventHandlers,
|
||||
IClock clock,
|
||||
IServiceProvider serviceProvider,
|
||||
ILogger<WorkflowRunner> logger)
|
||||
{
|
||||
this.activityInvoker = activityInvoker;
|
||||
this.workflowFactory = workflowFactory;
|
||||
this.workflowRegistry = workflowRegistry;
|
||||
this.workflowInstanceStore = workflowInstanceStore;
|
||||
this.workflowEventHandlers = workflowEventHandlers;
|
||||
this.clock = clock;
|
||||
this.serviceProvider = serviceProvider;
|
||||
this.logger = logger;
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> RunAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return ExecuteAsync(workflow, false, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> RunAsync(
|
||||
WorkflowDefinitionVersion workflowDefinition,
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, correlationId: correlationId);
|
||||
var startActivities = workflow.Activities.Find(startActivityIds);
|
||||
|
||||
return ExecuteAsync(workflow, false, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> RunAsync<T>(
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default) where T : IWorkflow, new()
|
||||
{
|
||||
var workflow = workflowFactory.CreateWorkflow<T>(input, correlationId: correlationId);
|
||||
var startActivities = workflow.Activities.Find(startActivityIds);
|
||||
|
||||
return ExecuteAsync(workflow, false, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> ResumeAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return ExecuteAsync(workflow, true, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> ResumeAsync<T>(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = null,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default) where T : IWorkflow, new()
|
||||
{
|
||||
var workflow = workflowFactory.CreateWorkflow<T>(input, workflowInstance);
|
||||
var startActivities = workflow.Activities.Find(startActivityIds);
|
||||
return ExecuteAsync(workflow, true, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<WorkflowExecutionContext> ResumeAsync(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = null,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var definition = await workflowRegistry.GetWorkflowDefinitionAsync(
|
||||
workflowInstance.DefinitionId,
|
||||
VersionOptions.SpecificVersion(workflowInstance.Version),
|
||||
cancellationToken);
|
||||
var workflow = workflowFactory.CreateWorkflow(definition, input, workflowInstance);
|
||||
return await ExecuteAsync(workflow, true, startActivityIds, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<WorkflowExecutionContext>> TriggerAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var startedExecutionContexts = await RunManyAsync(
|
||||
activityType,
|
||||
input,
|
||||
correlationId,
|
||||
activityStatePredicate,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
var resumedExecutionContexts = await ResumeManyAsync(
|
||||
activityType,
|
||||
input,
|
||||
correlationId,
|
||||
activityStatePredicate,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
return startedExecutionContexts.Concat(resumedExecutionContexts);
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<WorkflowExecutionContext>> ResumeManyAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowInstances = await workflowInstanceStore
|
||||
.ListByBlockingActivityAsync(activityType, correlationId, cancellationToken)
|
||||
.ToListAsync();
|
||||
|
||||
if (activityStatePredicate != null)
|
||||
workflowInstances = workflowInstances.Where(x => activityStatePredicate(x.Item2.State)).ToList();
|
||||
|
||||
return await ResumeManyAsync(
|
||||
workflowInstances,
|
||||
input,
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<WorkflowExecutionContext>> RunManyAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowDefinitions = await workflowRegistry.ListByStartActivityAsync(activityType, cancellationToken);
|
||||
|
||||
if (activityStatePredicate != null)
|
||||
workflowDefinitions = workflowDefinitions.Where(x => activityStatePredicate(x.Item2.State));
|
||||
|
||||
workflowDefinitions = await FilterRunningSingletonsAsync(
|
||||
workflowDefinitions,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
return await RunManyAsync(workflowDefinitions, input, correlationId, cancellationToken);
|
||||
}
|
||||
|
||||
private Task<WorkflowExecutionContext> ExecuteAsync(
|
||||
Workflow workflow,
|
||||
bool resume,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var startActivities = startActivityIds != null
|
||||
? workflow.Activities.Find(startActivityIds)
|
||||
: Enumerable.Empty<IActivity>();
|
||||
|
||||
return ExecuteAsync(workflow, resume, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<WorkflowExecutionContext> ExecuteAsync(
|
||||
Workflow workflow,
|
||||
bool resume,
|
||||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowExecutionContext = await CreateWorkflowExecutionContextAsync(
|
||||
workflow,
|
||||
startActivities,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
var start = !resume;
|
||||
|
||||
while (workflowExecutionContext.HasScheduledActivities)
|
||||
{
|
||||
var currentActivity = workflowExecutionContext.PopScheduledActivity();
|
||||
|
||||
var result = start
|
||||
? await ExecuteActivityAsync(workflowExecutionContext, currentActivity, cancellationToken)
|
||||
: await ResumeActivityAsync(workflowExecutionContext, currentActivity, cancellationToken);
|
||||
|
||||
if (result == null)
|
||||
break;
|
||||
|
||||
await result.ExecuteAsync(this, workflowExecutionContext, cancellationToken);
|
||||
|
||||
workflowExecutionContext.IsFirstPass = false;
|
||||
start = true;
|
||||
}
|
||||
|
||||
await FinalizeWorkflowExecutionAsync(workflowExecutionContext, cancellationToken);
|
||||
|
||||
return workflowExecutionContext;
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<WorkflowExecutionContext>> RunManyAsync(
|
||||
IEnumerable<(WorkflowDefinitionVersion, ActivityDefinition)> workflowDefinitions,
|
||||
Variables input,
|
||||
string correlationId,
|
||||
CancellationToken cancellationToken1)
|
||||
{
|
||||
var executionContexts = new List<WorkflowExecutionContext>();
|
||||
|
||||
foreach (var (workflowDefinition, activityDefinition) in workflowDefinitions)
|
||||
{
|
||||
var startActivityIds = workflowDefinition.Activities
|
||||
.Where(x => x.Id == activityDefinition.Id)
|
||||
.Select(x => x.Id);
|
||||
|
||||
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, correlationId: correlationId);
|
||||
|
||||
var executionContext = await ExecuteAsync(
|
||||
workflow,
|
||||
false,
|
||||
startActivityIds,
|
||||
cancellationToken1
|
||||
);
|
||||
executionContexts.Add(executionContext);
|
||||
}
|
||||
|
||||
return executionContexts;
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<WorkflowExecutionContext>> ResumeManyAsync(
|
||||
IEnumerable<(WorkflowInstance, ActivityInstance)> workflowInstances,
|
||||
Variables input,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var executionContexts = new List<WorkflowExecutionContext>();
|
||||
var workflowInstanceGroups = workflowInstances.GroupBy(x => x.Item1);
|
||||
|
||||
foreach (var workflowInstanceGroup in workflowInstanceGroups)
|
||||
{
|
||||
var workflowInstance = workflowInstanceGroup.Key;
|
||||
|
||||
var workflowDefinition = await workflowRegistry.GetWorkflowDefinitionAsync(
|
||||
workflowInstance.DefinitionId,
|
||||
VersionOptions.SpecificVersion(workflowInstance.Version),
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, workflowInstance);
|
||||
|
||||
foreach (var activity in workflowInstanceGroup)
|
||||
{
|
||||
var executionContext = await ExecuteAsync(
|
||||
workflow,
|
||||
true,
|
||||
new[] { activity.Item2.Id },
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
executionContexts.Add(executionContext);
|
||||
}
|
||||
}
|
||||
|
||||
return executionContexts;
|
||||
}
|
||||
|
||||
private async Task FinalizeWorkflowExecutionAsync(
|
||||
WorkflowExecutionContext workflowExecutionContext,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (!workflowExecutionContext.Workflow.BlockingActivities.Any() &&
|
||||
workflowExecutionContext.Workflow.IsExecuting())
|
||||
{
|
||||
workflowExecutionContext.Finish();
|
||||
}
|
||||
else
|
||||
{
|
||||
// Notify event handlers that halting activities are about to be executed.
|
||||
await workflowEventHandlers.InvokeAsync(
|
||||
async x => await x.InvokingHaltedActivitiesAsync(workflowExecutionContext, cancellationToken),
|
||||
logger
|
||||
);
|
||||
|
||||
// Invoke Halted event on activity drivers that halted the workflow.
|
||||
while (workflowExecutionContext.HasScheduledHaltingActivities)
|
||||
{
|
||||
var currentActivity = workflowExecutionContext.PopScheduledHaltingActivity();
|
||||
var result = await ExecuteActivityHaltedAsync(
|
||||
workflowExecutionContext,
|
||||
currentActivity,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
await result.ExecuteAsync(this, workflowExecutionContext, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
// Notify event handlers that workflow execution has ended.
|
||||
await workflowEventHandlers.InvokeAsync(
|
||||
async x => await x.WorkflowInvokedAsync(workflowExecutionContext, cancellationToken),
|
||||
logger
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<IActivityExecutionResult> ExecuteActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
async () => await activityInvoker.ExecuteAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<IActivityExecutionResult> ResumeActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
async () => await activityInvoker.ResumeAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<IActivityExecutionResult> InvokeActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
Func<Task<IActivityExecutionResult>> executeAction,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
workflowContext.Workflow.Status = WorkflowStatus.Cancelled;
|
||||
workflowContext.Workflow.FinishedAt = clock.GetCurrentInstant();
|
||||
return null;
|
||||
}
|
||||
|
||||
return await executeAction();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
FaultWorkflow(workflowContext, activity, ex);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private async Task<IActivityExecutionResult> ExecuteActivityHaltedAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
async () => await activityInvoker.HaltedAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private void FaultWorkflow(WorkflowExecutionContext workflowContext, IActivity activity, Exception ex)
|
||||
{
|
||||
logger.LogError(
|
||||
ex,
|
||||
"An unhandled error occurred while executing an activity. Putting the workflow in the faulted state."
|
||||
);
|
||||
workflowContext.Fault(activity, ex);
|
||||
}
|
||||
|
||||
private async Task<WorkflowExecutionContext> CreateWorkflowExecutionContextAsync(
|
||||
Workflow workflow,
|
||||
IEnumerable<IActivity> startActivities,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var workflowExecutionContext = new WorkflowExecutionContext(workflow, clock, serviceProvider);
|
||||
var startActivityList = startActivities?.ToList() ?? workflow.GetStartActivities().Take(1).ToList();
|
||||
|
||||
foreach (var startActivity in startActivityList)
|
||||
{
|
||||
if (await startActivity.CanExecuteAsync(workflowExecutionContext, cancellationToken))
|
||||
workflowExecutionContext.ScheduleActivity(startActivity);
|
||||
}
|
||||
|
||||
if (workflowExecutionContext.HasScheduledActivities)
|
||||
{
|
||||
workflow.BlockingActivities.RemoveWhere(startActivityList.Contains);
|
||||
|
||||
if (workflowExecutionContext.Workflow.Status == WorkflowStatus.Pending)
|
||||
workflowExecutionContext.Start();
|
||||
}
|
||||
|
||||
return workflowExecutionContext;
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<(WorkflowDefinitionVersion, ActivityDefinition)>> FilterRunningSingletonsAsync(
|
||||
IEnumerable<(WorkflowDefinitionVersion, ActivityDefinition)> workflowDefinitions,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var definitions = workflowDefinitions.ToList();
|
||||
var transients = definitions.Where(x => !x.Item1.IsSingleton).ToList();
|
||||
var singletons = definitions.Where(x => x.Item1.IsSingleton).ToList();
|
||||
var result = transients.ToList();
|
||||
|
||||
foreach (var definition in singletons)
|
||||
{
|
||||
var instances = await workflowInstanceStore.ListByStatusAsync(
|
||||
definition.Item1.DefinitionId,
|
||||
WorkflowStatus.Running,
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
if (!instances.Any())
|
||||
{
|
||||
result.Add(definition);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -189,10 +189,10 @@ namespace Elsa.Dashboard.Areas.Elsa.Controllers
|
|||
return new WorkflowDefinitionListItemModel
|
||||
{
|
||||
WorkflowDefinition = workflowDefinition,
|
||||
AbortedCount = instances.Count(x => x.Status == WorkflowStatus.Aborted),
|
||||
AbortedCount = instances.Count(x => x.Status == WorkflowStatus.Cancelled),
|
||||
FaultedCount = instances.Count(x => x.Status == WorkflowStatus.Faulted),
|
||||
FinishedCount = instances.Count(x => x.Status == WorkflowStatus.Finished),
|
||||
ExecutingCount = instances.Count(x => x.Status == WorkflowStatus.Executing),
|
||||
FinishedCount = instances.Count(x => x.Status == WorkflowStatus.Completed),
|
||||
ExecutingCount = instances.Count(x => x.Status == WorkflowStatus.Running),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -76,7 +76,7 @@
|
|||
asp-action="Index"
|
||||
asp-controller="WorkflowInstance"
|
||||
asp-route-definitionId="@latest.WorkflowDefinition.DefinitionId"
|
||||
asp-route-status="Executing">
|
||||
asp-route-status="Running">
|
||||
@($"{latest.ExecutingCount} Executing")
|
||||
</a>
|
||||
</span>
|
||||
|
|
@ -106,7 +106,7 @@
|
|||
asp-action="Index"
|
||||
asp-controller="WorkflowInstance"
|
||||
asp-route-definitionId="@latest.WorkflowDefinition.DefinitionId"
|
||||
asp-route-status="Finished">
|
||||
asp-route-status="Completed">
|
||||
@($"{latest.FinishedCount} Finished")
|
||||
</a>
|
||||
</span>
|
||||
|
|
@ -121,7 +121,7 @@
|
|||
asp-action="Index"
|
||||
asp-controller="WorkflowInstance"
|
||||
asp-route-definitionId="@latest.WorkflowDefinition.DefinitionId"
|
||||
asp-route-status="Aborted">
|
||||
asp-route-status="Cancelled">
|
||||
@($"{latest.AbortedCount} Aborted")
|
||||
</a>
|
||||
</span>
|
||||
|
|
|
|||
|
|
@ -7,10 +7,10 @@ namespace Elsa.Dashboard.Extensions
|
|||
public static string GetStatusClass(this WorkflowStatus workflowStatus) =>
|
||||
workflowStatus switch
|
||||
{
|
||||
WorkflowStatus.Executing => "bg-info",
|
||||
WorkflowStatus.Running => "bg-info",
|
||||
WorkflowStatus.Faulted => "bg-warning",
|
||||
WorkflowStatus.Finished => "bg-success",
|
||||
WorkflowStatus.Aborted => "bg-info",
|
||||
WorkflowStatus.Completed => "bg-success",
|
||||
WorkflowStatus.Cancelled => "bg-info",
|
||||
_ => "bg-default",
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -69,7 +69,7 @@ namespace Elsa.Persistence.DocumentDb.Services
|
|||
|
||||
var query = client
|
||||
.CreateDocumentQuery<WorkflowInstanceDocument>(storage.CollectionUri)
|
||||
.Where(x => x.Status == WorkflowStatus.Executing);
|
||||
.Where(x => x.Status == WorkflowStatus.Running);
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(correlationId))
|
||||
{
|
||||
|
|
|
|||
|
|
@ -118,7 +118,7 @@ namespace Elsa.Persistence.EntityFrameworkCore.Services
|
|||
.Include(x => x.BlockingActivities)
|
||||
.AsQueryable();
|
||||
|
||||
query = query.Where(x => x.Status == WorkflowStatus.Executing);
|
||||
query = query.Where(x => x.Status == WorkflowStatus.Running);
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(correlationId))
|
||||
query = query.Where(x => x.CorrelationId == correlationId);
|
||||
|
|
|
|||
|
|
@ -70,7 +70,7 @@ namespace Elsa.Persistence.MongoDb.Services
|
|||
{
|
||||
var query = collection.AsQueryable();
|
||||
|
||||
query = query.Where(x => x.Status == WorkflowStatus.Executing);
|
||||
query = query.Where(x => x.Status == WorkflowStatus.Running);
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(correlationId))
|
||||
query = query.Where(x => x.CorrelationId == correlationId);
|
||||
|
|
|
|||
|
|
@ -78,7 +78,7 @@ namespace Elsa.Persistence.YesSql.Services
|
|||
{
|
||||
var query = session.Query<WorkflowInstanceDocument, WorkflowInstanceBlockingActivitiesIndex>();
|
||||
|
||||
query = query.Where(x => x.WorkflowStatus == WorkflowStatus.Executing);
|
||||
query = query.Where(x => x.WorkflowStatus == WorkflowStatus.Running);
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(correlationId))
|
||||
query = query.Where(x => x.CorrelationId == correlationId);
|
||||
|
|
|
|||
|
|
@ -21,8 +21,8 @@ namespace Sample01
|
|||
.BuildServiceProvider();
|
||||
|
||||
// Invoke the workflow.
|
||||
var invoker = services.GetService<IWorkflowInvoker>();
|
||||
await invoker.StartAsync<HelloWorldWorkflow>();
|
||||
var invoker = services.GetService<IWorkflowRunner>();
|
||||
await invoker.RunAsync<HelloWorldWorkflow>();
|
||||
|
||||
Console.ReadLine();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,8 +30,8 @@ namespace Sample02
|
|||
.Build();
|
||||
|
||||
// Start the workflow.
|
||||
var invoker = services.GetService<IWorkflowInvoker>();
|
||||
await invoker.StartAsync(workflowDefinition);
|
||||
var invoker = services.GetService<IWorkflowRunner>();
|
||||
await invoker.RunAsync(workflowDefinition);
|
||||
|
||||
Console.ReadLine();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -39,8 +39,8 @@ namespace Sample03
|
|||
};
|
||||
|
||||
// Run the workflow.
|
||||
var invoker = services.GetService<IWorkflowInvoker>();
|
||||
await invoker.StartAsync(workflowDefinition);
|
||||
var invoker = services.GetService<IWorkflowRunner>();
|
||||
await invoker.RunAsync(workflowDefinition);
|
||||
|
||||
Console.ReadLine();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -31,8 +31,8 @@ namespace Sample04
|
|||
var workflow = workflowFactory.CreateWorkflow<CalculatorWorkflow>();
|
||||
|
||||
// Start the workflow.
|
||||
var invoker = services.GetService<IWorkflowInvoker>();
|
||||
await invoker.StartAsync(workflow);
|
||||
var invoker = services.GetService<IWorkflowRunner>();
|
||||
await invoker.RunAsync(workflow);
|
||||
|
||||
Console.WriteLine("Workflow has ended. Here are the activities that have executed:");
|
||||
foreach (var logEntry in workflow.ExecutionLog)
|
||||
|
|
|
|||
|
|
@ -22,9 +22,9 @@ namespace Sample09
|
|||
var services = BuildServices();
|
||||
var serializer = services.GetRequiredService<IWorkflowSerializer>();
|
||||
var workflow = serializer.Deserialize<WorkflowDefinitionVersion>(json, JsonTokenFormatter.FormatName);
|
||||
var invoker = services.GetRequiredService<IWorkflowInvoker>();
|
||||
var invoker = services.GetRequiredService<IWorkflowRunner>();
|
||||
|
||||
await invoker.StartAsync(workflow);
|
||||
await invoker.RunAsync(workflow);
|
||||
}
|
||||
|
||||
private static IServiceProvider BuildServices()
|
||||
|
|
|
|||
|
|
@ -50,8 +50,8 @@ namespace Sample10
|
|||
VersionOptions.Latest);
|
||||
|
||||
// Execute the workflow.
|
||||
var invoker = scope.ServiceProvider.GetRequiredService<IWorkflowInvoker>();
|
||||
var executionContext = await invoker.StartAsync(workflowDefinition);
|
||||
var invoker = scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
|
||||
var executionContext = await invoker.RunAsync(workflowDefinition);
|
||||
|
||||
// Persist the workflow instance.
|
||||
var instanceStore = scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
|
||||
|
|
|
|||
|
|
@ -20,13 +20,13 @@ namespace Sample11
|
|||
var services = BuildServices();
|
||||
var registry = services.GetService<IWorkflowRegistry>();
|
||||
var workflowDefinition = await registry.GetWorkflowDefinitionAsync<CorrelationWorkflow>();
|
||||
var invoker = services.GetRequiredService<IWorkflowInvoker>();
|
||||
var invoker = services.GetRequiredService<IWorkflowRunner>();
|
||||
|
||||
Console.WriteLine("How many workflow instances should be started? Enter a number:");
|
||||
var instanceCount = int.Parse(Console.ReadLine());
|
||||
|
||||
for (var i = 0; i < instanceCount; i++)
|
||||
await invoker.StartAsync(workflowDefinition, correlationId: $"document {i + 1}");
|
||||
await invoker.RunAsync(workflowDefinition, correlationId: $"document {i + 1}");
|
||||
|
||||
var retry = true;
|
||||
|
||||
|
|
|
|||
|
|
@ -22,11 +22,11 @@ namespace Sample12
|
|||
var services = BuildServices();
|
||||
var registry = services.GetService<IWorkflowRegistry>();
|
||||
var workflowDefinition = await registry.GetWorkflowDefinitionAsync<UserTaskWorkflow>();
|
||||
var invoker = services.GetRequiredService<IWorkflowInvoker>();
|
||||
var invoker = services.GetRequiredService<IWorkflowRunner>();
|
||||
|
||||
// Start the workflow.
|
||||
var correlationId = Guid.NewGuid().ToString("N");
|
||||
await invoker.StartAsync(workflowDefinition, correlationId: correlationId);
|
||||
await invoker.RunAsync(workflowDefinition, correlationId: correlationId);
|
||||
WorkflowExecutionContext executionContext;
|
||||
|
||||
do
|
||||
|
|
|
|||
|
|
@ -27,8 +27,8 @@ namespace Sample13
|
|||
var workflow = workflowFactory.CreateWorkflow<ActivityOutputWorkflow>();
|
||||
|
||||
// Start the workflow.
|
||||
var invoker = services.GetService<IWorkflowInvoker>();
|
||||
await invoker.StartAsync(workflow);
|
||||
var invoker = services.GetService<IWorkflowRunner>();
|
||||
await invoker.RunAsync(workflow);
|
||||
|
||||
Console.ReadLine();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -54,8 +54,8 @@ namespace Sample14
|
|||
VersionOptions.Latest);
|
||||
|
||||
// Execute the workflow.
|
||||
var invoker = scope.ServiceProvider.GetRequiredService<IWorkflowInvoker>();
|
||||
var executionContext = await invoker.StartAsync(workflowDefinition);
|
||||
var invoker = scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
|
||||
var executionContext = await invoker.RunAsync(workflowDefinition);
|
||||
|
||||
// Persist the workflow instance.
|
||||
var instanceStore = scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
|
||||
|
|
|
|||
|
|
@ -41,8 +41,8 @@ namespace Sample15
|
|||
VersionOptions.Latest);
|
||||
|
||||
// Execute the workflow.
|
||||
var invoker = scope.ServiceProvider.GetRequiredService<IWorkflowInvoker>();
|
||||
var executionContext = await invoker.StartAsync(workflowDefinition);
|
||||
var invoker = scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
|
||||
var executionContext = await invoker.RunAsync(workflowDefinition);
|
||||
|
||||
// Persist the workflow instance.
|
||||
var instanceStore = scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
|
||||
|
|
|
|||
|
|
@ -13,8 +13,8 @@ namespace Sample17
|
|||
private static async Task Main(string[] args)
|
||||
{
|
||||
var services = BuildServices();
|
||||
var workflowInvoker = services.GetRequiredService<IWorkflowInvoker>();
|
||||
var workflowExecutionContext = await workflowInvoker.StartAsync<CreatePersonWorkflow>();
|
||||
var workflowInvoker = services.GetRequiredService<IWorkflowRunner>();
|
||||
var workflowExecutionContext = await workflowInvoker.RunAsync<CreatePersonWorkflow>();
|
||||
var document = workflowExecutionContext.CurrentScope.GetVariable<Person>("Document");
|
||||
|
||||
Console.WriteLine("Created document: {0}", document);
|
||||
|
|
|
|||
|
|
@ -31,8 +31,8 @@ namespace Sample18
|
|||
{
|
||||
await host.StartAsync();
|
||||
|
||||
var workflowInvoker = host.Services.GetRequiredService<IWorkflowInvoker>();
|
||||
await workflowInvoker.StartAsync<EmailReminderWorkflow>();
|
||||
var workflowInvoker = host.Services.GetRequiredService<IWorkflowRunner>();
|
||||
await workflowInvoker.RunAsync<EmailReminderWorkflow>();
|
||||
|
||||
await host.WaitForShutdownAsync();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -29,9 +29,9 @@ namespace Sample20
|
|||
.BuildServiceProvider();
|
||||
|
||||
// Invoke the workflows.
|
||||
var invoker = services.GetService<IWorkflowInvoker>();
|
||||
var executionContext1 = await invoker.StartAsync<ExecuteMethodWorkflow>();
|
||||
var executionContext2 = await invoker.StartAsync<SplitObjectWorkflow>();
|
||||
var invoker = services.GetService<IWorkflowRunner>();
|
||||
var executionContext1 = await invoker.RunAsync<ExecuteMethodWorkflow>();
|
||||
var executionContext2 = await invoker.RunAsync<SplitObjectWorkflow>();
|
||||
|
||||
var serializer = services.GetRequiredService<IWorkflowSerializer>();
|
||||
|
||||
|
|
|
|||
|
|
@ -9,12 +9,12 @@ namespace Elsa.Server.GraphQL.Mutations
|
|||
{
|
||||
public class RunWorkflow : IMutationProvider
|
||||
{
|
||||
private readonly IWorkflowInvoker workflowInvoker;
|
||||
private readonly IWorkflowRunner workflowRunner;
|
||||
private readonly IWorkflowDefinitionStore workflowDefinitionStore;
|
||||
|
||||
public RunWorkflow(IWorkflowInvoker workflowInvoker, IWorkflowDefinitionStore workflowDefinitionStore)
|
||||
public RunWorkflow(IWorkflowRunner workflowRunner, IWorkflowDefinitionStore workflowDefinitionStore)
|
||||
{
|
||||
this.workflowInvoker = workflowInvoker;
|
||||
this.workflowRunner = workflowRunner;
|
||||
this.workflowDefinitionStore = workflowDefinitionStore;
|
||||
}
|
||||
|
||||
|
|
@ -36,7 +36,7 @@ namespace Elsa.Server.GraphQL.Mutations
|
|||
return null;
|
||||
|
||||
var correlationId = context.GetArgument<string>("correlationId");
|
||||
var executionContext = await workflowInvoker.StartAsync(workflowDefinition, correlationId: correlationId, cancellationToken: context.CancellationToken);
|
||||
var executionContext = await workflowRunner.RunAsync(workflowDefinition, correlationId: correlationId, cancellationToken: context.CancellationToken);
|
||||
return executionContext.Workflow.ToInstance();
|
||||
});
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue