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

Implemented workflow correlation (#54)

Browse source
This commit is contained in:
Sipke Schoorstra 2019-07-25 17:13:18 +02:00 committed by GitHub
parent 26c70d4e0f
commit 1a70cfd18c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
29 changed files with 422 additions and 169 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
Samples.sln
View file

@ -59,6 +59,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sample10", "samples\Sample1
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.AutoMapper.Extensions", "src\core\Elsa.AutoMapper.Extensions\Elsa.AutoMapper.Extensions.csproj", "{F7633BE5-0335-4148-8FB7-EAED9A2276A2}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sample11", "samples\Sample11\Sample11.csproj", "{0C3BB425-2FED-4ADF-8DBE-D68DF0CA5A41}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -141,6 +143,10 @@ Global
{F7633BE5-0335-4148-8FB7-EAED9A2276A2}.Debug|Any CPU.Build.0 = Debug|Any CPU
{F7633BE5-0335-4148-8FB7-EAED9A2276A2}.Release|Any CPU.ActiveCfg = Release|Any CPU
{F7633BE5-0335-4148-8FB7-EAED9A2276A2}.Release|Any CPU.Build.0 = Release|Any CPU
{0C3BB425-2FED-4ADF-8DBE-D68DF0CA5A41}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{0C3BB425-2FED-4ADF-8DBE-D68DF0CA5A41}.Debug|Any CPU.Build.0 = Debug|Any CPU
{0C3BB425-2FED-4ADF-8DBE-D68DF0CA5A41}.Release|Any CPU.ActiveCfg = Release|Any CPU
{0C3BB425-2FED-4ADF-8DBE-D68DF0CA5A41}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@ -168,6 +174,7 @@ Global
{905C1067-2119-443D-B5FE-DA8E027FAD62} = {600C4AE0-0585-4181-9CEC-2BB43430C713}
{E0076858-353F-4302-9690-920204E38BF6} = {5E5E1E84-DDBC-40D6-B891-0D563A15A44A}
{F7633BE5-0335-4148-8FB7-EAED9A2276A2} = {35F44BE9-13D0-417C-A01A-F0787BEE6DC3}
{0C3BB425-2FED-4ADF-8DBE-D68DF0CA5A41} = {5E5E1E84-DDBC-40D6-B891-0D563A15A44A}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {8B0975FD-7050-48B0-88C5-48C33378E158}

4
samples/Sample07/DocumentApprovalWorkflow.cs
View file

@ -59,12 +59,12 @@ namespace Sample07
{
fork
.When("Approve")
.Then<SignalEvent>(activity => activity.SignalName = "approve")
.Then<SignalEvent>(activity => activity.Signal = new PlainTextExpression("approve"))
.Then("join-signals");
fork
.When("Reject")
.Then<SignalEvent>(activity => activity.SignalName = "reject")
.Then<SignalEvent>(activity => activity.Signal = new PlainTextExpression("reject"))
.Then("join-signals");
}
)

20
samples/Sample11/CorrelationWorkflow.cs Normal file
View file

@ -0,0 +1,20 @@
using Elsa.Activities.Console.Activities;
using Elsa.Activities.Primitives;
using Elsa.Expressions;
using Elsa.Services;
using Elsa.Services.Models;
namespace Sample11
{
public class CorrelationWorkflow : IWorkflow
{
public void Build(IWorkflowBuilder builder)
{
builder
.StartWith<WriteLine>(activity => activity.TextExpression = new JavaScriptExpression<string>("`Workflow started with correlation ID \"${correlationId()}\".`"))
.Then<SignalEvent>(activity => activity.Signal = new PlainTextExpression("Proceed"))
.Then<WriteLine>(activity => activity.TextExpression = new JavaScriptExpression<string>("`Signal received for workflow with correlation ID: \"${correlationId()}\"`"))
.Then<WriteLine>(activity => activity.TextExpression = new PlainTextExpression("Workflow finished."));
}
}
}

64
samples/Sample11/Program.cs Normal file
View file

@ -0,0 +1,64 @@
using System;
using System.Linq;
using System.Threading.Tasks;
using Elsa.Activities.Console.Extensions;
using Elsa.Activities.Primitives;
using Elsa.Extensions;
using Elsa.Models;
using Elsa.Persistence.Memory;
using Elsa.Runtime;
using Elsa.Services;
using Microsoft.Extensions.DependencyInjection;
namespace Sample11
{
/// <summary>
/// Demonstrates workflow correlation.
/// </summary>
public class Program
{
public static async Task Main(string[] args)
{
var services = BuildServices();
var registry = services.GetService<IWorkflowRegistry>();
var workflowDefinition = registry.RegisterWorkflow<CorrelationWorkflow>();
var invoker = services.GetRequiredService<IWorkflowInvoker>();
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.InvokeAsync(workflowDefinition, correlationId: $"document {i + 1}");
}
var retry = true;
while (retry)
{
Console.WriteLine();
Console.WriteLine("Now, enter the correlation ID of the workflow to resume:");
var correlationId = Console.ReadLine();
// Resume one workflow using the specified correlation ID.
var triggeredExecutionContexts = (await invoker.TriggerAsync(nameof(SignalEvent), new Variables { ["Signal"] = "Proceed" }, correlationId)).ToList();
Console.WriteLine("{0} workflow was resumed. Would you like to trigger another?", triggeredExecutionContexts.Count);
retry = string.Equals("y", Console.ReadLine(), StringComparison.OrdinalIgnoreCase);
}
Console.WriteLine("Bye!");
}
private static IServiceProvider BuildServices()
{
return new ServiceCollection()
.AddWorkflows()
.AddStartupRunner()
.AddConsoleActivities()
.AddMemoryWorkflowDefinitionStore()
.AddMemoryWorkflowInstanceStore()
.BuildServiceProvider();
}
}
}

13
samples/Sample11/Sample11.csproj Normal file
View file

@ -0,0 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>netcoreapp2.2</TargetFramework>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\activities\Elsa.Activities.Console\Elsa.Activities.Console.csproj" />
<ProjectReference Include="..\..\src\core\Elsa.Core\Elsa.Core.csproj" />
</ItemGroup>
</Project>

30
src/activities/Elsa.Activities.Http/Activities/SignalEvent.cs
View file

@ -1,30 +0,0 @@
using Elsa.Results;
using Elsa.Services;
using Elsa.Services.Models;
namespace Elsa.Activities.Http.Activities
{
public class SignalEvent : Activity
{
public string SignalName
{
get => GetState<string>();
set => SetState(value);
}
protected override bool OnCanExecute(WorkflowExecutionContext context)
{
return context.Workflow.Input.ContainsKey("signal") && (string) context.Workflow.Input["signal"] == SignalName;
}
protected override ActivityExecutionResult OnExecute(WorkflowExecutionContext context)
{
return Halt();
}
protected override ActivityExecutionResult OnResume(WorkflowExecutionContext context)
{
return Done();
}
}
}

3
src/activities/Elsa.Activities.Http/Extensions/ServiceCollectionExtensions.cs
View file

@ -22,8 +22,7 @@ namespace Elsa.Activities.Http.Extensions
services
.AddActivity<HttpRequestEvent>()
.AddActivity<HttpResponseTask>()
.AddActivity<HttpRequestAction>()
.AddActivity<SignalEvent>();
.AddActivity<HttpRequestAction>();
services
.AddSingleton<ISharedAccessSignatureService, SharedAccessSignatureService>()

5
src/activities/Elsa.Activities.Http/RequestHandlers/Handlers/SignalRequestHandler.cs
View file

@ -6,6 +6,7 @@ using CSharpFunctionalExtensions;
using Elsa.Activities.Http.Activities;
using Elsa.Activities.Http.Models;
using Elsa.Activities.Http.Services;
using Elsa.Activities.Primitives;
using Elsa.Extensions;
using Elsa.Models;
using Elsa.Persistence;
@ -94,12 +95,12 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
var input = new Variables
{
["signal"] = signal.Name
["Signal"] = signal.Name
};
var workflowDefinition = workflowRegistry.GetById(workflowInstance.DefinitionId);
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, workflowInstance);
var blockingSignalActivities = workflow.BlockingActivities.Where(x => x is SignalEvent).Cast<SignalEvent>().Where(x => x.SignalName == signal.Name).ToList();
var blockingSignalActivities = workflow.BlockingActivities.ToList();
await workflowInvoker.ResumeAsync(workflow, blockingSignalActivities, cancellationToken);
if (!httpContext.Response.HasStarted)

2
src/activities/Elsa.Activities.Http/RequestHandlers/Handlers/TriggerRequestHandler.cs
View file

@ -41,7 +41,7 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
var requestPath = new Uri(httpContext.Request.Path.ToString(), UriKind.Relative);
var method = httpContext.Request.Method;
var workflowsToStart = Filter(registry.ListByStartActivity(nameof(HttpRequestEvent)), requestPath, method).ToList();
var workflowsToResume = Filter(await workflowInstanceStore.ListByBlockingActivityAsync<HttpRequestEvent>(cancellationToken), requestPath, method).ToList();
var workflowsToResume = Filter(await workflowInstanceStore.ListByBlockingActivityAsync<HttpRequestEvent>(cancellationToken: cancellationToken), requestPath, method).ToList();
if (!workflowsToStart.Any() && !workflowsToResume.Any())
{

62
src/activities/Elsa.Activities.MassTransit/Consumers/WorkflowConsumer.cs
View file

@ -1,31 +1,33 @@
using System.Threading.Tasks;
using Elsa.Activities.MassTransit.Activities;
using Elsa.Models;
using Elsa.Services;
using MassTransit;
namespace Elsa.Activities.MassTransit.Consumers
{
public class WorkflowConsumer<T> : IConsumer<T> where T : class
{
private readonly IWorkflowInvoker workflowInvoker;
public WorkflowConsumer(IWorkflowInvoker workflowInvoker)
{
this.workflowInvoker = workflowInvoker;
}
public async Task Consume(ConsumeContext<T> context)
{
var message = context.Message;
var activityType = nameof(ReceiveMassTransitMessage);
var input = new Variables { ["message"] = message };
await workflowInvoker.TriggerAsync(
activityType,
input,
x => ReceiveMassTransitMessage.GetMessageType(x) == message.GetType(),
context.CancellationToken);
}
}
using System.Threading.Tasks;
using Elsa.Activities.MassTransit.Activities;
using Elsa.Models;
using Elsa.Services;
using MassTransit;
namespace Elsa.Activities.MassTransit.Consumers
{
public class WorkflowConsumer<T> : IConsumer<T> where T : class
{
private readonly IWorkflowInvoker workflowInvoker;
public WorkflowConsumer(IWorkflowInvoker workflowInvoker)
{
this.workflowInvoker = workflowInvoker;
}
public async Task Consume(ConsumeContext<T> context)
{
var message = context.Message;
var activityType = nameof(ReceiveMassTransitMessage);
var input = new Variables { ["message"] = message };
var correlationId = context.CorrelationId?.ToString();
await workflowInvoker.TriggerAsync(
activityType,
input,
correlationId,
x => ReceiveMassTransitMessage.GetMessageType(x) == message.GetType(),
context.CancellationToken);
}
}
}

22
src/core/Elsa.Abstractions/Extensions/WorkflowInstanceStoreExtensions.cs Normal file
View file

@ -0,0 +1,22 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Services.Models;
namespace Elsa.Extensions
{
public static class WorkflowInstanceStoreExtensions
{
public static async Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync<TActivity>(
this IWorkflowInstanceStore store,
string correlationId = default,
CancellationToken cancellationToken = default) where TActivity : IActivity
{
var items = await store.ListByBlockingActivityAsync(nameof(TActivity), correlationId, cancellationToken);
return items.Select(x => (x.Item1, x.Item2));
}
}
}

70
src/core/Elsa.Abstractions/Models/Variables.cs
View file

@ -1,31 +1,41 @@
using System.Collections.Generic;
namespace Elsa.Models
{
public class Variables : Dictionary<string, object>
{
public static readonly Variables Empty = new Variables();
public Variables()
{
}
public Variables(Variables other)
{
foreach (var variable in other)
{
this[variable.Key] = variable.Value;
}
}
public object GetVariable(string name)
{
return ContainsKey(name) ? this[name] : null;
}
public T GetVariable<T>(string name)
{
return ContainsKey(name) ? (T)this[name] : default(T);
}
}
using System.Collections.Generic;
namespace Elsa.Models
{
public class Variables : Dictionary<string, object>
{
public static readonly Variables Empty = new Variables();
public Variables()
{
}
public Variables(Variables other)
{
foreach (var variable in other)
{
this[variable.Key] = variable.Value;
}
}
public object GetVariable(string name)
{
return ContainsKey(name) ? this[name] : null;
}
public T GetVariable<T>(string name)
{
return ContainsKey(name) ? (T)this[name] : default(T);
}
public bool HasVariable(string name, object value)
{
return GetVariable(name) == value;
}
public bool HasVariable(string name)
{
return ContainsKey(name);
}
}
}

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

@ -8,6 +8,7 @@ namespace Elsa.Models
public string Id { get; set; }
public string DefinitionId { get; set; }
public WorkflowStatus Status { get; set; }
public string CorrelationId { get; set; }
public Instant CreatedAt { get; set; }
public Instant? StartedAt { get; set; }
public Instant? HaltedAt { get; set; }

14
src/core/Elsa.Abstractions/Persistence/IWorkflowInstanceStore.cs
View file

@ -1,9 +1,7 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Models;
using Elsa.Services.Models;
using WorkflowInstance = Elsa.Models.WorkflowInstance;
namespace Elsa.Persistence
@ -12,19 +10,11 @@ namespace Elsa.Persistence
{
Task SaveAsync(WorkflowInstance instance, CancellationToken cancellationToken = default);
Task<WorkflowInstance> GetByIdAsync(string id, CancellationToken cancellationToken = default);
Task<WorkflowInstance> GetByCorrelationIdAsync(string correlationId, CancellationToken cancellationToken = default);
Task<IEnumerable<WorkflowInstance>> ListByDefinitionAsync(string definitionId, CancellationToken cancellationToken = default);
Task<IEnumerable<WorkflowInstance>> ListAllAsync(CancellationToken cancellationToken = default);
Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync(string activityType, CancellationToken cancellationToken = default);
Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync(string activityType, string correlationId = default, CancellationToken cancellationToken = default);
Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(string definitionId, WorkflowStatus status, CancellationToken cancellationToken = default);
Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(WorkflowStatus status, CancellationToken cancellationToken = default);
}
public static class WorkflowInstanceStoreExtensions
{
public static async Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync<TActivity>(this IWorkflowInstanceStore store, CancellationToken cancellationToken) where TActivity : IActivity
{
var items = await store.ListByBlockingActivityAsync(nameof(TActivity), cancellationToken);
return items.Select(x => (x.Item1, x.Item2));
}
}
}

25
src/core/Elsa.Abstractions/Services/IWorkflowInvoker.cs
View file

@ -15,35 +15,36 @@ namespace Elsa.Services
IEnumerable<IActivity> startActivityIds = default,
CancellationToken cancellationToken = default
);
Task<WorkflowExecutionContext> InvokeAsync(
WorkflowDefinition workflowDefinition,
Variables input = null,
WorkflowInstance workflowInstance = null,
Variables input = default,
WorkflowInstance workflowInstance = default,
IEnumerable<string> startActivityIds = default,
string correlationId = default,
CancellationToken cancellationToken = default
);
Task<WorkflowExecutionContext> InvokeAsync<T>(
WorkflowInstance workflowInstance = null,
Variables input = null,
WorkflowInstance workflowInstance = default,
Variables input = default,
IEnumerable<string> startActivityIds = default,
CancellationToken cancellationToken = default
) where T:IWorkflow, new();
) where T : IWorkflow, new();
Task<WorkflowExecutionContext> InvokeAsync(
WorkflowInstance workflowInstance,
Variables input = null,
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 TriggerAsync(
string activityType,
Variables input,
Task<IEnumerable<WorkflowExecutionContext>> TriggerAsync(string activityType,
Variables input = default,
string correlationId = default,
Func<JObject, bool> activityStatePredicate = default,
CancellationToken cancellationToken = default);
}

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

@ -1,4 +1,5 @@
using System.Collections.Generic;
using System;
using System.Collections.Generic;
using System.Linq;
using Elsa.Comparers;
using Elsa.Models;
@ -14,15 +15,14 @@ namespace Elsa.Services.Models
string definitionId,
IEnumerable<IActivity> activities,
IEnumerable<Connection> connections,
Variables input = null,
WorkflowInstance workflowInstance = null) : this()
Variables input = default,
string correlationId = default) : this()
{
Id = id;
DefinitionId = definitionId;
Activities = activities.ToList();
Connections = connections.ToList();
Input = new Variables(input ?? Variables.Empty);
Initialize(workflowInstance);
}
public Workflow()
@ -34,6 +34,7 @@ namespace Elsa.Services.Models
public string Id { get; set; }
public string DefinitionId { get; }
public string CorrelationId { get; set; }
public WorkflowStatus Status { get; set; }
public Instant CreatedAt { get; set; }
public Instant? StartedAt { get; set; }
@ -55,6 +56,7 @@ namespace Elsa.Services.Models
{
Id = Id,
DefinitionId = DefinitionId,
CorrelationId = CorrelationId,
Status = Status,
CreatedAt = CreatedAt,
StartedAt = StartedAt,
@ -68,14 +70,15 @@ namespace Elsa.Services.Models
};
}
private void Initialize(WorkflowInstance instance)
public void Initialize(WorkflowInstance instance)
{
if(instance == null)
return;
throw new ArgumentNullException(nameof(instance));
var activityLookup = Activities.ToDictionary(x => x.Id);
Id = instance.Id;
CorrelationId = instance.CorrelationId;
Status = instance.Status;
CreatedAt = instance.CreatedAt;
StartedAt = instance.StartedAt;

36
src/core/Elsa.Core/Activities/Primitives/Correlate.cs Normal file
View file

@ -0,0 +1,36 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Expressions;
using Elsa.Extensions;
using Elsa.Results;
using Elsa.Services;
using Elsa.Services.Models;
namespace Elsa.Activities.Primitives
{
/// <summary>
/// Sets the CorrelationId of the workflow to a given value.
/// </summary>
public class Correlate : Activity
{
private readonly IWorkflowExpressionEvaluator expressionEvaluator;
public Correlate(IWorkflowExpressionEvaluator expressionEvaluator)
{
this.expressionEvaluator = expressionEvaluator;
}
public WorkflowExpression<string> Expression
{
get => GetState<WorkflowExpression<string>>();
set => SetState(value);
}
protected override async Task<ActivityExecutionResult> OnExecuteAsync(WorkflowExecutionContext workflowContext, CancellationToken cancellationToken)
{
var value = await expressionEvaluator.EvaluateAsync(Expression, workflowContext, cancellationToken);
workflowContext.Workflow.CorrelationId = value?.ToString();
return Done();
}
}
}

45
src/core/Elsa.Core/Activities/Primitives/SignalEvent.cs Normal file
View file

@ -0,0 +1,45 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Expressions;
using Elsa.Extensions;
using Elsa.Results;
using Elsa.Services;
using Elsa.Services.Models;
namespace Elsa.Activities.Primitives
{
/// <summary>
/// Halts workflow execution until the specified signal is received.
/// </summary>
public class SignalEvent : Activity
{
private readonly IWorkflowExpressionEvaluator expressionEvaluator;
public SignalEvent(IWorkflowExpressionEvaluator expressionEvaluator)
{
this.expressionEvaluator = expressionEvaluator;
}
public WorkflowExpression<string> Signal
{
get => GetState<WorkflowExpression<string>>();
set => SetState(value);
}
protected override async Task<bool> OnCanExecuteAsync(WorkflowExecutionContext context, CancellationToken cancellationToken)
{
var signal = await expressionEvaluator.EvaluateAsync(Signal, context, cancellationToken);
return context.Workflow.Input.HasVariable("Signal", signal);
}
protected override ActivityExecutionResult OnExecute(WorkflowExecutionContext context)
{
return Halt(true);
}
protected override ActivityExecutionResult OnResume(WorkflowExecutionContext context)
{
return Done();
}
}
}

11
src/core/Elsa.Core/Extensions/ServiceCollectionExtensions.cs
View file

@ -2,7 +2,6 @@ using System;
using Elsa.Activities.ControlFlow;
using Elsa.Activities.Primitives;
using Elsa.Expressions;
using Elsa.Persistence;
using Elsa.Scripting;
using Elsa.Serialization;
using Elsa.Serialization.Formatters;
@ -49,7 +48,8 @@ namespace Elsa.Extensions
.AddTransient<IWorkflowBuilder, WorkflowBuilder>()
.AddSingleton<Func<IWorkflowBuilder>>(sp => sp.GetRequiredService<IWorkflowBuilder>)
.AddSingleton<IWorkflowRegistry, WorkflowRegistry>()
.AddPrimitiveActivities();
.AddPrimitiveActivities()
.AddControlFlowActivities();
}
public static IServiceCollection AddActivity<T>(this IServiceCollection services)
@ -64,6 +64,13 @@ namespace Elsa.Extensions
{
return services
.AddActivity<SetVariable>()
.AddActivity<Correlate>()
.AddActivity<SignalEvent>();
}
private static IServiceCollection AddControlFlowActivities(this IServiceCollection services)
{
return services
.AddActivity<ForEach>()
.AddActivity<Fork>()
.AddActivity<Join>()

1
src/core/Elsa.Core/Extensions/WorkflowInstanceCollectionExtensions.cs
View file

@ -1,7 +1,6 @@
using System.Collections.Generic;
using System.Linq;
using Elsa.Models;
using Elsa.Services.Models;
namespace Elsa.Extensions
{

18
src/core/Elsa.Core/Persistence/Memory/MemoryWorkflowInstanceStore.cs
View file

@ -1,4 +1,3 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
@ -24,6 +23,12 @@ namespace Elsa.Persistence.Memory
return Task.FromResult(instance);
}
public Task<WorkflowInstance> GetByCorrelationIdAsync(string correlationId, CancellationToken cancellationToken = default)
{
var instance = workflowInstances.Values.FirstOrDefault(x => x.CorrelationId == correlationId);
return Task.FromResult(instance);
}
public Task<IEnumerable<WorkflowInstance>> ListByDefinitionAsync(string definitionId, CancellationToken cancellationToken)
{
var workflows = workflowInstances.Values.Where(x => x.DefinitionId == definitionId);
@ -36,21 +41,26 @@ namespace Elsa.Persistence.Memory
return Task.FromResult(workflows);
}
public Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync(string activityType, CancellationToken cancellationToken)
public Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync(string activityType, string correlationId = default, CancellationToken cancellationToken = default)
{
var query = workflowInstances.Values.GetBlockingActivities().Where(x => x.Item2.TypeName == activityType);
if (!string.IsNullOrWhiteSpace(correlationId))
query = query.Where(x => x.Item1.CorrelationId == correlationId);
return Task.FromResult(query);
}
public Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(string definitionId, WorkflowStatus status, CancellationToken cancellationToken)
{
throw new NotImplementedException();
var query = workflowInstances.Values.Where(x => x.DefinitionId == definitionId && x.Status == status);
return Task.FromResult(query);
}
public Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(WorkflowStatus status, CancellationToken cancellationToken)
{
throw new NotImplementedException();
var query = workflowInstances.Values.Where(x => x.Status == status);
return Task.FromResult(query);
}
}
}

1
src/core/Elsa.Core/Scripting/CommonScriptEngineConfigurator.cs
View file

@ -14,6 +14,7 @@ namespace Elsa.Scripting
engine.SetValue("input", (Func<string, object>) (name => workflow.Input.GetVariable(name)));
engine.SetValue("variable", (Func<string, object>) (name => context.CurrentScope.GetVariable(name)));
engine.SetValue("lastResult", (Func<string, object>) (name => context.CurrentScope.LastResult));
engine.SetValue("correlationId", (Func<string, object>) (name => context.Workflow.CorrelationId));
foreach (var variable in workflowExecutionContext.CurrentScope.Variables)
{

7
src/core/Elsa.Core/Services/WorkflowFactory.cs
View file

@ -32,7 +32,12 @@ namespace Elsa.Services
var activities = CreateActivities(definition.Activities).ToList();
var connections = CreateConnections(definition.Connections, activities);
var id = idGenerator.Generate();
return new Workflow(id, definition.Id, activities, connections, input, workflowInstance);
var workflow = new Workflow(id, definition.Id, activities, connections, input);
if(workflowInstance != null)
workflow.Initialize(workflowInstance);
return workflow;
}
private IEnumerable<Connection> CreateConnections(IEnumerable<ConnectionDefinition> connectionBlueprints, IEnumerable<IActivity> activities)

