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

Incremental work on async HTTP workflow execution

Browse source
This commit is contained in:
Sipke Schoorstra 2021-05-15 22:17:25 +02:00
parent e95e43bf36
commit f79e696d74
5 changed files with 65 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
src/activities/Elsa.Activities.Http/Extensions/HttpRequestExtensions.cs
View file

@ -54,5 +54,16 @@ namespace Elsa.Activities.Http.Extensions
correlationId = null;
return false;
}
public static bool GetUseDispatch(this HttpRequest request)
{
if (request.Query.ContainsKey("x-dispatch"))
return true;
if (request.Headers.ContainsKey("X-Dispatch"))
return true;
return false;
}
}
}

14
src/activities/Elsa.Activities.Http/Middleware/HttpRequestMiddleware.cs
View file

@ -15,19 +15,27 @@ namespace Elsa.Activities.Http.Middleware
public HttpEndpointMiddleware(RequestDelegate next) => _next = next;
public async Task InvokeAsync(HttpContext httpContext, IMediator mediator)
public async Task InvokeAsync(HttpContext httpContext, IMediator mediator, IWorkflowInstanceDispatcher workflowInstanceDispatcher)
{
var path = httpContext.Request.Path.Value.ToLowerInvariant();
var method = httpContext.Request.Method!.ToLowerInvariant();
var cancellationToken = httpContext.RequestAborted;
httpContext.Request.TryGetCorrelationId(out var correlationId);
var useDispatch = httpContext.Request.GetUseDispatch();
const string activityType = nameof(HttpEndpoint);
var trigger = new HttpEndpointBookmark(path, method, null);
var bookmark = new HttpEndpointBookmark(path, method, correlationId?.ToLowerInvariant());
var workflowInstanceCount = await mediator.Send(new TriggerWorkflowsRequest(activityType, bookmark, trigger, default, correlationId, default, TenantId), cancellationToken);
var triggerRequest = new TriggerWorkflowsRequest(activityType, bookmark, trigger, default, correlationId, default, TenantId, Execute: !useDispatch);
var response = await mediator.Send(triggerRequest, cancellationToken);
if (workflowInstanceCount == 0)
if (useDispatch)
{
// We created workflow instances but haven't scheduled them for execution yet. Use the dispatcher to schedule them for execution.
workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest())
}
if (response.WorkflowInstanceIds.Count == 0)
{
await _next(httpContext);
}

13
src/core/Elsa.Abstractions/Dispatch/Models.cs
View file

@ -1,10 +1,15 @@
using Elsa.Bookmarks;
using System.Collections.Generic;
using Elsa.Bookmarks;
using MediatR;
namespace Elsa.Dispatch
{
public record TriggerWorkflowsRequest(string ActivityType, IBookmark Bookmark, IBookmark Trigger, object? Input = default, string? CorrelationId = default, string? WorkflowInstanceId = default, string? ContextId = default, string? TenantId = default) : IRequest<int>;
public record ExecuteWorkflowDefinitionRequest(string WorkflowDefinitionId, string? ActivityId = default, object? Input = default, string? CorrelationId = default, string? ContextId = default, string? TenantId = default) : IRequest<Unit>;
public record ExecuteWorkflowInstanceRequest(string WorkflowInstanceId, string ActivityId, object? Input = default) : IRequest<Unit>;
public record TriggerWorkflowsRequest(string ActivityType, IBookmark Bookmark, IBookmark Trigger, object? Input = default, string? CorrelationId = default, string? WorkflowInstanceId = default, string? ContextId = default, string? TenantId = default, bool Execute = true) : IRequest<TriggerWorkflowsResponse>;
public record ExecuteWorkflowDefinitionRequest(string WorkflowDefinitionId, string? ActivityId = default, object? Input = default, string? CorrelationId = default, string? ContextId = default, string? TenantId = default, bool Execute = true) : IRequest<ExecuteWorkflowDefinitionResponse>;
public record ExecuteWorkflowInstanceRequest(string WorkflowInstanceId, string ActivityId, object? Input = default, bool Execute = true) : IRequest<Unit>;
public record TriggerWorkflowsResponse(ICollection<PendingWorkflow> PendingWorkflows);
public record ExecuteWorkflowDefinitionResponse(PendingWorkflow? PendingWorkflow = default);
public record PendingWorkflow(string WorkflowInstanceId, string ActivityId);
}

17
src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowDefinition.cs
View file

@ -14,7 +14,7 @@ using IDistributedLockProvider = Elsa.Services.IDistributedLockProvider;
namespace Elsa.Dispatch.Handlers
{
public class ExecuteWorkflowDefinition : IRequestHandler<ExecuteWorkflowDefinitionRequest>
public class ExecuteWorkflowDefinition : IRequestHandler<ExecuteWorkflowDefinitionRequest, ExecuteWorkflowDefinitionResponse>
{
private readonly IStartsWorkflow _startsWorkflow;
private readonly IWorkflowRegistry _workflowRegistry;
@ -39,19 +39,19 @@ namespace Elsa.Dispatch.Handlers
_logger = logger;
}
public async Task<Unit> Handle(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken)
public async Task<ExecuteWorkflowDefinitionResponse> Handle(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken)
{
var workflowDefinitionId = request.WorkflowDefinitionId;
var tenantId = request.TenantId;
var workflowBlueprint = await _workflowRegistry.GetAsync(workflowDefinitionId, tenantId, VersionOptions.Published, cancellationToken);
if (!ValidatePreconditions(workflowDefinitionId, workflowBlueprint))
return Unit.Value;
return new ExecuteWorkflowDefinitionResponse();
var lockKey = $"execute-workflow-definition:tenant:{tenantId}:workflow-definition:{workflowDefinitionId}";
var correlationId = request.CorrelationId;
var correlationLockHandle = default(IDistributedSynchronizationHandle?);
// If we are creating a correlated workflow, make sure to acquire a lock on it to prevent duplicate workflow instances from being created.
if (!string.IsNullOrWhiteSpace(correlationId))
{
_logger.LogDebug("Acquiring lock on correlation ID {CorrelationId}", correlationId);
@ -63,13 +63,18 @@ namespace Elsa.Dispatch.Handlers
try
{
// Acquire a lock on the workflow definition so that we can ensure singleton-workflows never execute more than one instance.
var lockKey = $"execute-workflow-definition:tenant:{tenantId}:workflow-definition:{workflowDefinitionId}";
await using var handle = await _distributedLockProvider.AcquireLockAsync(lockKey, _elsaOptions.DistributedLockTimeout, cancellationToken);
if (handle == null)
throw new LockAcquisitionException($"Failed to acquire a lock on {lockKey}");
if (!workflowBlueprint!.IsSingleton || await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId) == false)
await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken);
{
var workflowInstance = await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken);
return new ExecuteWorkflowDefinitionResponse(workflowInstance.Id);
}
}
finally
{
@ -77,7 +82,7 @@ namespace Elsa.Dispatch.Handlers
await correlationLockHandle.DisposeAsync();
}
return Unit.Value;
return new ExecuteWorkflowDefinitionResponse();
}
private bool ValidatePreconditions(string? workflowDefinitionId, IWorkflowBlueprint? workflowBlueprint)

46
src/core/Elsa.Core/Dispatch/Handlers/TriggerWorkflows.cs
View file

@ -2,7 +2,6 @@
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.Signaling;
using Elsa.Bookmarks;
using Elsa.Exceptions;
using Elsa.Models;
@ -17,7 +16,7 @@ using Open.Linq.AsyncExtensions;
namespace Elsa.Dispatch.Handlers
{
public class TriggerWorkflows : IRequestHandler<TriggerWorkflowsRequest, int>
public class TriggerWorkflows : IRequestHandler<TriggerWorkflowsRequest, TriggerWorkflowsResponse>
{
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IBookmarkFinder _bookmarkFinder;
@ -45,7 +44,7 @@ namespace Elsa.Dispatch.Handlers
_logger = logger;
}
public async Task<int> Handle(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
public async Task<TriggerWorkflowsResponse> Handle(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
{
var correlationId = request.CorrelationId;
@ -58,30 +57,30 @@ namespace Elsa.Dispatch.Handlers
return await TriggerWorkflowsAsync(request, cancellationToken);
}
private async Task<int> TriggerWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
private async Task<TriggerWorkflowsResponse> TriggerWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
{
var bookmarkResultsQuery = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, request.CorrelationId, request.TenantId, cancellationToken);
var bookmarkResults = bookmarkResultsQuery.ToList();
var triggeredCount = bookmarkResults.GroupBy(x => x.WorkflowInstanceId).Select(x => x.Key).Distinct().Count();
var triggeredPendingWorkflows = bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList();
await ResumeWorkflowsAsync(bookmarkResults, request.Input, request.Execute, cancellationToken);
var startWorkflowsResponse = await StartWorkflowsAsync(request, cancellationToken);
var pendingWorkflows = triggeredPendingWorkflows.Concat(startWorkflowsResponse.PendingWorkflows).Distinct().ToList();
await ResumeWorkflowsAsync(bookmarkResults, request.Input, cancellationToken);
var startedCount = await StartWorkflowsAsync(request, cancellationToken);
return startedCount + triggeredCount;
return new TriggerWorkflowsResponse(pendingWorkflows);
}
private async Task<int> ResumeSpecificWorkflowInstanceAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
private async Task<TriggerWorkflowsResponse> ResumeSpecificWorkflowInstanceAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
{
var bookmarkResultsQuery = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, request.CorrelationId, request.TenantId, cancellationToken);
bookmarkResultsQuery = bookmarkResultsQuery.Where(x => x.WorkflowInstanceId == request.WorkflowInstanceId);
var bookmarkResults = bookmarkResultsQuery.ToList();
var triggeredCount = bookmarkResults.GroupBy(x => x.WorkflowInstanceId).Select(x => x.Key).Distinct().Count();
await ResumeWorkflowsAsync(bookmarkResults, request.Input, request.Execute, cancellationToken);
var pendingWorkflows = bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList();
await ResumeWorkflowsAsync(bookmarkResults, request.Input, cancellationToken);
return triggeredCount;
return new TriggerWorkflowsResponse(pendingWorkflows);
}
private async Task<int> ResumeOrStartCorrelatedWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
private async Task<TriggerWorkflowsResponse> ResumeOrStartCorrelatedWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
{
var correlationId = request.CorrelationId!;
var lockKey = correlationId;
@ -102,37 +101,38 @@ namespace Elsa.Dispatch.Handlers
{
_logger.LogDebug("{WorkflowInstanceCount} existing workflows found with correlation ID '{CorrelationId}' will be queued for execution", correlatedWorkflowInstanceCount, correlationId);
var bookmarkResults = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, correlationId, request.TenantId, cancellationToken).ToList();
await ResumeWorkflowsAsync(bookmarkResults, request.Input, cancellationToken);
return correlatedWorkflowInstanceCount;
await ResumeWorkflowsAsync(bookmarkResults, request.Input, request.Execute, cancellationToken);
return new TriggerWorkflowsResponse(bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList());
}
}
return await StartWorkflowsAsync(request, cancellationToken);
}
private async Task<int> StartWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
private async Task<TriggerWorkflowsResponse> StartWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
{
_logger.LogDebug("Triggering workflows using {ActivityType}", request.ActivityType);
var filter = request.Trigger;
var triggers = (await _triggerFinder.FindTriggersAsync(request.ActivityType, filter, request.TenantId, cancellationToken)).ToList();
var pendingWorkflows = new List<PendingWorkflow>();
foreach (var trigger in triggers)
{
var workflowBlueprint = trigger.WorkflowBlueprint;
var response = await _mediator.Send(new ExecuteWorkflowDefinitionRequest(workflowBlueprint.Id, trigger.ActivityId, request.Input, request.CorrelationId, request.ContextId, workflowBlueprint.TenantId, request.Execute), cancellationToken);
await _mediator.Send(
new ExecuteWorkflowDefinitionRequest(workflowBlueprint.Id, trigger.ActivityId, request.Input, request.CorrelationId, request.ContextId, workflowBlueprint.TenantId),
cancellationToken);
if (response.PendingWorkflow != null)
pendingWorkflows.Add(response.PendingWorkflow);
}
return triggers.Count;
return new TriggerWorkflowsResponse(pendingWorkflows);
}
private async Task ResumeWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input, CancellationToken cancellationToken)
private async Task ResumeWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input, bool execute, CancellationToken cancellationToken)
{
foreach (var result in results)
await _mediator.Send(new ExecuteWorkflowInstanceRequest(result.WorkflowInstanceId, result.ActivityId, input), cancellationToken);
await _mediator.Send(new ExecuteWorkflowInstanceRequest(result.WorkflowInstanceId, result.ActivityId, input, execute), cancellationToken);
}
}
}