diff --git a/src/core/Elsa.Abstractions/Services/IWorkflowInterruptor.cs b/src/core/Elsa.Abstractions/Services/IWorkflowInterruptor.cs index a5db5aa1d..2b8551c0f 100644 --- a/src/core/Elsa.Abstractions/Services/IWorkflowInterruptor.cs +++ b/src/core/Elsa.Abstractions/Services/IWorkflowInterruptor.cs @@ -1,4 +1,5 @@ -using System.Threading; +using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; using Elsa.Models; using Elsa.Services.Models; @@ -14,5 +15,6 @@ namespace Elsa.Services Task InterruptActivityAsync(IWorkflowBlueprint workflowBlueprint, WorkflowInstance workflowInstance, string activityId, object? input = default, CancellationToken cancellationToken = default); Task InterruptActivityTypeAsync(IWorkflowBlueprint workflowBlueprint, WorkflowInstance workflowInstance, string activityType, object? input = default, CancellationToken cancellationToken = default); Task InterruptActivityTypeAsync(WorkflowInstance workflowInstance, string activityType, object? input = default, CancellationToken cancellationToken = default); + Task> InterruptActivityTypeAsync(string activityType, object? input = default, CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Services/WorkflowInterruptor.cs b/src/core/Elsa.Core/Services/WorkflowInterruptor.cs index a051f390c..9fe4903ac 100644 --- a/src/core/Elsa.Core/Services/WorkflowInterruptor.cs +++ b/src/core/Elsa.Core/Services/WorkflowInterruptor.cs @@ -1,7 +1,10 @@ -using System.Linq; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; using Elsa.Exceptions; +using Elsa.Extensions; using Elsa.Models; using Elsa.Services.Models; @@ -11,11 +14,13 @@ namespace Elsa.Services { private readonly IWorkflowRunner _workflowRunner; private readonly IWorkflowRegistry _workflowRegistry; + private readonly IWorkflowInstanceManager _workflowInstanceManager; - public WorkflowInterruptor(IWorkflowRunner workflowRunner, IWorkflowRegistry workflowRegistry) + public WorkflowInterruptor(IWorkflowRunner workflowRunner, IWorkflowRegistry workflowRegistry, IWorkflowInstanceManager workflowInstanceManager) { _workflowRunner = workflowRunner; _workflowRegistry = workflowRegistry; + _workflowInstanceManager = workflowInstanceManager; } public async Task InterruptActivityAsync(WorkflowInstance workflowInstance, string activityId, object? input, CancellationToken cancellationToken) @@ -53,7 +58,18 @@ namespace Elsa.Services return await InterruptActivityTypeAsync(workflowBlueprint!, workflowInstance, activityType, input, cancellationToken); } - private async Task GetWorkflowBlueprintAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken) => + public async Task> InterruptActivityTypeAsync(string activityType, object? input, CancellationToken cancellationToken) => + await InterruptActivityTypeInternalAsync(activityType, input, cancellationToken).ToListAsync(cancellationToken); + + private async Task GetWorkflowBlueprintAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken) => await _workflowRegistry.GetWorkflowAsync(workflowInstance.WorkflowDefinitionId, workflowInstance.TenantId, VersionOptions.SpecificVersion(workflowInstance.Version), cancellationToken); + + private async IAsyncEnumerable InterruptActivityTypeInternalAsync(string activityType, object? input, [EnumeratorCancellation] CancellationToken cancellationToken) + { + var workflowInstances = await _workflowInstanceManager.ListByBlockingActivityTypeAsync(activityType, cancellationToken); + + foreach (var workflowInstance in workflowInstances) + yield return await InterruptActivityTypeAsync(workflowInstance, activityType, input, cancellationToken); + } } } \ No newline at end of file diff --git a/src/samples/aspnet/Elsa.Samples.CustomActivityTypeProviders/Endpoints/WakeUp.cs b/src/samples/aspnet/Elsa.Samples.CustomActivityTypeProviders/Endpoints/WakeUp.cs index 6b4223d45..f5e2435e0 100644 --- a/src/samples/aspnet/Elsa.Samples.CustomActivityTypeProviders/Endpoints/WakeUp.cs +++ b/src/samples/aspnet/Elsa.Samples.CustomActivityTypeProviders/Endpoints/WakeUp.cs @@ -1,7 +1,6 @@ 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; @@ -13,25 +12,18 @@ namespace Elsa.Samples.CustomActivityTypeProviders.Endpoints public class WakeUp : Controller { private readonly IWorkflowInterruptor _workflowInterruptor; - private readonly IWorkflowInstanceManager _workflowInstanceManager; - public WakeUp(IWorkflowInterruptor workflowInterruptor, IWorkflowInstanceManager workflowInstanceManager) + public WakeUp(IWorkflowInterruptor workflowInterruptor) { _workflowInterruptor = workflowInterruptor; - _workflowInstanceManager = workflowInstanceManager; } [HttpGet] public async Task 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."); + var workflowInstances = (await _workflowInterruptor.InterruptActivityTypeAsync(nameof(Sleep), cancellationToken: cancellationToken)).ToList(); + return Ok($"Interrupted {workflowInstances.Count} workflows."); } } } \ No newline at end of file