Add Workflow Interruptor API (#482)
This commit is contained in:
parent
0facd8f93f
commit
e32af24a3e
|
|
@ -162,6 +162,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.WhileLoopWorke
|
|||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.CustomAttributesChildWorker", "src\samples\worker\Elsa.Samples.CustomAttributesChildWorker\Elsa.Samples.CustomAttributesChildWorker.csproj", "{AEA00027-F280-43A4-AC60-E2AF3014CF34}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.CustomActivityTypeProviders", "src\samples\aspnet\Elsa.Samples.CustomActivityTypeProviders\Elsa.Samples.CustomActivityTypeProviders.csproj", "{E0090EF3-C4A7-4262-9300-1D395BE25DD0}"
|
||||
EndProject
|
||||
Global
|
||||
GlobalSection(SolutionConfigurationPlatforms) = preSolution
|
||||
Debug|Any CPU = Debug|Any CPU
|
||||
|
|
@ -393,6 +395,10 @@ Global
|
|||
{AEA00027-F280-43A4-AC60-E2AF3014CF34}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{AEA00027-F280-43A4-AC60-E2AF3014CF34}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{AEA00027-F280-43A4-AC60-E2AF3014CF34}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{E0090EF3-C4A7-4262-9300-1D395BE25DD0}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{E0090EF3-C4A7-4262-9300-1D395BE25DD0}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{E0090EF3-C4A7-4262-9300-1D395BE25DD0}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{E0090EF3-C4A7-4262-9300-1D395BE25DD0}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
EndGlobalSection
|
||||
GlobalSection(SolutionProperties) = preSolution
|
||||
HideSolutionNode = FALSE
|
||||
|
|
@ -470,6 +476,7 @@ Global
|
|||
{FD2D1BD0-1229-4DCF-BE70-6BFD396DD489} = {E42743A0-FBDD-4150-9D53-6000496D9B87}
|
||||
{EA3832C4-8079-4E84-AF8C-12744E4EFD44} = {E42743A0-FBDD-4150-9D53-6000496D9B87}
|
||||
{AEA00027-F280-43A4-AC60-E2AF3014CF34} = {E42743A0-FBDD-4150-9D53-6000496D9B87}
|
||||
{E0090EF3-C4A7-4262-9300-1D395BE25DD0} = {22E75696-6FE9-436A-9097-EE21C603F818}
|
||||
EndGlobalSection
|
||||
GlobalSection(ExtensibilityGlobals) = postSolution
|
||||
SolutionGuid = {8B0975FD-7050-48B0-88C5-48C33378E158}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,8 @@
|
|||
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpUseContinuousIndentInsideBracesMigration/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=Configurer/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=Interruptable/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=Interruptor/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=localizer/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=Materializer/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=Noda/@EntryIndexedValue">True</s:Boolean>
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.ActivityResults;
|
||||
using Elsa.Services.Models;
|
||||
|
|
@ -23,6 +24,11 @@ namespace Elsa.ActivityProviders
|
|||
/// </summary>
|
||||
public string? Description { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Anything you want to store with this activity type.
|
||||
/// </summary>
|
||||
public IDictionary<string, object> Annotations { get; set; } = new Dictionary<string, object>();
|
||||
|
||||
/// <summary>
|
||||
/// Returns a value of whether the specified activity can execute.
|
||||
/// </summary>
|
||||
|
|
|
|||
|
|
@ -0,0 +1,11 @@
|
|||
using System.Threading.Tasks;
|
||||
using Elsa.Services;
|
||||
using Elsa.Services.Models;
|
||||
|
||||
namespace Elsa
|
||||
{
|
||||
public static class ActivityActivatorExtensions
|
||||
{
|
||||
public static async Task<T> ActivateActivityAsync<T>(this IActivityActivator activator, ActivityExecutionContext context) where T : IActivity => (T) await activator.ActivateActivityAsync(context, typeof(T));
|
||||
}
|
||||
}
|
||||
|
|
@ -1,21 +1,11 @@
|
|||
// using System;
|
||||
// using System.Collections.Generic;
|
||||
// using System.Threading;
|
||||
// using System.Threading.Tasks;
|
||||
// using Elsa.Models;
|
||||
// using Elsa.Services.Models;
|
||||
// using Microsoft.Extensions.DependencyInjection;
|
||||
//
|
||||
// namespace Elsa.Services
|
||||
// {
|
||||
// public interface IActivityActivator
|
||||
// {
|
||||
// Task<IActivity> ActivateActivity(IServiceScope serviceScope, string activityTypeName, Action<IActivity>? setup = default, CancellationToken cancellationToken = default);
|
||||
// IActivity ActivateActivity(IServiceScope serviceScope, IActivityBlueprint activityBlueprint);
|
||||
// T ActivateActivity<T>(IServiceScope serviceScope, Action<T>? configure = default) where T : class, IActivity;
|
||||
// IActivity ActivateActivity(IServiceScope serviceScope, Type type);
|
||||
// IActivity ActivateActivity(IServiceScope serviceScope, ActivityDefinition activityDefinition);
|
||||
// IEnumerable<Type> GetActivityTypes();
|
||||
// Type? GetActivityType(string activityTypeName);
|
||||
// }
|
||||
// }
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Services.Models;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
public interface IActivityActivator
|
||||
{
|
||||
Task<IActivity> ActivateActivityAsync(ActivityExecutionContext context, Type type);
|
||||
}
|
||||
}
|
||||
18
src/core/Elsa.Abstractions/Services/IWorkflowInterruptor.cs
Normal file
18
src/core/Elsa.Abstractions/Services/IWorkflowInterruptor.cs
Normal file
|
|
@ -0,0 +1,18 @@
|
|||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Models;
|
||||
using Elsa.Services.Models;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
/// <summary>
|
||||
/// Causes a suspended workflow's trigger to be interrupted
|
||||
/// </summary>
|
||||
public interface IWorkflowInterruptor
|
||||
{
|
||||
Task<WorkflowInstance> InterruptActivityAsync(WorkflowInstance workflowInstance, string activityId, object? input = default, CancellationToken cancellationToken = default);
|
||||
Task<WorkflowInstance> InterruptActivityAsync(IWorkflowBlueprint workflowBlueprint, WorkflowInstance workflowInstance, string activityId, object? input = default, CancellationToken cancellationToken = default);
|
||||
Task<WorkflowInstance> InterruptActivityTypeAsync(IWorkflowBlueprint workflowBlueprint, WorkflowInstance workflowInstance, string activityType, object? input = default, CancellationToken cancellationToken = default);
|
||||
Task<WorkflowInstance> InterruptActivityTypeAsync(WorkflowInstance workflowInstance, string activityType, object? input = default, CancellationToken cancellationToken = default);
|
||||
}
|
||||
}
|
||||
|
|
@ -15,11 +15,13 @@ namespace Elsa.ActivityTypeProviders
|
|||
public class TypeBasedActivityProvider : IActivityTypeProvider
|
||||
{
|
||||
private readonly IActivityDescriber _activityDescriber;
|
||||
private readonly IActivityActivator _activityActivator;
|
||||
private readonly Lazy<IDictionary<string, Type>> _lazyActivityTypeLookup;
|
||||
|
||||
public TypeBasedActivityProvider(IServiceProvider serviceProvider, IActivityDescriber activityDescriber)
|
||||
public TypeBasedActivityProvider(IServiceProvider serviceProvider, IActivityDescriber activityDescriber, IActivityActivator activityActivator)
|
||||
{
|
||||
_activityDescriber = activityDescriber;
|
||||
_activityActivator = activityActivator;
|
||||
|
||||
_lazyActivityTypeLookup = new Lazy<IDictionary<string, Type>>(
|
||||
() =>
|
||||
|
|
@ -63,13 +65,6 @@ namespace Elsa.ActivityTypeProviders
|
|||
|
||||
private IEnumerable<Type> GetActivityTypes() => ActivityTypeLookup.Values.ToList();
|
||||
|
||||
private static async Task<IActivity> ActivateActivity(ActivityExecutionContext context, Type type)
|
||||
{
|
||||
var activity = (IActivity) ActivatorUtilities.GetServiceOrCreateInstance(context.ServiceScope.ServiceProvider, type);
|
||||
activity.Data = context.ActivityInstance.Data;
|
||||
activity.Id = context.ActivityInstance.Id;
|
||||
await context.WorkflowExecutionContext.WorkflowBlueprint.ActivityPropertyProviders.SetActivityPropertiesAsync(activity, context, context.CancellationToken);
|
||||
return activity;
|
||||
}
|
||||
private Task<IActivity> ActivateActivity(ActivityExecutionContext context, Type type) => _activityActivator.ActivateActivityAsync(context, type);
|
||||
}
|
||||
}
|
||||
|
|
@ -14,6 +14,7 @@ using Elsa.Converters;
|
|||
using Elsa.Data.Extensions;
|
||||
using Elsa.Expressions;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.HostedServices;
|
||||
using Elsa.Indexes;
|
||||
using Elsa.Mapping;
|
||||
using Elsa.Metadata;
|
||||
|
|
@ -25,6 +26,7 @@ using Elsa.Triggers;
|
|||
using Elsa.WorkflowProviders;
|
||||
using MediatR;
|
||||
using Microsoft.Extensions.DependencyInjection.Extensions;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using NodaTime;
|
||||
using Rebus.ServiceProvider;
|
||||
|
||||
|
|
@ -54,27 +56,27 @@ namespace Microsoft.Extensions.DependencyInjection
|
|||
return services;
|
||||
}
|
||||
|
||||
public static IServiceCollection AddActivity<T>(this IServiceCollection services)
|
||||
where T : class, IActivity
|
||||
{
|
||||
return services
|
||||
public static IServiceCollection AddActivity<T>(this IServiceCollection services) where T : class, IActivity =>
|
||||
services
|
||||
.AddTransient<T>()
|
||||
.AddTransient<IActivity>(sp => sp.GetRequiredService<T>());
|
||||
}
|
||||
|
||||
public static IServiceCollection AddWorkflow<T>(this IServiceCollection services) where T : class, IWorkflow
|
||||
{
|
||||
return services
|
||||
.AddSingleton<T>()
|
||||
.AddSingleton<IWorkflow>(sp => sp.GetRequiredService<T>());
|
||||
services.TryAddSingleton<T>();
|
||||
services.TryAddSingleton<IWorkflow>(sp => sp.GetRequiredService<T>());
|
||||
return services;
|
||||
}
|
||||
|
||||
public static IServiceCollection AddWorkflow(this IServiceCollection services, IWorkflow workflow)
|
||||
{
|
||||
return services
|
||||
public static IServiceCollection AddWorkflow(this IServiceCollection services, IWorkflow workflow) =>
|
||||
services
|
||||
.AddSingleton(workflow.GetType(), workflow)
|
||||
.AddTransient(sp => workflow);
|
||||
}
|
||||
|
||||
public static IServiceCollection StartWorkflow<T>(this IServiceCollection services) where T : class, IWorkflow =>
|
||||
services
|
||||
.AddWorkflow<T>()
|
||||
.AddHostedService<StartWorkflow<T>>();
|
||||
|
||||
private static IServiceCollection AddMediatR(this ElsaOptions options) => options.Services.AddMediatR(mediatr => mediatr.AsScoped(), typeof(IActivity));
|
||||
|
||||
|
|
@ -96,7 +98,9 @@ namespace Microsoft.Extensions.DependencyInjection
|
|||
.AddScoped<IExpressionEvaluator, ExpressionEvaluator>()
|
||||
.AddScoped<IWorkflowRegistry, WorkflowRegistry>()
|
||||
.AddSingleton<IWorkflowSchedulerQueue, WorkflowSchedulerQueue>()
|
||||
.AddSingleton<IActivityActivator, ActivityActivator>()
|
||||
.AddScoped<IWorkflowRunner, WorkflowRunner>()
|
||||
.AddScoped<IWorkflowInterruptor, WorkflowInterruptor>()
|
||||
.AddSingleton<IActivityDescriber, ActivityDescriber>()
|
||||
.AddSingleton<IWorkflowFactory, WorkflowFactory>()
|
||||
.AddSingleton<IActivityFactory, ActivityFactory>()
|
||||
|
|
@ -110,7 +114,6 @@ namespace Microsoft.Extensions.DependencyInjection
|
|||
.AddIndexProvider<WorkflowDefinitionIndexProvider>()
|
||||
.AddIndexProvider<WorkflowInstanceIndexProvider>()
|
||||
.AddStartupRunner()
|
||||
//.AddSingleton<IActivityActivator, ActivityActivator>()
|
||||
.AddSingleton<IActivityTypeService, ActivityTypeService>()
|
||||
.AddSingleton<IActivityTypeProvider, TypeBasedActivityProvider>()
|
||||
.AddWorkflowProvider<ProgrammaticWorkflowProvider>()
|
||||
|
|
|
|||
|
|
@ -36,5 +36,13 @@ namespace Elsa.Extensions
|
|||
manager
|
||||
.Query<WorkflowInstanceIndex>(x => x.WorkflowStatus == workflowStatus)
|
||||
.ListAsync();
|
||||
|
||||
public static Task<IEnumerable<WorkflowInstance>> ListByBlockingActivityTypeAsync(
|
||||
this IWorkflowInstanceManager manager,
|
||||
string activityType,
|
||||
CancellationToken cancellationToken = default) =>
|
||||
manager
|
||||
.Query<WorkflowInstanceBlockingActivitiesIndex>(x => x.ActivityType == activityType)
|
||||
.ListAsync();
|
||||
}
|
||||
}
|
||||
28
src/core/Elsa.Core/HostedServices/StartWorkflow.cs
Normal file
28
src/core/Elsa.Core/HostedServices/StartWorkflow.cs
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Builders;
|
||||
using Elsa.Services;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
|
||||
namespace Elsa.HostedServices
|
||||
{
|
||||
/// <summary>
|
||||
/// A hosted service that starts a workflow of the specified type.
|
||||
/// </summary>
|
||||
public class StartWorkflow<T> : IHostedService where T : IWorkflow
|
||||
{
|
||||
private readonly IServiceProvider _serviceProvider;
|
||||
public StartWorkflow(IServiceProvider serviceProvider) => _serviceProvider = serviceProvider;
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
using var scope = _serviceProvider.CreateScope();
|
||||
var workflowRunner = scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
|
||||
await workflowRunner.RunWorkflowAsync<T>(cancellationToken: cancellationToken);
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
|
@ -46,17 +46,23 @@ namespace Elsa.Indexes
|
|||
|
||||
context.For<WorkflowInstanceBlockingActivitiesIndex>()
|
||||
.Map(
|
||||
workflowInstance => workflowInstance.BlockingActivities
|
||||
.Select(
|
||||
activity => new WorkflowInstanceBlockingActivitiesIndex
|
||||
{
|
||||
ActivityId = activity.ActivityId,
|
||||
ActivityType = activity.ActivityType,
|
||||
CorrelationId = workflowInstance.CorrelationId,
|
||||
TenantId = workflowInstance.TenantId,
|
||||
WorkflowStatus = workflowInstance.Status,
|
||||
CreatedAt = workflowInstance.CreatedAt.ToDateTimeOffset()
|
||||
}));
|
||||
workflowInstance =>
|
||||
{
|
||||
if (workflowInstance.Status != WorkflowStatus.Suspended || !workflowInstance.BlockingActivities.Any())
|
||||
return default;
|
||||
|
||||
return workflowInstance.BlockingActivities
|
||||
.Select(
|
||||
activity => new WorkflowInstanceBlockingActivitiesIndex
|
||||
{
|
||||
ActivityId = activity.ActivityId,
|
||||
ActivityType = activity.ActivityType,
|
||||
CorrelationId = workflowInstance.CorrelationId,
|
||||
TenantId = workflowInstance.TenantId,
|
||||
WorkflowStatus = workflowInstance.Status,
|
||||
CreatedAt = workflowInstance.CreatedAt.ToDateTimeOffset()
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,71 +1,19 @@
|
|||
// using System;
|
||||
// using System.Collections.Generic;
|
||||
// using System.Linq;
|
||||
// using System.Threading;
|
||||
// using System.Threading.Tasks;
|
||||
// using Elsa.ActivityProviders;
|
||||
// using Elsa.Models;
|
||||
// using Elsa.Services.Models;
|
||||
// using Microsoft.Extensions.DependencyInjection;
|
||||
//
|
||||
// namespace Elsa.Services
|
||||
// {
|
||||
// public class ActivityActivator : IActivityActivator
|
||||
// {
|
||||
// private readonly IActivityTypeService _activityTypeService;
|
||||
//
|
||||
// public ActivityActivator(IActivityTypeService activityTypeService)
|
||||
// {
|
||||
// _activityTypeService = activityTypeService;
|
||||
// }
|
||||
//
|
||||
// public async Task<IActivity> ActivateActivity(IServiceScope serviceScope, string activityTypeName, Action<IActivity>? setup, CancellationToken cancellationToken)
|
||||
// {
|
||||
// var activityType = await _activityTypeService.GetActivityTypeAsync(activityTypeName, cancellationToken);
|
||||
// var activity = activityType.
|
||||
//
|
||||
// setup?.Invoke(activity);
|
||||
// return activity;
|
||||
// }
|
||||
//
|
||||
// public T ActivateActivity<T>(IServiceScope serviceScope, Action<T>? setup = null) where T : class, IActivity
|
||||
// {
|
||||
// var activity = ActivatorUtilities.GetServiceOrCreateInstance<T>(serviceScope.ServiceProvider);
|
||||
// setup?.Invoke(activity);
|
||||
// return activity;
|
||||
// }
|
||||
//
|
||||
// public IActivity ActivateActivity(IServiceScope serviceScope, Type type) => (IActivity)ActivatorUtilities.GetServiceOrCreateInstance(serviceScope.ServiceProvider, type);
|
||||
//
|
||||
// public IActivity ActivateActivity(IServiceScope serviceScope, IActivityBlueprint activityBlueprint)
|
||||
// {
|
||||
// return ActivateActivity(
|
||||
// serviceScope,
|
||||
// activityBlueprint.Type,
|
||||
// activity =>
|
||||
// {
|
||||
// activity.Id = activityBlueprint.Id;
|
||||
// activity.Name = activityBlueprint.Name;
|
||||
// activity.PersistWorkflow = activityBlueprint.PersistWorkflow;
|
||||
// activity.SaveWorkflowContext = activityBlueprint.SaveWorkflowContext;
|
||||
// activity.LoadWorkflowContext = activityBlueprint.LoadWorkflowContext;
|
||||
// });
|
||||
// }
|
||||
//
|
||||
// public IActivity ActivateActivity(IServiceScope serviceScope, ActivityDefinition activityDefinition)
|
||||
// {
|
||||
// var activity = ActivateActivity(serviceScope, activityDefinition.Type);
|
||||
// activity.Description = activityDefinition.Description;
|
||||
// activity.Id = activityDefinition.ActivityId;
|
||||
// activity.Name = activityDefinition.Name;
|
||||
// activity.DisplayName = activityDefinition.DisplayName;
|
||||
// activity.PersistWorkflow = activityDefinition.PersistWorkflow;
|
||||
// activity.LoadWorkflowContext = activityDefinition.LoadWorkflowContext;
|
||||
// activity.SaveWorkflowContext = activityDefinition.SaveWorkflowContext;
|
||||
// return activity;
|
||||
// }
|
||||
//
|
||||
// public IEnumerable<Type> GetActivityTypes() => ActivityTypeLookup.Values.ToList();
|
||||
// public Type? GetActivityType(string activityTypeName) => ActivityTypeLookup.ContainsKey(activityTypeName) ? ActivityTypeLookup[activityTypeName] : default;
|
||||
// }
|
||||
// }
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Services.Models;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
public class ActivityActivator : IActivityActivator
|
||||
{
|
||||
public async Task<IActivity> ActivateActivityAsync(ActivityExecutionContext context, Type type)
|
||||
{
|
||||
var activity = (IActivity) ActivatorUtilities.GetServiceOrCreateInstance(context.ServiceScope.ServiceProvider, type);
|
||||
activity.Data = context.ActivityInstance.Data;
|
||||
activity.Id = context.ActivityInstance.Id;
|
||||
await context.WorkflowExecutionContext.WorkflowBlueprint.ActivityPropertyProviders.SetActivityPropertiesAsync(activity, context, context.CancellationToken);
|
||||
return activity;
|
||||
}
|
||||
}
|
||||
}
|
||||
59
src/core/Elsa.Core/Services/WorkflowInterruptor.cs
Normal file
59
src/core/Elsa.Core/Services/WorkflowInterruptor.cs
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Exceptions;
|
||||
using Elsa.Models;
|
||||
using Elsa.Services.Models;
|
||||
|
||||
namespace Elsa.Services
|
||||
{
|
||||
public class WorkflowInterruptor : IWorkflowInterruptor
|
||||
{
|
||||
private readonly IWorkflowRunner _workflowRunner;
|
||||
private readonly IWorkflowRegistry _workflowRegistry;
|
||||
|
||||
public WorkflowInterruptor(IWorkflowRunner workflowRunner, IWorkflowRegistry workflowRegistry)
|
||||
{
|
||||
_workflowRunner = workflowRunner;
|
||||
_workflowRegistry = workflowRegistry;
|
||||
}
|
||||
|
||||
public async Task<WorkflowInstance> InterruptActivityAsync(WorkflowInstance workflowInstance, string activityId, object? input, CancellationToken cancellationToken)
|
||||
{
|
||||
var workflowBlueprint = await GetWorkflowBlueprintAsync(workflowInstance, cancellationToken);
|
||||
return await InterruptActivityAsync(workflowBlueprint!, workflowInstance, activityId, input, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<WorkflowInstance> InterruptActivityAsync(IWorkflowBlueprint workflowBlueprint, WorkflowInstance workflowInstance, string activityId, object? input, CancellationToken cancellationToken)
|
||||
{
|
||||
if (workflowInstance.Status != WorkflowStatus.Suspended)
|
||||
throw new WorkflowException("Cannot interrupt workflows that are not in the Suspended state.");
|
||||
|
||||
var blockingActivity = workflowInstance.BlockingActivities.SingleOrDefault(x => x.ActivityId == activityId);
|
||||
|
||||
if (blockingActivity == null)
|
||||
throw new WorkflowException($"No blocking activity with ID {activityId} found.");
|
||||
|
||||
return await _workflowRunner.RunWorkflowAsync(workflowBlueprint, workflowInstance, activityId, input, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<WorkflowInstance> InterruptActivityTypeAsync(IWorkflowBlueprint workflowBlueprint, WorkflowInstance workflowInstance, string activityType, object? input, CancellationToken cancellationToken)
|
||||
{
|
||||
var blockingActivities = workflowInstance.BlockingActivities.Where(x => x.ActivityType == activityType).ToList();
|
||||
|
||||
foreach (var blockingActivity in blockingActivities)
|
||||
workflowInstance = await InterruptActivityAsync(workflowBlueprint, workflowInstance, blockingActivity.ActivityId, input, cancellationToken);
|
||||
|
||||
return workflowInstance;
|
||||
}
|
||||
|
||||
public async Task<WorkflowInstance> InterruptActivityTypeAsync(WorkflowInstance workflowInstance, string activityType, object? input, CancellationToken cancellationToken)
|
||||
{
|
||||
var workflowBlueprint = await GetWorkflowBlueprintAsync(workflowInstance, cancellationToken);
|
||||
return await InterruptActivityTypeAsync(workflowBlueprint!, workflowInstance, activityType, input, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<IWorkflowBlueprint?> GetWorkflowBlueprintAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken) =>
|
||||
await _workflowRegistry.GetWorkflowAsync(workflowInstance.WorkflowDefinitionId, workflowInstance.TenantId, VersionOptions.SpecificVersion(workflowInstance.Version), cancellationToken);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,14 @@
|
|||
using Elsa.Activities.Timers;
|
||||
using Elsa.Activities.Timers.Services;
|
||||
using Elsa.Services;
|
||||
using NodaTime;
|
||||
|
||||
namespace Elsa.Samples.CustomActivityTypeProviders.Activities
|
||||
{
|
||||
public class Sleep : TimerEvent
|
||||
{
|
||||
public Sleep(IWorkflowInstanceManager workflowInstanceManager, IWorkflowScheduler workflowScheduler, IClock clock) : base(workflowInstanceManager, workflowScheduler, clock)
|
||||
{
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,12 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk.Web">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net5.0</TargetFramework>
|
||||
<LangVersion>latest</LangVersion>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\core\Elsa\Elsa.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Samples.CustomActivityTypeProviders.Activities;
|
||||
using Elsa.Services;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
|
||||
namespace Elsa.Samples.CustomActivityTypeProviders.Endpoints
|
||||
{
|
||||
[ApiController]
|
||||
[Route("wakeup")]
|
||||
public class WakeUp : Controller
|
||||
{
|
||||
private readonly IWorkflowInterruptor _workflowInterruptor;
|
||||
private readonly IWorkflowInstanceManager _workflowInstanceManager;
|
||||
|
||||
public WakeUp(IWorkflowInterruptor workflowInterruptor, IWorkflowInstanceManager workflowInstanceManager)
|
||||
{
|
||||
_workflowInterruptor = workflowInterruptor;
|
||||
_workflowInstanceManager = workflowInstanceManager;
|
||||
}
|
||||
|
||||
[HttpGet]
|
||||
public async Task<IActionResult> Handle(CancellationToken cancellationToken)
|
||||
{
|
||||
// Get all workflows blocked on the "Sleep" activity.
|
||||
var suspendedWorkflows = (await _workflowInstanceManager.ListByBlockingActivityTypeAsync(nameof(Sleep), cancellationToken)).ToList();
|
||||
|
||||
// Interrupt each workflow by triggering the "Sleep" activity.
|
||||
foreach (var workflowInstance in suspendedWorkflows)
|
||||
await _workflowInterruptor.InterruptActivityTypeAsync(workflowInstance, nameof(Sleep), cancellationToken: cancellationToken);
|
||||
|
||||
return Ok($"Interrupted {suspendedWorkflows.Count} workflows.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
using Microsoft.AspNetCore.Hosting;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
|
||||
namespace Elsa.Samples.CustomActivityTypeProviders
|
||||
{
|
||||
public class Program
|
||||
{
|
||||
public static void Main(string[] args)
|
||||
{
|
||||
CreateHostBuilder(args).Build().Run();
|
||||
}
|
||||
|
||||
public static IHostBuilder CreateHostBuilder(string[] args) =>
|
||||
Host.CreateDefaultBuilder(args)
|
||||
.ConfigureWebHostDefaults(webBuilder => webBuilder.UseStartup<Startup>());
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
{
|
||||
"profiles": {
|
||||
"Elsa.Samples.CustomActivityTypeProviders": {
|
||||
"commandName": "Project",
|
||||
"dotnetRunMessages": "true",
|
||||
"launchBrowser": false,
|
||||
"applicationUrl": "https://localhost:6171",
|
||||
"environmentVariables": {
|
||||
"ASPNETCORE_ENVIRONMENT": "Development"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
using Elsa.Samples.CustomActivityTypeProviders.Activities;
|
||||
using Elsa.Samples.CustomActivityTypeProviders.Workflows;
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Hosting;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.Samples.CustomActivityTypeProviders
|
||||
{
|
||||
public class Startup
|
||||
{
|
||||
public void ConfigureServices(IServiceCollection services)
|
||||
{
|
||||
services.AddControllers();
|
||||
|
||||
services
|
||||
.AddElsa()
|
||||
.AddConsoleActivities()
|
||||
.AddHttpActivities()
|
||||
.AddTimerActivities()
|
||||
.AddActivity<Sleep>()
|
||||
.StartWorkflow<InterruptableWorkflow>();
|
||||
}
|
||||
|
||||
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
|
||||
{
|
||||
app.UseDeveloperExceptionPage();
|
||||
app.UseHttpActivities();
|
||||
app.UseRouting();
|
||||
app.UseEndpoints(endpoints => endpoints.MapControllers());
|
||||
app.UseWelcomePage();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
using Elsa.Activities.Console;
|
||||
using Elsa.Builders;
|
||||
using Elsa.Samples.CustomActivityTypeProviders.Activities;
|
||||
using NodaTime;
|
||||
|
||||
namespace Elsa.Samples.CustomActivityTypeProviders.Workflows
|
||||
{
|
||||
public class InterruptableWorkflow : IWorkflow
|
||||
{
|
||||
public void Build(IWorkflowBuilder workflow)
|
||||
{
|
||||
workflow
|
||||
.WriteLine("This workflow will sleep for 5 minutes before it continues.")
|
||||
.WriteLine("Can't wait that long? Send me a message at https://localhost:6171/wakeup.")
|
||||
.Then<Sleep>(sleep => sleep.Set(x => x.Timeout, Duration.FromMinutes(5)))
|
||||
.WriteLine("Done.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
{
|
||||
"Logging": {
|
||||
"LogLevel": {
|
||||
"Default": "Information",
|
||||
"Microsoft": "Warning",
|
||||
"Microsoft.Hosting.Lifetime": "Information"
|
||||
}
|
||||
},
|
||||
"AllowedHosts": "*"
|
||||
}
|
||||
|
|
@ -0,0 +1 @@
|
|||
GET https://localhost:6171/wakeup
|
||||
|
|
@ -9,7 +9,7 @@ namespace Elsa.Samples.HelloWorldHttp
|
|||
public void ConfigureServices(IServiceCollection services)
|
||||
{
|
||||
services
|
||||
.AddElsa(option => option.UsePersistence(db => db.UseSqLite("Data Source=elsa.db;Cache=Shared")))
|
||||
.AddElsa()
|
||||
.AddHttpActivities()
|
||||
.AddWorkflow<HelloHttpWorkflow>();
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue