Implement workflow instance deleter with support for locking

This commit is contained in:
Sipke Schoorstra 2021-09-09 21:06:11 +02:00
parent 0184003b85
commit 86efb968c4
5 changed files with 101 additions and 18 deletions

View file

@ -0,0 +1,19 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Models;
namespace Elsa.Services
{
public interface IWorkflowInstanceDeleter
{
Task<DeleteWorkflowInstanceResult> DeleteAsync(string workflowInstanceId, CancellationToken cancellationToken = default);
}
public record DeleteWorkflowInstanceResult(DeleteWorkflowInstanceResultStatus Status, WorkflowInstance? WorkflowInstance);
public enum DeleteWorkflowInstanceResultStatus
{
Ok,
NotFound
}
}

View file

@ -0,0 +1,33 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Exceptions;
using Elsa.Options;
using Elsa.Services;
namespace Elsa.Decorators
{
public class LockingWorkflowInstanceDeleter : IWorkflowInstanceDeleter
{
private readonly IWorkflowInstanceDeleter _workflowInstanceDeleter;
private readonly IDistributedLockProvider _distributedLockProvider;
private readonly ElsaOptions _elsaOptions;
public LockingWorkflowInstanceDeleter(IWorkflowInstanceDeleter workflowInstanceDeleter, IDistributedLockProvider distributedLockProvider, ElsaOptions elsaOptions)
{
_workflowInstanceDeleter = workflowInstanceDeleter;
_distributedLockProvider = distributedLockProvider;
_elsaOptions = elsaOptions;
}
public async Task<DeleteWorkflowInstanceResult> DeleteAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
{
var workflowInstanceLockKey = $"workflow-instance:{workflowInstanceId}";
await using var workflowInstanceLockHandle = await _distributedLockProvider.AcquireLockAsync(workflowInstanceLockKey, _elsaOptions.DistributedLockTimeout, cancellationToken);
if (workflowInstanceLockHandle == null)
throw new LockAcquisitionException("Could not acquire a lock within the configured amount of time");
return await _workflowInstanceDeleter.DeleteAsync(workflowInstanceId, cancellationToken);
}
}
}

View file

@ -56,7 +56,7 @@ namespace Microsoft.Extensions.DependencyInjection
{
TypeDescriptor.AddAttributes(typeof(Type), new TypeConverterAttribute(typeof(TypeTypeConverter)));
}
public static IServiceCollection AddElsaCore(
this IServiceCollection services,
Action<ElsaOptionsBuilder>? configure = default)
@ -91,11 +91,13 @@ namespace Microsoft.Extensions.DependencyInjection
.AddConfiguration()
.AddCoreActivities();
services.Decorate<IWorkflowDefinitionStore, InitializingWorkflowDefinitionStore>();
services.Decorate<IWorkflowDefinitionStore, EventPublishingWorkflowDefinitionStore>();
services.Decorate<IWorkflowInstanceStore, EventPublishingWorkflowInstanceStore>();
services.Decorate<IWorkflowInstanceExecutor, LockingWorkflowInstanceExecutor>();
services.Decorate<IWorkflowInstanceCanceller, LockingWorkflowInstanceCanceller>();
services
.Decorate<IWorkflowDefinitionStore, InitializingWorkflowDefinitionStore>()
.Decorate<IWorkflowDefinitionStore, EventPublishingWorkflowDefinitionStore>()
.Decorate<IWorkflowInstanceStore, EventPublishingWorkflowInstanceStore>()
.Decorate<IWorkflowInstanceExecutor, LockingWorkflowInstanceExecutor>()
.Decorate<IWorkflowInstanceCanceller, LockingWorkflowInstanceCanceller>()
.Decorate<IWorkflowInstanceDeleter, LockingWorkflowInstanceDeleter>();
//TenantId default source
services.TryAddScoped<ITenantAccessor, DefaultTenantAccessor>();
@ -128,7 +130,7 @@ namespace Microsoft.Extensions.DependencyInjection
elsaOptions.Services.AddTransient<IHandleMessages<TMessage>, TConsumer>();
return elsaOptions;
}
public static ElsaOptionsBuilder AddPubSubConsumer<TConsumer, TMessage>(this ElsaOptionsBuilder elsaOptions, string? queueName = default) where TConsumer : class, IHandleMessages<TMessage>
{
elsaOptions.Services.AddTransient<IHandleMessages<TMessage>, TConsumer>();
@ -139,7 +141,7 @@ namespace Microsoft.Extensions.DependencyInjection
public static IServiceCollection AddActivityPropertyOptionsProvider<T>(this IServiceCollection services) where T : class, IActivityPropertyOptionsProvider => services.AddSingleton<IActivityPropertyOptionsProvider, T>();
public static IServiceCollection AddRuntimeSelectItemsProvider<T>(this IServiceCollection services) where T : class, IRuntimeSelectListItemsProvider => services.AddScoped<IRuntimeSelectListItemsProvider, T>();
public static IServiceCollection AddActivityTypeProvider<T>(this IServiceCollection services) where T : class, IActivityTypeProvider => services.AddSingleton<IActivityTypeProvider, T>();
public static IServiceCollection AddWorkflowStorageProvider<T>(this IServiceCollection services) where T : class, IWorkflowStorageProvider =>
services
.AddSingleton<T>()
@ -176,6 +178,7 @@ namespace Microsoft.Extensions.DependencyInjection
.AddScoped<IWorkflowTriggerInterruptor, WorkflowTriggerInterruptor>()
.AddScoped<IWorkflowReviver, WorkflowReviver>()
.AddScoped<IWorkflowInstanceCanceller, WorkflowInstanceCanceller>()
.AddScoped<IWorkflowInstanceDeleter, WorkflowInstanceDeleter>()
.AddSingleton<IWorkflowFactory, WorkflowFactory>()
.AddTransient<IWorkflowBlueprintMaterializer, WorkflowBlueprintMaterializer>()
.AddSingleton<IWorkflowBlueprintReflector, WorkflowBlueprintReflector>()
@ -189,7 +192,7 @@ namespace Microsoft.Extensions.DependencyInjection
.AddTransient<ICreatesActivityExecutionContextForActivityBlueprint, ActivityExecutionContextForActivityBlueprintFactory>()
.AddTransient<IGetsStartActivities, GetsStartActivitiesProvider>()
;
// Data Protection.
services.AddDataProtection();
@ -215,14 +218,14 @@ namespace Microsoft.Extensions.DependencyInjection
.AddWorkflowProvider<DatabaseWorkflowProvider>();
services.Configure<BlobStorageWorkflowProviderOptions>(o => o.BlobStorageFactory = StorageFactory.Blobs.InMemory);
// Workflow Storage Providers.
services
.AddSingleton<IWorkflowStorageService, WorkflowStorageService>()
.AddWorkflowStorageProvider<TransientWorkflowStorageProvider>()
.AddWorkflowStorageProvider<WorkflowInstanceWorkflowStorageProvider>()
.AddWorkflowStorageProvider<BlobStorageWorkflowStorageProvider>();
services.Configure<BlobStorageWorkflowStorageProviderOptions>(o => o.BlobStorageFactory = StorageFactory.Blobs.InMemory);
// Metadata.

View file

@ -0,0 +1,28 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Persistence;
namespace Elsa.Services.Workflows
{
public class WorkflowInstanceDeleter : IWorkflowInstanceDeleter
{
private readonly IWorkflowInstanceStore _workflowInstanceStore;
public WorkflowInstanceDeleter(IWorkflowInstanceStore workflowInstanceStore)
{
_workflowInstanceStore = workflowInstanceStore;
}
public async Task<DeleteWorkflowInstanceResult> DeleteAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
{
var workflowInstance = await _workflowInstanceStore.FindByIdAsync(workflowInstanceId, cancellationToken);
if (workflowInstance == null)
return new DeleteWorkflowInstanceResult(DeleteWorkflowInstanceResultStatus.NotFound, null);
await _workflowInstanceStore.DeleteAsync(workflowInstance, cancellationToken);
return new DeleteWorkflowInstanceResult(DeleteWorkflowInstanceResultStatus.Ok, workflowInstance);
}
}
}

View file

@ -1,6 +1,7 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Persistence;
using Elsa.Services;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using Swashbuckle.AspNetCore.Annotations;
@ -13,11 +14,11 @@ namespace Elsa.Server.Api.Endpoints.WorkflowInstances
[Produces("application/json")]
public class Delete : Controller
{
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IWorkflowInstanceDeleter _workflowInstanceDeleter;
public Delete(IWorkflowInstanceStore workflowInstanceStore)
public Delete(IWorkflowInstanceDeleter workflowInstanceDeleter)
{
_workflowInstanceStore = workflowInstanceStore;
_workflowInstanceDeleter = workflowInstanceDeleter;
}
[HttpDelete]
@ -31,12 +32,11 @@ namespace Elsa.Server.Api.Endpoints.WorkflowInstances
]
public async Task<IActionResult> Handle(string id, CancellationToken cancellationToken = default)
{
var workflowInstance = await _workflowInstanceStore.FindByIdAsync(id, cancellationToken);
var result = await _workflowInstanceDeleter.DeleteAsync(id, cancellationToken);
if (workflowInstance == null)
if (result.Status == DeleteWorkflowInstanceResultStatus.NotFound)
return NotFound();
await _workflowInstanceStore.DeleteAsync(workflowInstance, cancellationToken);
return NoContent();
}
}