80
src/core/Elsa.Core/Services/WorkflowInvoker.cs
View file

@ -60,20 +60,23 @@ namespace Elsa.Services
public Task<WorkflowExecutionContext> InvokeAsync(
WorkflowDefinition workflowDefinition,
Variables input = null,
WorkflowInstance workflowInstance = null,
IEnumerable<string> startActivityIds = default,
Variables input = default,
WorkflowInstance workflowInstance = default,
IEnumerable<string> startActivityIds = default,
string correlationId = default,
CancellationToken cancellationToken = default)
{
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, workflowInstance);
var startActivities = workflow.Activities.Find(startActivityIds);
workflow.CorrelationId = correlationId;
return InvokeAsync(workflow, startActivities, cancellationToken);
}
public Task<WorkflowExecutionContext> InvokeAsync<T>(
WorkflowInstance workflowInstance = null,
Variables input = null,
IEnumerable<string> startActivityIds = default,
IEnumerable<string> startActivityIds = default,
CancellationToken cancellationToken = default) where T : IWorkflow, new()
{
var workflow = workflowFactory.CreateWorkflow<T>(input, workflowInstance);
@ -81,19 +84,23 @@ namespace Elsa.Services
return InvokeAsync(workflow, startActivities, cancellationToken);
}
public Task<WorkflowExecutionContext> InvokeAsync(WorkflowInstance workflowInstance, Variables input = null, IEnumerable<string> startActivityIds = default, CancellationToken cancellationToken = default)
public Task<WorkflowExecutionContext> InvokeAsync(
WorkflowInstance workflowInstance,
Variables input = null,
IEnumerable<string> startActivityIds = default,
CancellationToken cancellationToken = default)
{
var definition = workflowRegistry.GetById(workflowInstance.DefinitionId);
return InvokeAsync(definition, input, workflowInstance, startActivityIds, cancellationToken);
return InvokeAsync(definition, input, workflowInstance, startActivityIds, workflowInstance.CorrelationId, cancellationToken);
}
public async Task TriggerAsync(
string activityType,
Variables input,
public async Task<IEnumerable<WorkflowExecutionContext>> TriggerAsync(string activityType,
Variables input = default,
string correlationId = default,
Func<JObject, bool> activityStatePredicate = default,
CancellationToken cancellationToken = default)
{
var workflowInstances = await workflowInstanceStore.ListByBlockingActivityAsync(activityType, cancellationToken).ToListAsync();
var workflowInstances = await workflowInstanceStore.ListByBlockingActivityAsync(activityType, correlationId, cancellationToken).ToListAsync();
var workflowDefinitions = workflowRegistry.ListByStartActivity(activityType).ToList();
if (activityStatePredicate != null)
@ -102,29 +109,55 @@ namespace Elsa.Services
if (activityStatePredicate != null)
workflowInstances = workflowInstances.Where(x => activityStatePredicate(x.Item2.State)).ToList();
await StartWorkflowsAsync(workflowDefinitions, input, cancellationToken);
await ResumeWorkflowsAsync(workflowInstances, input, cancellationToken);
var startedExecutionContexts = await StartWorkflowsAsync(workflowDefinitions, input, correlationId, cancellationToken);
var resumedExecutionContexts = await ResumeWorkflowsAsync(workflowInstances, input, correlationId, cancellationToken);
return startedExecutionContexts.Concat(resumedExecutionContexts);
}
private async Task StartWorkflowsAsync(IEnumerable<(WorkflowDefinition, ActivityDefinition)> workflowDefinitions, Variables variables, CancellationToken cancellationToken1)
private async Task<IEnumerable<WorkflowExecutionContext>> StartWorkflowsAsync(
IEnumerable<(WorkflowDefinition, ActivityDefinition)> workflowDefinitions,
Variables variables,
string correlationId,
CancellationToken cancellationToken1)
{
var executionContexts = new List<WorkflowExecutionContext>();
foreach (var (workflowDefinition, activityDefinition) in workflowDefinitions)
{
var startActivities = workflowDefinition.Activities.Where(x => x.Id == activityDefinition.Id).Select(x => x.Id);
await InvokeAsync(workflowDefinition, variables, startActivityIds: startActivities, cancellationToken: cancellationToken1);
var executionContext = await InvokeAsync(
workflowDefinition,
variables,
startActivityIds: startActivities,
correlationId: correlationId,
cancellationToken: cancellationToken1
);
executionContexts.Add(executionContext);
}
return executionContexts;
}
private async Task ResumeWorkflowsAsync(IEnumerable<(WorkflowInstance, ActivityInstance)> workflowInstances, Variables input, CancellationToken cancellationToken)
private async Task<IEnumerable<WorkflowExecutionContext>> ResumeWorkflowsAsync(
IEnumerable<(WorkflowInstance, ActivityInstance)> workflowInstances,
Variables input,
string correlationId,
CancellationToken cancellationToken)
{
var executionContexts = new List<WorkflowExecutionContext>();
foreach (var (workflowInstance, startActivityInstance) in workflowInstances)
{
var workflowDefinition = workflowRegistry.GetById(workflowInstance.DefinitionId);
workflowInstance.Status = WorkflowStatus.Resuming;
await InvokeAsync(workflowDefinition, input, workflowInstance, new[] { startActivityInstance.Id }, cancellationToken);
var executionContext = await InvokeAsync(workflowDefinition, input, workflowInstance, new[] { startActivityInstance.Id }, correlationId, cancellationToken);
executionContexts.Add(executionContext);
}
return executionContexts;
}
private async Task ExecuteWorkflowAsync(WorkflowExecutionContext workflowExecutionContext, CancellationToken cancellationToken)
@ -210,7 +243,7 @@ namespace Elsa.Services
return null;
}
private async Task<ActivityExecutionResult> ExecuteActivityHaltedAsync(WorkflowExecutionContext workflowContext, IActivity activity, CancellationToken cancellationToken)
{
return await ExecuteActivityAsync(workflowContext, activity, () => activityInvoker.HaltedAsync(workflowContext, activity, cancellationToken), cancellationToken);
@ -222,11 +255,14 @@ namespace Elsa.Services
workflowContext.Fault(activity, ex);
}
private async Task<WorkflowExecutionContext> CreateWorkflowExecutionContextAsync(Workflow workflow, IEnumerable<IActivity> startActivities, CancellationToken cancellationToken)
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();
await workflowExecutionContext.ScheduleActivitiesAsync(startActivityList);
if (workflowExecutionContext.HasScheduledActivities)

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

