Fix object disposed bug (#116)
* Refactor buggy SignalRequestHandler
For some reason, the following code is problematic:
await DecryptToken()
.BindAsync(GetWorkflowInstanceAsync)
.BindAsync(CheckIfExecutingAsync)
.MapAsync(ResumeWorkflowAsync);
Basically, any lambda in the BindAsync and MapAsync could execute even after the containing method returned
* Implement better message handler registration without adding multiple IMediator services
* Abstracted Mediator service registration methods.
* Add Liquid support for JObject.
This commit is contained in:
Sipke Schoorstra
GitHub
parent
48777d8eb3
commit
57f116c239
|
|
@ -27,7 +27,6 @@
|
|||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="LanguageExt.Core" Version="3.3.29" />
|
||||
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="3.0.0" />
|
||||
<PackageReference Include="Microsoft.AspNetCore.Http" Version="2.2.2" />
|
||||
<PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.2.0" />
|
||||
|
|
|
|||
|
|
@ -1,12 +1,12 @@
|
|||
using System;
|
||||
using Elsa.Activities.Http.Activities;
|
||||
using Elsa.Activities.Http.Formatters;
|
||||
using Elsa.Activities.Http.Liquid;
|
||||
using Elsa.Activities.Http.Options;
|
||||
using Elsa.Activities.Http.RequestHandlers.Handlers;
|
||||
using Elsa.Activities.Http.Services;
|
||||
using Elsa.Scripting;
|
||||
using Elsa.Scripting.JavaScript;
|
||||
using MediatR;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Scripting.Liquid.Extensions;
|
||||
using Microsoft.AspNetCore.Http;
|
||||
using Microsoft.AspNetCore.Mvc.Infrastructure;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
|
@ -38,11 +38,12 @@ namespace Elsa.Activities.Http.Extensions
|
|||
.AddSingleton<IActionContextAccessor, ActionContextAccessor>()
|
||||
.AddSingleton<IAbsoluteUrlProvider, DefaultAbsoluteUrlProvider>()
|
||||
.AddHttpContextAccessor()
|
||||
.AddMediatR(typeof(HttpActivitiesServiceCollectionExtensions))
|
||||
.AddNotificationHandlers(typeof(HttpActivitiesServiceCollectionExtensions))
|
||||
.AddDataProtection();
|
||||
|
||||
|
||||
services.AddLiquidFilter<SignalUrlFilter>("signal_url");
|
||||
|
||||
return services
|
||||
.AddScoped(sp => sp.GetRequiredService<IHttpContextAccessor>().HttpContext)
|
||||
.AddRequestHandler<TriggerRequestHandler>()
|
||||
.AddRequestHandler<SignalRequestHandler>();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,46 @@
|
|||
using System;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Activities.Http.Models;
|
||||
using Elsa.Activities.Http.Services;
|
||||
using Elsa.Scripting.Liquid.Services;
|
||||
using Elsa.Services.Models;
|
||||
using Fluid;
|
||||
using Fluid.Values;
|
||||
|
||||
namespace Elsa.Activities.Http.Liquid
|
||||
{
|
||||
public class SignalUrlFilter : ILiquidFilter
|
||||
{
|
||||
private readonly ITokenService tokenService;
|
||||
private readonly IAbsoluteUrlProvider absoluteUrlProvider;
|
||||
|
||||
public SignalUrlFilter(ITokenService tokenService, IAbsoluteUrlProvider absoluteUrlProvider)
|
||||
{
|
||||
this.tokenService = tokenService;
|
||||
this.absoluteUrlProvider = absoluteUrlProvider;
|
||||
}
|
||||
|
||||
public ValueTask<FluidValue> ProcessAsync(FluidValue input, FilterArguments arguments, TemplateContext context)
|
||||
{
|
||||
var workflowContextValue = context.GetValue("WorkflowExecutionContext");
|
||||
|
||||
if (workflowContextValue.IsNil())
|
||||
throw new ArgumentException("WorkflowExecutionContext missing while invoking 'signal_url'");
|
||||
|
||||
var workflowContext = (WorkflowExecutionContext)workflowContextValue.ToObjectValue();
|
||||
var signalName = input.ToStringValue();
|
||||
var url = GenerateUrl(signalName, workflowContext);
|
||||
return new ValueTask<FluidValue>(new StringValue(url));
|
||||
}
|
||||
|
||||
private string GenerateUrl(string signal, WorkflowExecutionContext workflowExecutionContext)
|
||||
{
|
||||
var workflowInstanceId = workflowExecutionContext.Workflow.Id;
|
||||
var payload = new Signal(signal, workflowInstanceId);
|
||||
var token = tokenService.CreateToken(payload);
|
||||
var url = $"/workflows/signal?token={token}";
|
||||
|
||||
return absoluteUrlProvider.ToAbsoluteUrl(url).ToString();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -3,12 +3,11 @@ using System.Net;
|
|||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Activities.Http.Models;
|
||||
using Elsa.Activities.Http.RequestHandlers.Results;
|
||||
using Elsa.Activities.Http.Services;
|
||||
using Elsa.Models;
|
||||
using Elsa.Persistence;
|
||||
using Elsa.Services;
|
||||
using LanguageExt;
|
||||
using LanguageExt.Common;
|
||||
using Microsoft.AspNetCore.Http;
|
||||
|
||||
namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
||||
|
|
@ -42,58 +41,39 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
|
||||
public async Task<IRequestHandlerResult> HandleRequestAsync()
|
||||
{
|
||||
await DecryptToken()
|
||||
.BindAsync(GetWorkflowInstanceAsync)
|
||||
.BindAsync(CheckIfExecutingAsync)
|
||||
.MapAsync(ResumeWorkflowAsync);
|
||||
var signal = DecryptToken();
|
||||
|
||||
return default;
|
||||
if (signal == null)
|
||||
return new NotFoundResult();
|
||||
|
||||
var workflowInstance = await GetWorkflowInstanceAsync(signal);
|
||||
|
||||
if (workflowInstance == null)
|
||||
return new NotFoundResult();
|
||||
|
||||
if (!CheckIfExecuting(workflowInstance))
|
||||
return new BadRequestResult($"Cannot signal a workflow with status other than {WorkflowStatus.Executing}. Actual workflow status: {workflowInstance.Status}.");
|
||||
|
||||
await ResumeWorkflowAsync(workflowInstance, signal);
|
||||
|
||||
return new AcceptedResult();
|
||||
}
|
||||
|
||||
private Either<Error, Signal> DecryptToken()
|
||||
private Signal DecryptToken()
|
||||
{
|
||||
var token = httpContext.Request.Query["token"];
|
||||
|
||||
if (tokenService.TryDecryptToken(token, out Signal signal))
|
||||
{
|
||||
return signal;
|
||||
}
|
||||
|
||||
httpContext.Response.StatusCode = (int)HttpStatusCode.NotFound;
|
||||
return Error.New("Invalid token");
|
||||
return tokenService.TryDecryptToken(token, out Signal signal) ? signal : default;
|
||||
}
|
||||
|
||||
private async Task<Either<Error, (WorkflowInstance, Signal)>> GetWorkflowInstanceAsync(Signal signal)
|
||||
private async Task<WorkflowInstance> GetWorkflowInstanceAsync(Signal signal) =>
|
||||
await workflowInstanceStore.GetByIdAsync(signal.WorkflowInstanceId, cancellationToken);
|
||||
|
||||
private bool CheckIfExecuting(WorkflowInstance workflowInstance) =>
|
||||
workflowInstance.Status == WorkflowStatus.Executing;
|
||||
|
||||
private async Task ResumeWorkflowAsync(WorkflowInstance workflowInstance, Signal signal)
|
||||
{
|
||||
var workflowInstance =
|
||||
await workflowInstanceStore.GetByIdAsync(signal.WorkflowInstanceId, cancellationToken);
|
||||
|
||||
if (workflowInstance != null)
|
||||
return (workflowInstance, signal);
|
||||
|
||||
httpContext.Response.StatusCode = (int)HttpStatusCode.NotFound;
|
||||
return Error.New("Workflow not found");
|
||||
}
|
||||
|
||||
private async Task<Either<Error, (WorkflowInstance, Signal)>> CheckIfExecutingAsync(
|
||||
(WorkflowInstance, Signal) tuple)
|
||||
{
|
||||
var (workflowInstance, signal) = tuple;
|
||||
|
||||
if (workflowInstance.Status == WorkflowStatus.Executing)
|
||||
return (workflowInstance, signal);
|
||||
|
||||
httpContext.Response.StatusCode = (int)HttpStatusCode.BadRequest;
|
||||
await httpContext.Response.WriteAsync(
|
||||
$"Cannot signal a workflow with status other than {WorkflowStatus.Executing}. Actual workflow status: {workflowInstance.Status}.",
|
||||
cancellationToken);
|
||||
return Error.New("Cannot resume workflow that is not executing.");
|
||||
}
|
||||
|
||||
private async Task ResumeWorkflowAsync((WorkflowInstance, Signal) tuple)
|
||||
{
|
||||
var (workflowInstance, signal) = tuple;
|
||||
|
||||
var input = new Variables
|
||||
{
|
||||
["Signal"] = signal.Name
|
||||
|
|
@ -107,11 +87,6 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
var workflow = workflowFactory.CreateWorkflow(workflowDefinition, input, workflowInstance);
|
||||
var blockingSignalActivities = workflow.BlockingActivities.ToList();
|
||||
await workflowInvoker.ResumeAsync(workflow, blockingSignalActivities, cancellationToken);
|
||||
|
||||
if (!httpContext.Response.HasStarted)
|
||||
{
|
||||
httpContext.Response.StatusCode = (int)HttpStatusCode.Accepted;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -24,16 +24,16 @@ namespace Elsa.Activities.Http.RequestHandlers.Handlers
|
|||
private readonly CancellationToken cancellationToken;
|
||||
|
||||
public TriggerRequestHandler(
|
||||
HttpContext httpContext,
|
||||
IHttpContextAccessor httpContext,
|
||||
IWorkflowInvoker workflowInvoker,
|
||||
IWorkflowRegistry registry,
|
||||
IWorkflowInstanceStore workflowInstanceStore)
|
||||
{
|
||||
this.httpContext = httpContext;
|
||||
this.httpContext = httpContext.HttpContext;
|
||||
this.workflowInvoker = workflowInvoker;
|
||||
this.registry = registry;
|
||||
this.workflowInstanceStore = workflowInstanceStore;
|
||||
cancellationToken = httpContext.RequestAborted;
|
||||
cancellationToken = httpContext.HttpContext.RequestAborted;
|
||||
}
|
||||
|
||||
public async Task<IRequestHandlerResult> HandleRequestAsync()
|
||||
|
|
|
|||
|
|
@ -7,10 +7,24 @@ namespace Elsa.Activities.Http.RequestHandlers.Results
|
|||
{
|
||||
public class BadRequestResult : IRequestHandlerResult
|
||||
{
|
||||
public Task ExecuteResultAsync(HttpContext httpContext, RequestDelegate next)
|
||||
public BadRequestResult()
|
||||
{
|
||||
}
|
||||
|
||||
public BadRequestResult(string message)
|
||||
{
|
||||
Message = message;
|
||||
}
|
||||
|
||||
public string Message { get; }
|
||||
|
||||
|
||||
public async Task ExecuteResultAsync(HttpContext httpContext, RequestDelegate next)
|
||||
{
|
||||
httpContext.Response.StatusCode = (int)HttpStatusCode.BadRequest;
|
||||
return Task.CompletedTask;
|
||||
|
||||
if(!string.IsNullOrWhiteSpace(Message))
|
||||
await httpContext.Response.WriteAsync(Message, httpContext.RequestAborted);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -5,7 +5,6 @@ using Elsa.Expressions;
|
|||
using Elsa.Extensions;
|
||||
using Elsa.Models;
|
||||
using Elsa.Results;
|
||||
using Elsa.Scripting.JavaScript;
|
||||
using Elsa.Scripting.JavaScript.Services;
|
||||
using Elsa.Services;
|
||||
using Elsa.Services.Models;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,34 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Activities.Workflows.Activities;
|
||||
using Elsa.Models;
|
||||
using Elsa.Services;
|
||||
using Newtonsoft.Json.Linq;
|
||||
|
||||
namespace Elsa.Activities.Workflows.Extensions
|
||||
{
|
||||
public static class WorkflowInvokerExtensions
|
||||
{
|
||||
public static async Task TriggerSignalAsync(
|
||||
this IWorkflowInvoker workflowInvoker,
|
||||
string signalName,
|
||||
Variables input = default,
|
||||
Func<JObject, bool> activityStatePredicate = null,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var combinedInput = new Variables(
|
||||
new Dictionary<string, object>
|
||||
{
|
||||
["Signal"] = signalName
|
||||
});
|
||||
|
||||
if (input != null)
|
||||
combinedInput.AddVariables(input);
|
||||
|
||||
await workflowInvoker.TriggerAsync(nameof(Signaled), combinedInput, correlationId, activityStatePredicate, cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -3,7 +3,7 @@ using Microsoft.Extensions.DependencyInjection;
|
|||
|
||||
namespace Elsa.Activities.Workflows.Extensions
|
||||
{
|
||||
public static class WorkflowActivityServiceCollectionExtensions
|
||||
public static class WorkflowsServiceCollectionExtensions
|
||||
{
|
||||
public static IServiceCollection AddWorkflowActivities(this IServiceCollection services)
|
||||
{
|
||||
|
|
@ -1,16 +1,20 @@
|
|||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Expressions;
|
||||
using Elsa.Services;
|
||||
using Elsa.Services.Models;
|
||||
|
||||
namespace Elsa.Extensions
|
||||
{
|
||||
public static class WorkflowExpressionEvaluatorExtensions
|
||||
{
|
||||
public static async Task<T> EvaluateAsync<T>(this IWorkflowExpressionEvaluator evaluator, IWorkflowExpression<T> expression, WorkflowExecutionContext workflowExecutionContext, CancellationToken cancellationToken)
|
||||
{
|
||||
return (T)await evaluator.EvaluateAsync(expression, typeof(T), workflowExecutionContext, cancellationToken);
|
||||
}
|
||||
}
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Expressions;
|
||||
using Elsa.Services;
|
||||
using Elsa.Services.Models;
|
||||
|
||||
namespace Elsa.Extensions
|
||||
{
|
||||
public static class WorkflowExpressionEvaluatorExtensions
|
||||
{
|
||||
public static async Task<T> EvaluateAsync<T>(
|
||||
this IWorkflowExpressionEvaluator evaluator,
|
||||
IWorkflowExpression<T> expression,
|
||||
WorkflowExecutionContext workflowExecutionContext,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return (T)await evaluator.EvaluateAsync(expression, typeof(T), workflowExecutionContext, cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -10,7 +10,7 @@ namespace Elsa.Models
|
|||
{
|
||||
}
|
||||
|
||||
public Variables(Variables other) : this((IEnumerable<KeyValuePair<string, object>>) other)
|
||||
public Variables(Variables other) : this((IEnumerable<KeyValuePair<string, object>>)other)
|
||||
{
|
||||
}
|
||||
|
||||
|
|
@ -29,7 +29,21 @@ namespace Elsa.Models
|
|||
|
||||
public T GetVariable<T>(string name)
|
||||
{
|
||||
return ContainsKey(name) ? (T) this[name] : default(T);
|
||||
return ContainsKey(name) ? (T)this[name] : default(T);
|
||||
}
|
||||
|
||||
public void AddVariable(string name, object value)
|
||||
{
|
||||
this[name] = value;
|
||||
}
|
||||
|
||||
public void AddVariables(Variables variables) =>
|
||||
AddVariables((IEnumerable<KeyValuePair<string, object>>)variables);
|
||||
|
||||
public void AddVariables(IEnumerable<KeyValuePair<string, object>> variables)
|
||||
{
|
||||
foreach (var variable in variables)
|
||||
AddVariable(variable.Key, variable.Value);
|
||||
}
|
||||
|
||||
public bool HasVariable(string name, object value)
|
||||
|
|
|
|||
|
|
@ -1,7 +0,0 @@
|
|||
namespace Elsa.Services
|
||||
{
|
||||
internal interface IScopedWorkflowInvoker : IWorkflowInvoker
|
||||
{
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -1,7 +1,12 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Expressions;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Models;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using NodaTime;
|
||||
|
||||
namespace Elsa.Services.Models
|
||||
|
|
@ -20,6 +25,7 @@ namespace Elsa.Services.Models
|
|||
IsFirstPass = true;
|
||||
scheduledActivities = new Stack<IActivity>();
|
||||
scheduledHaltingActivities = new Stack<IActivity>();
|
||||
ExpressionEvaluator = serviceProvider.GetRequiredService<IWorkflowExpressionEvaluator>();
|
||||
}
|
||||
|
||||
public Workflow Workflow { get; }
|
||||
|
|
@ -53,6 +59,7 @@ namespace Elsa.Services.Models
|
|||
public IActivity PopScheduledActivity() => CurrentActivity = scheduledActivities.Pop();
|
||||
public void ScheduleHaltingActivity(IActivity activity) => scheduledHaltingActivities.Push(activity);
|
||||
public IActivity PopScheduledHaltingActivity() => scheduledHaltingActivities.Pop();
|
||||
public IWorkflowExpressionEvaluator ExpressionEvaluator { get; }
|
||||
|
||||
public void SetVariable(string name, object value)
|
||||
{
|
||||
|
|
@ -75,6 +82,9 @@ namespace Elsa.Services.Models
|
|||
return scope.GetVariable(name);
|
||||
}
|
||||
|
||||
public Task<T> EvaluateAsync<T>(IWorkflowExpression<T> expression, CancellationToken cancellationToken) =>
|
||||
ExpressionEvaluator.EvaluateAsync(expression, this, cancellationToken);
|
||||
|
||||
public void SetLastResult(object value) => CurrentScope.LastResult = value;
|
||||
|
||||
public void Start()
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using Elsa;
|
||||
using Elsa.Activities;
|
||||
using Elsa.AutoMapper.Extensions;
|
||||
|
|
@ -30,8 +29,8 @@ namespace Microsoft.Extensions.DependencyInjection
|
|||
Action<ElsaBuilder> configure = null)
|
||||
{
|
||||
var configuration = new ElsaBuilder(services);
|
||||
configuration.UseWorkflowsCore();
|
||||
configuration.UseMediatR();
|
||||
configuration.AddWorkflowsCore();
|
||||
configuration.AddMediatR();
|
||||
configure?.Invoke(configuration);
|
||||
EnsurePersistence(configuration);
|
||||
EnsureCaching(configuration);
|
||||
|
|
@ -53,51 +52,14 @@ namespace Microsoft.Extensions.DependencyInjection
|
|||
.AddTransient<IActivity>(sp => sp.GetRequiredService<T>());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Registers the specified service only if none already exists for the specified provider type.
|
||||
/// </summary>
|
||||
public static IServiceCollection TryAddProvider<TService, TProvider>(
|
||||
this IServiceCollection services,
|
||||
ServiceLifetime lifetime)
|
||||
private static IServiceCollection AddMediatR(this ElsaBuilder configuration)
|
||||
{
|
||||
return services.TryAddProvider(typeof(TService), typeof(TProvider), lifetime);
|
||||
return configuration.Services.AddMediatR(
|
||||
mediatr => mediatr.AsSingleton(),
|
||||
typeof(ElsaServiceCollectionExtensions));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Registers the specified service only if none already exists for the specified provider type.
|
||||
/// </summary>
|
||||
public static IServiceCollection TryAddProvider(
|
||||
this IServiceCollection services,
|
||||
Type serviceType,
|
||||
Type providerType,
|
||||
ServiceLifetime lifetime)
|
||||
{
|
||||
var descriptor = services.FirstOrDefault(
|
||||
x => x.ServiceType == serviceType && x.ImplementationType == providerType
|
||||
);
|
||||
|
||||
if (descriptor == null)
|
||||
{
|
||||
descriptor = new ServiceDescriptor(serviceType, providerType, lifetime);
|
||||
services.Add(descriptor);
|
||||
}
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
public static IServiceCollection Replace<TService, TImplementation>(
|
||||
this IServiceCollection services,
|
||||
ServiceLifetime lifetime)
|
||||
{
|
||||
return services.Replace(new ServiceDescriptor(typeof(TService), typeof(TImplementation), lifetime));
|
||||
}
|
||||
|
||||
private static IServiceCollection UseMediatR(this ElsaBuilder configuration)
|
||||
{
|
||||
return configuration.Services.AddMediatR(typeof(ElsaServiceCollectionExtensions));
|
||||
}
|
||||
|
||||
private static ElsaBuilder UseWorkflowsCore(this ElsaBuilder configuration)
|
||||
private static ElsaBuilder AddWorkflowsCore(this ElsaBuilder configuration)
|
||||
{
|
||||
var services = configuration.Services;
|
||||
services.TryAddSingleton<IClock>(SystemClock.Instance);
|
||||
|
|
@ -112,14 +74,13 @@ namespace Microsoft.Extensions.DependencyInjection
|
|||
.TryAddProvider<ITokenFormatter, YamlTokenFormatter>(ServiceLifetime.Singleton)
|
||||
.TryAddProvider<ITokenFormatter, XmlTokenFormatter>(ServiceLifetime.Singleton)
|
||||
.TryAddProvider<IExpressionEvaluator, LiteralEvaluator>(ServiceLifetime.Singleton)
|
||||
.AddScoped<IWorkflowFactory, WorkflowFactory>()
|
||||
.AddSingleton<IActivityInvoker, ActivityInvoker>()
|
||||
.AddTransient<IWorkflowFactory, WorkflowFactory>()
|
||||
.AddScoped<IActivityInvoker, ActivityInvoker>()
|
||||
.AddScoped<IWorkflowExpressionEvaluator, WorkflowExpressionEvaluator>()
|
||||
.AddSingleton<IWorkflowSerializerProvider, WorkflowSerializerProvider>()
|
||||
.AddSingleton<IWorkflowRegistry, WorkflowRegistry>()
|
||||
.AddTransient<IWorkflowRegistry, WorkflowRegistry>()
|
||||
.AddScoped<IWorkflowEventHandler, PersistenceWorkflowEventHandler>()
|
||||
.AddSingleton<IWorkflowInvoker, WorkflowInvoker>()
|
||||
.AddScoped<IScopedWorkflowInvoker, ScopedWorkflowInvoker>()
|
||||
.AddScoped<IWorkflowInvoker, WorkflowInvoker>()
|
||||
.AddScoped<IActivityResolver, ActivityResolver>()
|
||||
.AddScoped<IWorkflowEventHandler, ActivityLoggingWorkflowEventHandler>()
|
||||
.AddTransient<IWorkflowProvider, StoreWorkflowProvider>()
|
||||
|
|
@ -151,12 +112,6 @@ namespace Microsoft.Extensions.DependencyInjection
|
|||
configuration.Services.AddMemoryCache();
|
||||
}
|
||||
|
||||
private static bool HasService<T>(this IServiceCollection services) =>
|
||||
services.Any(x => x.ServiceType == typeof(T));
|
||||
|
||||
private static bool HasService<T>(this ElsaBuilder configuration) =>
|
||||
configuration.Services.HasService<T>();
|
||||
|
||||
private static IServiceCollection AddPrimitiveActivities(this IServiceCollection services)
|
||||
{
|
||||
return services
|
||||
|
|
|
|||
|
|
@ -0,0 +1,26 @@
|
|||
using System;
|
||||
using System.Linq;
|
||||
using System.Reflection;
|
||||
using MediatR;
|
||||
using MediatR.Registration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.Extensions
|
||||
{
|
||||
public static class MessageHandlerServiceCollectionExtensions
|
||||
{
|
||||
public static IServiceCollection AddNotificationHandler<T, THandler>(this IServiceCollection services)
|
||||
where T : INotification
|
||||
where THandler : INotificationHandler<T>
|
||||
{
|
||||
return services.AddTransient(typeof(INotificationHandler<T>), typeof(THandler));
|
||||
}
|
||||
|
||||
public static IServiceCollection AddNotificationHandlers(this IServiceCollection services, params Type[] markerTypes)
|
||||
{
|
||||
var assemblies = markerTypes.Select(x => x.GetTypeInfo().Assembly);
|
||||
ServiceRegistrar.AddMediatRClasses(services, assemblies);
|
||||
return services;
|
||||
}
|
||||
}
|
||||
}
|
||||
56
src/core/Elsa.Core/Extensions/ServiceCollectionExtensions.cs
Normal file
56
src/core/Elsa.Core/Extensions/ServiceCollectionExtensions.cs
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
using System;
|
||||
using System.Linq;
|
||||
using Elsa;
|
||||
using Microsoft.Extensions.DependencyInjection.Extensions;
|
||||
|
||||
// ReSharper disable once CheckNamespace
|
||||
namespace Microsoft.Extensions.DependencyInjection
|
||||
{
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Registers the specified service only if none already exists for the specified provider type.
|
||||
/// </summary>
|
||||
public static IServiceCollection TryAddProvider<TService, TProvider>(
|
||||
this IServiceCollection services,
|
||||
ServiceLifetime lifetime)
|
||||
{
|
||||
return services.TryAddProvider(typeof(TService), typeof(TProvider), lifetime);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Registers the specified service only if none already exists for the specified provider type.
|
||||
/// </summary>
|
||||
public static IServiceCollection TryAddProvider(
|
||||
this IServiceCollection services,
|
||||
Type serviceType,
|
||||
Type providerType,
|
||||
ServiceLifetime lifetime)
|
||||
{
|
||||
var descriptor = services.FirstOrDefault(
|
||||
x => x.ServiceType == serviceType && x.ImplementationType == providerType
|
||||
);
|
||||
|
||||
if (descriptor == null)
|
||||
{
|
||||
descriptor = new ServiceDescriptor(serviceType, providerType, lifetime);
|
||||
services.Add(descriptor);
|
||||
}
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
public static IServiceCollection Replace<TService, TImplementation>(
|
||||
this IServiceCollection services,
|
||||
ServiceLifetime lifetime)
|
||||
{
|
||||
return services.Replace(new ServiceDescriptor(typeof(TService), typeof(TImplementation), lifetime));
|
||||
}
|
||||
|
||||
public static bool HasService<T>(this IServiceCollection services) =>
|
||||
services.Any(x => x.ServiceType == typeof(T));
|
||||
|
||||
public static bool HasService<T>(this ElsaBuilder configuration) =>
|
||||
configuration.Services.HasService<T>();
|
||||
}
|
||||
}
|
||||
|
|
@ -14,11 +14,11 @@ namespace Elsa.Metadata
|
|||
|
||||
public string Type { get; set; }
|
||||
public string DisplayName { get; set; }
|
||||
public string? Description { get; set; }
|
||||
public string? RuntimeDescription { get; set; }
|
||||
public string Description { get; set; }
|
||||
public string RuntimeDescription { get; set; }
|
||||
public string Category { get; set; }
|
||||
public string? Icon { get; set; }
|
||||
public object? Outcomes { get; set; }
|
||||
public string Icon { get; set; }
|
||||
public object Outcomes { get; set; }
|
||||
public ActivityPropertyDescriptor[] Properties { get; set; }
|
||||
}
|
||||
}
|
||||
|
|
@ -2,7 +2,7 @@ namespace Elsa.Metadata
|
|||
{
|
||||
public class ActivityPropertyDescriptor
|
||||
{
|
||||
public ActivityPropertyDescriptor(string name, string type, string label, string? hint = null, object? options = null)
|
||||
public ActivityPropertyDescriptor(string name, string type, string label, string hint = null, object options = null)
|
||||
{
|
||||
Name = name;
|
||||
Type = type;
|
||||
|
|
@ -14,7 +14,7 @@ namespace Elsa.Metadata
|
|||
public string Name { get; }
|
||||
public string Type { get; }
|
||||
public string Label { get; }
|
||||
public string? Hint { get; }
|
||||
public object? Options { get; }
|
||||
public string Hint { get; }
|
||||
public object Options { get; }
|
||||
}
|
||||
}
|
||||
|
|
@ -1,457 +0,0 @@
|
|||
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.Results;
|
||||
using Elsa.Services.Extensions;
|
||||
using Elsa.Services.Models;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Newtonsoft.Json.Linq;
|
||||
using NodaTime;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
internal class ScopedWorkflowInvoker : IScopedWorkflowInvoker
|
||||
{
|
||||
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 ScopedWorkflowInvoker(
|
||||
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<ActivityExecutionResult> ExecuteActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
() => activityInvoker.ExecuteAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<ActivityExecutionResult> ResumeActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
() => activityInvoker.ResumeAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<ActivityExecutionResult> InvokeActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
Func<Task<ActivityExecutionResult>> 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<ActivityExecutionResult> ExecuteActivityHaltedAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
() => 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,21 +1,49 @@
|
|||
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.Results;
|
||||
using Elsa.Services.Extensions;
|
||||
using Elsa.Services.Models;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Newtonsoft.Json.Linq;
|
||||
using NodaTime;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
public class WorkflowInvoker : IWorkflowInvoker
|
||||
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(IServiceProvider serviceProvider)
|
||||
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(
|
||||
|
|
@ -23,16 +51,7 @@ namespace Elsa.Services
|
|||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return Invoke(x => x.StartAsync(workflow, startActivities, cancellationToken));
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> StartAsync<T>(
|
||||
Variables input = default,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default) where T : IWorkflow, new()
|
||||
{
|
||||
return Invoke(x => x.StartAsync<T>(input, startActivityIds, correlationId, cancellationToken));
|
||||
return ExecuteAsync(workflow, false, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> StartAsync(
|
||||
|
|
@ -42,9 +61,22 @@ namespace Elsa.Services
|
|||
string correlationId = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return Invoke(
|
||||
x => x.StartAsync(workflowDefinition, input, startActivityIds, correlationId, cancellationToken)
|
||||
);
|
||||
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(
|
||||
|
|
@ -52,45 +84,374 @@ namespace Elsa.Services
|
|||
IEnumerable<IActivity> startActivities = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return Invoke(x => x.ResumeAsync(workflow, startActivities, cancellationToken));
|
||||
return ExecuteAsync(workflow, true, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> ResumeAsync<T>(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = default,
|
||||
Variables input = null,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
where T : IWorkflow, new()
|
||||
CancellationToken cancellationToken = default) where T : IWorkflow, new()
|
||||
{
|
||||
return Invoke(x => x.ResumeAsync<T>(workflowInstance, input, startActivityIds, cancellationToken));
|
||||
var workflow = workflowFactory.CreateWorkflow<T>(input, workflowInstance);
|
||||
var startActivities = workflow.Activities.Find(startActivityIds);
|
||||
return ExecuteAsync(workflow, true, startActivities, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<WorkflowExecutionContext> ResumeAsync(
|
||||
public async Task<WorkflowExecutionContext> ResumeAsync(
|
||||
WorkflowInstance workflowInstance,
|
||||
Variables input = default,
|
||||
Variables input = null,
|
||||
IEnumerable<string> startActivityIds = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return Invoke(x => x.ResumeAsync(workflowInstance, input, startActivityIds, cancellationToken));
|
||||
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 Task<IEnumerable<WorkflowExecutionContext>> TriggerAsync(
|
||||
public async Task<IEnumerable<WorkflowExecutionContext>> TriggerAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return Invoke(x => x.TriggerAsync(activityType, input, correlationId, activityStatePredicate, cancellationToken));
|
||||
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<T> Invoke<T>(Func<IScopedWorkflowInvoker, Task<T>> action)
|
||||
private async Task<IEnumerable<WorkflowExecutionContext>> ResumeManyAsync(
|
||||
string activityType,
|
||||
Variables input = default,
|
||||
string correlationId = default,
|
||||
Func<JObject, bool> activityStatePredicate = default,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
using (var scope = serviceProvider.CreateScope())
|
||||
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 invoker = scope.ServiceProvider.GetRequiredService<IScopedWorkflowInvoker>();
|
||||
return await action(invoker);
|
||||
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<ActivityExecutionResult> ExecuteActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
async () => await activityInvoker.ExecuteAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<ActivityExecutionResult> ResumeActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await InvokeActivityAsync(
|
||||
workflowContext,
|
||||
activity,
|
||||
async () => await activityInvoker.ResumeAsync(workflowContext, activity, cancellationToken),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<ActivityExecutionResult> InvokeActivityAsync(
|
||||
WorkflowExecutionContext workflowContext,
|
||||
IActivity activity,
|
||||
Func<Task<ActivityExecutionResult>> 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<ActivityExecutionResult> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -19,13 +19,13 @@ namespace Elsa.Services
|
|||
private readonly ISignal signal;
|
||||
|
||||
public WorkflowRegistry(
|
||||
IServiceProvider serviceProvider,
|
||||
IMemoryCache cache,
|
||||
ISignal signal)
|
||||
ISignal signal,
|
||||
IServiceProvider serviceProvider)
|
||||
{
|
||||
this.serviceProvider = serviceProvider;
|
||||
this.cache = cache;
|
||||
this.signal = signal;
|
||||
this.serviceProvider = serviceProvider;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<(WorkflowDefinitionVersion, ActivityDefinition)>> ListByStartActivityAsync(
|
||||
|
|
@ -72,8 +72,7 @@ namespace Elsa.Services
|
|||
});
|
||||
}
|
||||
|
||||
private async Task<ICollection<WorkflowDefinitionVersion>> LoadWorkflowDefinitionsAsync(
|
||||
CancellationToken cancellationToken)
|
||||
private async Task<ICollection<WorkflowDefinitionVersion>> LoadWorkflowDefinitionsAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
using var scope = serviceProvider.CreateScope();
|
||||
var providers = scope.ServiceProvider.GetServices<IWorkflowProvider>();
|
||||
|
|
|
|||
|
|
@ -3,9 +3,7 @@ using Elsa;
|
|||
using Elsa.Activities.ControlFlow.Extensions;
|
||||
using Elsa.Activities.UserTask.Extensions;
|
||||
using Elsa.Activities.Workflows.Extensions;
|
||||
using Elsa.Scripting.JavaScript;
|
||||
using Elsa.Scripting.JavaScript.Extensions;
|
||||
using Elsa.Scripting.Liquid;
|
||||
using Elsa.Scripting.Liquid.Extensions;
|
||||
|
||||
// ReSharper disable once CheckNamespace
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ using System;
|
|||
using Elsa.Dashboard.Options;
|
||||
using Elsa.Metadata;
|
||||
using Elsa.Services.Models;
|
||||
using Elsa.WorkflowDesigner;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Scrutor;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
using System.Collections;
|
||||
using System.Collections.Generic;
|
||||
using Elsa.Metadata;
|
||||
using Elsa.WorkflowDesigner.Models;
|
||||
|
||||
namespace Elsa.Dashboard.Options
|
||||
{
|
||||
|
|
|
|||
|
|
@ -25,7 +25,6 @@ namespace Elsa.Persistence.MongoDb.Extensions
|
|||
RegisterEnumAsStringConvention();
|
||||
BsonSerializer.RegisterSerializer(new JObjectSerializer());
|
||||
BsonSerializer.RegisterSerializer(new WorkflowExecutionScopeSerializer());
|
||||
//BsonSerializer.RegisterSerializer(new WorkflowInstanceSerializer());
|
||||
|
||||
elsaBuilder.Services
|
||||
.AddSingleton(sp => CreateDbClient(configuration, connectionStringName))
|
||||
|
|
@ -51,7 +50,7 @@ namespace Elsa.Persistence.MongoDb.Extensions
|
|||
{
|
||||
configuration.Services
|
||||
.AddMongoDbCollection<WorkflowInstance>("WorkflowInstances")
|
||||
.Replace<IWorkflowInstanceStore, MongoWorkflowInstanceStore>(ServiceLifetime.Scoped);
|
||||
.AddScoped<IWorkflowInstanceStore, MongoWorkflowInstanceStore>();
|
||||
|
||||
return configuration;
|
||||
}
|
||||
|
|
@ -61,7 +60,7 @@ namespace Elsa.Persistence.MongoDb.Extensions
|
|||
{
|
||||
configuration.Services
|
||||
.AddMongoDbCollection<WorkflowDefinitionVersion>("WorkflowDefinitions")
|
||||
.Replace<IWorkflowDefinitionStore, MongoWorkflowDefinitionStore>(ServiceLifetime.Scoped);
|
||||
.AddScoped<IWorkflowDefinitionStore, MongoWorkflowDefinitionStore>();
|
||||
|
||||
return configuration;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
using Elsa.Extensions;
|
||||
using Elsa.Scripting.JavaScript.Services;
|
||||
using Elsa.Services;
|
||||
using MediatR;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.Scripting.JavaScript.Extensions
|
||||
|
|
@ -11,7 +11,7 @@ namespace Elsa.Scripting.JavaScript.Extensions
|
|||
{
|
||||
return services
|
||||
.TryAddProvider<IExpressionEvaluator, JavaScriptExpressionEvaluator>(ServiceLifetime.Scoped)
|
||||
.AddMediatR(typeof(JavaScriptServiceCollectionExtensions));
|
||||
.AddNotificationHandlers(typeof(JavaScriptServiceCollectionExtensions));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +1,4 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
using Elsa.Extensions;
|
||||
using Elsa.Scripting.Liquid.Filters;
|
||||
using Elsa.Scripting.Liquid.Options;
|
||||
using Elsa.Scripting.Liquid.Services;
|
||||
using Elsa.Services;
|
||||
using MediatR;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.Scripting.Liquid.Extensions
|
||||
|
|
@ -14,7 +14,7 @@ namespace Elsa.Scripting.Liquid.Extensions
|
|||
return services
|
||||
.TryAddProvider<IExpressionEvaluator, LiquidExpressionEvaluator>(ServiceLifetime.Scoped)
|
||||
.AddMemoryCache()
|
||||
.AddMediatR(typeof(LiquidServiceCollectionExtensions))
|
||||
.AddNotificationHandlers(typeof(LiquidServiceCollectionExtensions))
|
||||
.AddScoped<ILiquidTemplateManager, LiquidTemplateManager>()
|
||||
.AddLiquidFilter<JsonFilter>("json");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ using Elsa.Services.Models;
|
|||
using Fluid;
|
||||
using Fluid.Values;
|
||||
using MediatR;
|
||||
using Newtonsoft.Json.Linq;
|
||||
|
||||
namespace Elsa.Scripting.Liquid.Handlers
|
||||
{
|
||||
|
|
@ -17,11 +18,14 @@ namespace Elsa.Scripting.Liquid.Handlers
|
|||
static CommonLiquidContextHandler()
|
||||
{
|
||||
FluidValue.SetTypeMapping<ExpandoObject>(x => new ObjectValue(x));
|
||||
FluidValue.SetTypeMapping<JObject>(o => new ObjectValue(o));
|
||||
FluidValue.SetTypeMapping<JValue>(o => FluidValue.Create(o.Value));
|
||||
}
|
||||
|
||||
|
||||
public Task Handle(EvaluatingLiquidExpression notification, CancellationToken cancellationToken)
|
||||
{
|
||||
var context = notification.TemplateContext;
|
||||
|
||||
context.MemberAccessStrategy.Register<LiquidPropertyAccessor, FluidValue>((x, name) => x.GetValueAsync(name));
|
||||
context.MemberAccessStrategy.Register<WorkflowExecutionContext, LiquidPropertyAccessor>("Input", x => new LiquidPropertyAccessor(name => ToFluidValue(x.Workflow.Input, name)));
|
||||
context.MemberAccessStrategy.Register<WorkflowExecutionContext, LiquidPropertyAccessor>("Output", x => new LiquidPropertyAccessor(name => ToFluidValue(x.Workflow.Output, name)));
|
||||
|
|
@ -30,10 +34,11 @@ namespace Elsa.Scripting.Liquid.Handlers
|
|||
context.MemberAccessStrategy.Register<LiquidObjectAccessor<IActivity>, LiquidObjectAccessor<object>>((x, activityName) => new LiquidObjectAccessor<object>(outputKey => GetActivityOutput(x, activityName, outputKey)));
|
||||
context.MemberAccessStrategy.Register<LiquidObjectAccessor<object>, object>((x, name) => x.GetValueAsync(name));
|
||||
context.MemberAccessStrategy.Register<ExpandoObject, object>((x, name) => ((IDictionary<string, object>)x)[name]);
|
||||
|
||||
context.MemberAccessStrategy.Register<JObject, object>((source, name) => source[name]);
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
|
||||
private Task<FluidValue> ToFluidValue(IDictionary<string, object> dictionary, string key)
|
||||
{
|
||||
return Task.FromResult(!dictionary.ContainsKey(key) ? default : FluidValue.Create(dictionary[key]));
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ namespace Elsa.Scripting.Liquid.Services
|
|||
private async Task<TemplateContext> CreateTemplateContextAsync(WorkflowExecutionContext workflowContext)
|
||||
{
|
||||
var context = new TemplateContext();
|
||||
context.SetValue("WorkflowExecutionContext", workflowContext);
|
||||
await mediator.Publish(new EvaluatingLiquidExpression(context, workflowContext));
|
||||
context.Model = workflowContext;
|
||||
return context;
|
||||
|
|
|
|||
Loading…
Reference in a new issue