@ -10,6 +10,7 @@ namespace Elsa.Persistence.YesSql.Documents
public string WorkflowInstanceId { get; set; }
public string DefinitionId { get; set; }
public WorkflowStatus Status { get; set; }
public string CorrelationId { get; set; }
public Instant CreatedAt { get; set; }
public Instant? StartedAt { get; set; }
public Instant? HaltedAt { get; set; }

6
src/persistence/Elsa.Persistence.YesSql/Extensions/StoreFactory.cs
View file

@ -1,13 +1,7 @@
using System;
using System.Data;
using System.Linq;
using System.Reflection;
using Elsa.Persistence.YesSql.Options;
using Elsa.Persistence.YesSql.Services;
using Elsa.Persistence.YesSql.StartupTasks;
using Elsa.Runtime;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Options;
using YesSql;
using YesSql.Indexes;

7
src/persistence/Elsa.Persistence.YesSql/Indexes/WorkflowInstanceIndex.cs
View file

@ -1,7 +1,5 @@
using System.Linq;
using Elsa.Models;
using Elsa.Services.Extensions;
using YesSql;
using YesSql.Indexes;
namespace Elsa.Persistence.YesSql.Indexes
@ -10,6 +8,7 @@ namespace Elsa.Persistence.YesSql.Indexes
{
public string WorkflowInstanceId { get; set; }
public string WorkflowDefinitionId { get; set; }
public string CorrelationId { get; set; }
public WorkflowStatus WorkflowStatus { get; set; }
}
@ -28,7 +27,8 @@ namespace Elsa.Persistence.YesSql.Indexes
workflowInstance => new WorkflowInstanceIndex
{
WorkflowDefinitionId = workflowInstance.Id,
WorkflowStatus = workflowInstance.Status
WorkflowStatus = workflowInstance.Status,
CorrelationId = workflowInstance.CorrelationId
}
);
@ -40,6 +40,7 @@ namespace Elsa.Persistence.YesSql.Indexes
{
WorkflowInstanceId = workflowInstance.Id,
WorkflowDefinitionId = workflowInstance.Id,
CorrelationId = workflowInstance.CorrelationId,
ActivityId = activity.ActivityId,
ActivityType = activity.ActivityType
}

19
src/persistence/Elsa.Persistence.YesSql/Services/YesSqlWorkflowInstanceStore.cs
View file

@ -1,4 +1,3 @@
using System.Collections;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
@ -41,6 +40,15 @@ namespace Elsa.Persistence.YesSql.Services
}
}
public async Task<WorkflowInstance> GetByCorrelationIdAsync(string correlationId, CancellationToken cancellationToken = default)
{
using (var session = sessionProvider.GetSession())
{
var document = await session.Query<WorkflowInstanceDocument, WorkflowInstanceIndex>(x => x.CorrelationId == correlationId).FirstOrDefaultAsync();
return mapper.Map<WorkflowInstance>(document);
}
}
public async Task<IEnumerable<WorkflowInstance>> ListByDefinitionAsync(string definitionId, CancellationToken cancellationToken)
{
using (var session = sessionProvider.GetSession())
@ -59,11 +67,16 @@ namespace Elsa.Persistence.YesSql.Services
}
}
public async Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync(string activityType, CancellationToken cancellationToken)
public async Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync(string activityType, string correlationId = default, CancellationToken cancellationToken = default)
{
using (var session = sessionProvider.GetSession())
{
var documents = await session.Query<WorkflowInstanceDocument, WorkflowInstanceBlockingActivitiesIndex>(x => x.ActivityType == activityType).ListAsync();
var query = session.Query<WorkflowInstanceDocument, WorkflowInstanceBlockingActivitiesIndex>(x => x.ActivityType == activityType);
if (string.IsNullOrWhiteSpace(correlationId))
query = query.Where(x => x.CorrelationId == correlationId);
var documents = await query.ListAsync();
var instances = mapper.Map<IEnumerable<WorkflowInstance>>(documents);
return instances.GetBlockingActivities();

2
src/persistence/Elsa.Persistence.YesSql/StartupTasks/StoreInitializationTask.cs
View file

@ -42,11 +42,13 @@ namespace Elsa.Persistence.YesSql.StartupTasks
.CreateMapIndexTable(nameof(WorkflowInstanceIndex), table => table
.Column<string>("WorkflowInstanceId")
.Column<string>("WorkflowDefinitionId")
.Column<string>("CorrelationId")
.Column<string>("WorkflowStatus")
)
.CreateMapIndexTable(nameof(WorkflowInstanceBlockingActivitiesIndex), table => table
.Column<string>("WorkflowInstanceId")
.Column<string>("WorkflowDefinitionId")
.Column<string>("CorrelationId")
.Column<string>("WorkflowStatus")
.Column<string>("ActivityId")
.Column<string>("ActivityType")