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

Extract workflow dispatch logic into reusable mediator command handlers

Browse source
This commit is contained in:
Sipke Schoorstra 2021-04-01 19:20:32 +02:00
parent ac61b7fc70
commit 53877b68fc
14 changed files with 256 additions and 420 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1,9 +1,10 @@
using Elsa.Bookmarks;
using MediatR;
namespace Elsa.Dispatch
{
public record TriggerWorkflowsRequest(string ActivityType, IBookmark Bookmark, IBookmark Trigger, object? Input = default, string? CorrelationId = default, string? ContextId = default, string? TenantId = default);
public record ExecuteWorkflowDefinitionRequest(string WorkflowDefinitionId, string? ActivityId = default, object? Input = default, string? CorrelationId = default, string? ContextId = default, string? TenantId = default);
public record ExecuteWorkflowInstanceRequest(string WorkflowInstanceId, string ActivityId, object? Input = default);
public record TriggerWorkflowsRequest(string ActivityType, IBookmark Bookmark, IBookmark Trigger, object? Input = default, string? CorrelationId = default, string? ContextId = default, string? TenantId = default) : IRequest<Unit>;
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>;
}

57
src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowDefinitionRequestConsumer.cs
View file

@ -1,62 +1,13 @@
using System.Threading.Tasks;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
using Elsa.Persistence.Specifications.WorkflowInstances;
using Elsa.Services;
using Elsa.Services.Models;
using Microsoft.Extensions.Logging;
using MediatR;
using Rebus.Handlers;
namespace Elsa.Dispatch.Consumers
{
public class ExecuteWorkflowDefinitionRequestConsumer : IHandleMessages<ExecuteWorkflowDefinitionRequest>
{
private readonly IStartsWorkflow _startsWorkflow;
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly ILogger _logger;
public ExecuteWorkflowDefinitionRequestConsumer(
IStartsWorkflow startsWorkflow,
IWorkflowRegistry workflowRegistry,
IWorkflowInstanceStore workflowInstanceStore,
ILogger<ExecuteWorkflowDefinitionRequestConsumer> logger)
{
_startsWorkflow = startsWorkflow;
_workflowRegistry = workflowRegistry;
_workflowInstanceStore = workflowInstanceStore;
_logger = logger;
}
public async Task Handle(ExecuteWorkflowDefinitionRequest message)
{
var workflowDefinitionId = message.WorkflowDefinitionId;
var tenantId = message.TenantId;
var workflowBlueprint = await _workflowRegistry.GetAsync(workflowDefinitionId, tenantId, VersionOptions.Published);
if (!ValidatePreconditions(workflowDefinitionId, workflowBlueprint))
return;
if (!workflowBlueprint!.IsSingleton || await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId) == false)
await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, message.ActivityId, message.Input, message.CorrelationId, message.ContextId);
}
private bool ValidatePreconditions(string? workflowDefinitionId, IWorkflowBlueprint? workflowBlueprint)
{
if (workflowBlueprint == null)
{
_logger.LogWarning("No workflow definition {WorkflowDefinitionId} found. Make sure the scheduled workflow definition is published and enabled", workflowDefinitionId);
return false;
}
return true;
}
private async Task<bool> GetWorkflowIsAlreadyExecutingAsync(string? tenantId, string workflowDefinitionId)
{
var specification = new TenantSpecification<WorkflowInstance>(tenantId).WithWorkflowDefinition(workflowDefinitionId).And(new WorkflowIsAlreadyExecutingSpecification());
return await _workflowInstanceStore.FindAsync(specification) != null;
}
private readonly IMediator _mediator;
public ExecuteWorkflowDefinitionRequestConsumer(IMediator mediator) => _mediator = mediator;
public async Task Handle(ExecuteWorkflowDefinitionRequest message) => await _mediator.Send(message);
}
}

52
src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowInstanceRequestConsumer.cs
View file

@ -1,10 +1,8 @@
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;
using Elsa.DistributedLocking;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Services;
using MediatR;
using Microsoft.Extensions.Logging;
using NodaTime;
using Rebus.Handlers;
@ -13,22 +11,19 @@ namespace Elsa.Dispatch.Consumers
{
public class ExecuteWorkflowInstanceRequestConsumer : IHandleMessages<ExecuteWorkflowInstanceRequest>
{
private readonly IResumesWorkflow _workflowRunner;
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IMediator _mediator;
private readonly IDistributedLockProvider _distributedLockProvider;
private readonly ICommandSender _commandSender;
private readonly ILogger _logger;
private readonly Stopwatch _stopwatch = new();
public ExecuteWorkflowInstanceRequestConsumer(
IResumesWorkflow workflowRunner,
IWorkflowInstanceStore workflowInstanceStore,
IMediator mediator,
IDistributedLockProvider distributedLockProvider,
ICommandSender commandSender,
ILogger<ExecuteWorkflowInstanceRequestConsumer> logger)
{
_workflowRunner = workflowRunner;
_workflowInstanceStore = workflowInstanceStore;
_mediator = mediator;
_distributedLockProvider = distributedLockProvider;
_commandSender = commandSender;
_logger = logger;
@ -51,15 +46,7 @@ namespace Elsa.Dispatch.Consumers
try
{
var workflowInstance = await _workflowInstanceStore.FindByIdAsync(message.WorkflowInstanceId);
if (!ValidatePreconditions(workflowInstanceId, workflowInstance, message.ActivityId))
return;
await _workflowRunner.ResumeWorkflowAsync(
workflowInstance!,
message.ActivityId,
message.Input);
await _mediator.Send(message);
}
finally
{
@ -68,34 +55,5 @@ namespace Elsa.Dispatch.Consumers
_logger.LogDebug("Held lock on workflow instance {WorkflowInstanceId} for {ElapsedTime}", workflowInstanceId, _stopwatch.Elapsed);
}
}
private bool ValidatePreconditions(string? workflowInstanceId, WorkflowInstance? workflowInstance, string? activityId)
{
if (workflowInstance == null)
{
_logger.LogWarning("Could not run workflow instance with ID {WorkflowInstanceId} because it does not exist", workflowInstanceId);
return false;
}
if (workflowInstance.WorkflowStatus != WorkflowStatus.Suspended && workflowInstance.WorkflowStatus != WorkflowStatus.Running)
{
_logger.LogWarning("Could not run workflow instance with ID {WorkflowInstanceId} because it has a status other than Suspended or Running. Its actual status is {WorkflowStatus}", workflowInstanceId, workflowInstance.WorkflowStatus);
return false;
}
if (activityId != null)
{
var activityIsBlocking = workflowInstance.BlockingActivities.Any(x => x.ActivityId == activityId);
var activityIsScheduled = workflowInstance.ScheduledActivities.Any(x => x.ActivityId == activityId) || workflowInstance.CurrentActivity?.ActivityId == activityId;
if (!activityIsBlocking && !activityIsScheduled)
{
_logger.LogWarning("Did not run workflow {WorkflowInstanceId} for activity {ActivityId} because the workflow is not blocked on that activity nor is that activity scheduled for execution", workflowInstanceId, activityId);
return false;
}
}
return true;
}
}
}

85
src/core/Elsa.Core/Dispatch/Consumers/TriggerWorkflowsRequestConsumer.cs
View file

@ -1,90 +1,13 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Elsa.Bookmarks;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
using Elsa.Services;
using Elsa.Triggers;
using Microsoft.Extensions.Logging;
using Open.Linq.AsyncExtensions;
using MediatR;
using Rebus.Handlers;
namespace Elsa.Dispatch.Consumers
{
public class TriggerWorkflowsRequestConsumer : IHandleMessages<TriggerWorkflowsRequest>
{
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IBookmarkFinder _bookmarkFinder;
private readonly ITriggerFinder _triggerFinder;
private readonly ICommandSender _commandSender;
private readonly ILogger _logger;
public TriggerWorkflowsRequestConsumer(
IWorkflowInstanceStore workflowInstanceStore,
IBookmarkFinder bookmarkFinder,
ITriggerFinder triggerFinder,
ICommandSender commandSender,
ILogger<TriggerWorkflowsRequestConsumer> logger)
{
_workflowInstanceStore = workflowInstanceStore;
_bookmarkFinder = bookmarkFinder;
_triggerFinder = triggerFinder;
_commandSender = commandSender;
_logger = logger;
}
public async Task Handle(TriggerWorkflowsRequest message)
{
var correlationId = message.CorrelationId;
// Find correlated workflows.
var correlatedWorkflowInstances = await _workflowInstanceStore.FindManyAsync(new CorrelationIdSpecification<WorkflowInstance>(correlationId)).ToList();
if (correlatedWorkflowInstances.Count > 0)
{
_logger.LogDebug("{WorkflowInstanceCount} existing workflows found with correlation ID '{CorrelationId}' will be queued for execution", correlatedWorkflowInstances.Count, correlationId);
var correlatedWorkflowInstanceIds = correlatedWorkflowInstances.Select(x => x.Id).ToHashSet();
var bookmarkFinderResults = await _bookmarkFinder.FindBookmarksAsync(message.ActivityType, message.Bookmark, message.TenantId).Where(x => correlatedWorkflowInstanceIds.Contains(x.WorkflowInstanceId)).ToList();
await EnqueueWorkflowsAsync(bookmarkFinderResults, message.Input);
return;
}
// No correlated workflows found, so go ahead and start new & resume existing workflows.
_logger.LogDebug("No workflows found with correlation ID '{CorrelationId}'. Starting new and resuming existing workflows", correlationId);
await StartWorkflowsAsync(message);
await ResumeWorkflowsAsync(message);
}
private async Task StartWorkflowsAsync(TriggerWorkflowsRequest message)
{
var filter = message.Trigger;
var results = await _triggerFinder.FindTriggersAsync(message.ActivityType, filter, message.TenantId);
foreach (var result in results)
{
var workflowBlueprint = result.WorkflowBlueprint;
await EnqueueWorkflowDefinition(workflowBlueprint.Id, workflowBlueprint.TenantId, result.ActivityId, message.Input, message.CorrelationId, message.ContextId);
}
}
private async Task ResumeWorkflowsAsync(TriggerWorkflowsRequest message)
{
var filter = message.Bookmark;
var results = await _bookmarkFinder.FindBookmarksAsync(message.ActivityType, filter, message.TenantId);
await EnqueueWorkflowsAsync(results, message.Input);
}
private async Task EnqueueWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input)
{
foreach (var result in results)
await EnqueueWorkflowInstance(result.WorkflowInstanceId, result.ActivityId, input);
}
public async Task EnqueueWorkflowInstance(string workflowInstanceId, string activityId, object? input) => await _commandSender.SendAsync(new ExecuteWorkflowInstanceRequest(workflowInstanceId, activityId, input));
public async Task EnqueueWorkflowDefinition(string workflowDefinitionId, string? tenantId, string activityId, object? input, string? correlationId, string? contextId) =>
await _commandSender.SendAsync(new ExecuteWorkflowDefinitionRequest(workflowDefinitionId, activityId, input, correlationId, contextId, tenantId));
private readonly IMediator _mediator;
public TriggerWorkflowsRequestConsumer(IMediator mediator) => _mediator = mediator;
public async Task Handle(TriggerWorkflowsRequest message) => await _mediator.Send(message);
}
}

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

@ -0,0 +1,65 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
using Elsa.Persistence.Specifications.WorkflowInstances;
using Elsa.Services;
using Elsa.Services.Models;
using MediatR;
using Microsoft.Extensions.Logging;
namespace Elsa.Dispatch.Handlers
{
public class ExecuteWorkflowDefinition : IRequestHandler<ExecuteWorkflowDefinitionRequest>
{
private readonly IStartsWorkflow _startsWorkflow;
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly ILogger _logger;
public ExecuteWorkflowDefinition(
IStartsWorkflow startsWorkflow,
IWorkflowRegistry workflowRegistry,
IWorkflowInstanceStore workflowInstanceStore,
ILogger<ExecuteWorkflowDefinition> logger)
{
_startsWorkflow = startsWorkflow;
_workflowRegistry = workflowRegistry;
_workflowInstanceStore = workflowInstanceStore;
_logger = logger;
}
public async Task<Unit> 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;
if (!workflowBlueprint!.IsSingleton || await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId) == false)
await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken);
return Unit.Value;
}
private bool ValidatePreconditions(string? workflowDefinitionId, IWorkflowBlueprint? workflowBlueprint)
{
if (workflowBlueprint == null)
{
_logger.LogWarning("No workflow definition {WorkflowDefinitionId} found. Make sure the scheduled workflow definition is published and enabled", workflowDefinitionId);
return false;
}
return true;
}
private async Task<bool> GetWorkflowIsAlreadyExecutingAsync(string? tenantId, string workflowDefinitionId)
{
var specification = new TenantSpecification<WorkflowInstance>(tenantId).WithWorkflowDefinition(workflowDefinitionId).And(new WorkflowIsAlreadyExecutingSpecification());
return await _workflowInstanceStore.FindAsync(specification) != null;
}
}
}

70
src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowInstance.cs Normal file
View file

@ -0,0 +1,70 @@
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Services;
using MediatR;
using Microsoft.Extensions.Logging;
namespace Elsa.Dispatch.Handlers
{
public class ExecuteWorkflowInstance : IRequestHandler<ExecuteWorkflowInstanceRequest>
{
private readonly IResumesWorkflow _workflowRunner;
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly ILogger _logger;
public ExecuteWorkflowInstance(IResumesWorkflow workflowRunner, IWorkflowInstanceStore workflowInstanceStore, ILogger<ExecuteWorkflowInstance> logger)
{
_workflowRunner = workflowRunner;
_workflowInstanceStore = workflowInstanceStore;
_logger = logger;
}
public async Task<Unit> Handle(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken)
{
var workflowInstanceId = request.WorkflowInstanceId;
var workflowInstance = await _workflowInstanceStore.FindByIdAsync(request.WorkflowInstanceId, cancellationToken: cancellationToken);
if (!ValidatePreconditions(workflowInstanceId, workflowInstance, request.ActivityId))
return Unit.Value;
await _workflowRunner.ResumeWorkflowAsync(
workflowInstance!,
request.ActivityId,
request.Input, cancellationToken);
return Unit.Value;
}
private bool ValidatePreconditions(string? workflowInstanceId, WorkflowInstance? workflowInstance, string? activityId)
{
if (workflowInstance == null)
{
_logger.LogWarning("Could not run workflow instance with ID {WorkflowInstanceId} because it does not exist", workflowInstanceId);
return false;
}
if (workflowInstance.WorkflowStatus != WorkflowStatus.Suspended && workflowInstance.WorkflowStatus != WorkflowStatus.Running)
{
_logger.LogWarning(
"Could not run workflow instance with ID {WorkflowInstanceId} because it has a status other than Suspended or Running. Its actual status is {WorkflowStatus}",
workflowInstanceId, workflowInstance.WorkflowStatus);
return false;
}
if (activityId == null)
return true;
var activityIsBlocking = workflowInstance.BlockingActivities.Any(x => x.ActivityId == activityId);
var activityIsScheduled = workflowInstance.ScheduledActivities.Any(x => x.ActivityId == activityId) || workflowInstance.CurrentActivity?.ActivityId == activityId;
if (activityIsBlocking || activityIsScheduled)
return true;
_logger.LogWarning("Did not run workflow {WorkflowInstanceId} for activity {ActivityId} because the workflow is not blocked on that activity nor is that activity scheduled for execution", workflowInstanceId, activityId);
return false;
}
}
}

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

@ -0,0 +1,78 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Bookmarks;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
using Elsa.Triggers;
using MediatR;
using Microsoft.Extensions.Logging;
using Open.Linq.AsyncExtensions;
namespace Elsa.Dispatch.Handlers
{
public class TriggerWorkflows : IRequestHandler<TriggerWorkflowsRequest>
{
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IBookmarkFinder _bookmarkFinder;
private readonly ITriggerFinder _triggerFinder;
private readonly IWorkflowDefinitionDispatcher _workflowDefinitionDispatcher;
private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher;
private readonly ILogger<TriggerWorkflows> _logger;
public TriggerWorkflows(
IWorkflowInstanceStore workflowInstanceStore,
IBookmarkFinder bookmarkFinder,
ITriggerFinder triggerFinder,
IWorkflowDefinitionDispatcher workflowDefinitionDispatcher,
IWorkflowInstanceDispatcher workflowInstanceDispatcher,
ILogger<TriggerWorkflows> logger)
{
_workflowInstanceStore = workflowInstanceStore;
_bookmarkFinder = bookmarkFinder;
_triggerFinder = triggerFinder;
_workflowDefinitionDispatcher = workflowDefinitionDispatcher;
_workflowInstanceDispatcher = workflowInstanceDispatcher;
_logger = logger;
}
public async Task<Unit> Handle(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
{
var correlationId = request.CorrelationId;
var correlatedWorkflowInstanceCount = await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification<WorkflowInstance>(correlationId), cancellationToken);
if (correlatedWorkflowInstanceCount > 0)
{
_logger.LogDebug("{WorkflowInstanceCount} existing workflows found with correlation ID '{CorrelationId}' will be queued for execution", correlatedWorkflowInstanceCount, correlationId);
var existingWorkflows = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, request.TenantId, cancellationToken).ToList();
await ResumeWorkflowsAsync(existingWorkflows, request.Input, cancellationToken);
}
else
{
_logger.LogDebug("No existing workflows found with correlation ID '{CorrelationId}'. Starting new workflow", correlationId);
await StartWorkflowsAsync(request, cancellationToken);
}
return Unit.Value;
}
private async Task StartWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
{
var filter = request.Trigger;
var triggers = await _triggerFinder.FindTriggersAsync(request.ActivityType, filter, request.TenantId, cancellationToken);
foreach (var trigger in triggers)
{
var workflowBlueprint = trigger.WorkflowBlueprint;
await _workflowDefinitionDispatcher.DispatchAsync(new ExecuteWorkflowDefinitionRequest(workflowBlueprint.Id, trigger.ActivityId, request.Input, request.CorrelationId, request.ContextId, workflowBlueprint.TenantId), cancellationToken);
}
}
private async Task ResumeWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input, CancellationToken cancellationToken)
{
foreach (var result in results)
await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(result.WorkflowInstanceId, result.ActivityId, input), cancellationToken);
}
}
}

1
src/server/Elsa.Server.Hangfire/Dispatch/HangfireWorkflowDispatcher.cs
View file

@ -6,7 +6,6 @@ using Elsa.Dispatch;
using Elsa.Server.Hangfire.Jobs;
using Hangfire;
using Hangfire.States;
using Humanizer;
namespace Elsa.Server.Hangfire.Dispatch
{

72
src/server/Elsa.Server.Hangfire/Jobs/CorrelatedWorkflowDefinitionJob.cs
View file

@ -1,76 +1,14 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Bookmarks;
using Elsa.Dispatch;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
using Elsa.Triggers;
using Microsoft.Extensions.Logging;
using Open.Linq.AsyncExtensions;
using MediatR;
namespace Elsa.Server.Hangfire.Jobs
{
public class CorrelatedWorkflowDefinitionJob
{
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IBookmarkFinder _bookmarkFinder;
private readonly ITriggerFinder _triggerFinder;
private readonly IWorkflowDefinitionDispatcher _workflowDefinitionDispatcher;
private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher;
private readonly ILogger<CorrelatedWorkflowDefinitionJob> _logger;
public CorrelatedWorkflowDefinitionJob(
IWorkflowInstanceStore workflowInstanceStore,
IBookmarkFinder bookmarkFinder,
ITriggerFinder triggerFinder,
IWorkflowDefinitionDispatcher workflowDefinitionDispatcher,
IWorkflowInstanceDispatcher workflowInstanceDispatcher,
ILogger<CorrelatedWorkflowDefinitionJob> logger)
{
_workflowInstanceStore = workflowInstanceStore;
_bookmarkFinder = bookmarkFinder;
_triggerFinder = triggerFinder;
_workflowDefinitionDispatcher = workflowDefinitionDispatcher;
_workflowInstanceDispatcher = workflowInstanceDispatcher;
_logger = logger;
}
public async Task ExecuteAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken = default)
{
var correlationId = request.CorrelationId;
var correlatedWorkflowInstanceCount = await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification<WorkflowInstance>(correlationId), cancellationToken);
if (correlatedWorkflowInstanceCount > 0)
{
_logger.LogDebug("{WorkflowInstanceCount} existing workflows found with correlation ID '{CorrelationId}' will be queued for execution", correlatedWorkflowInstanceCount, correlationId);
var existingWorkflows = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, request.TenantId, cancellationToken).ToList();
await ResumeWorkflowsAsync(existingWorkflows, request.Input, cancellationToken);
}
else
{
_logger.LogDebug("No existing workflows found with correlation ID '{CorrelationId}'. Starting new workflow", correlationId);
await StartWorkflowsAsync(request, cancellationToken);
}
}
private async Task StartWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
{
var filter = request.Trigger;
var triggers = await _triggerFinder.FindTriggersAsync(request.ActivityType, filter, request.TenantId, cancellationToken);
foreach (var trigger in triggers)
{
var workflowBlueprint = trigger.WorkflowBlueprint;
await _workflowDefinitionDispatcher.DispatchAsync(new ExecuteWorkflowDefinitionRequest(workflowBlueprint.Id, trigger.ActivityId, request.Input, request.CorrelationId, request.ContextId, workflowBlueprint.TenantId), cancellationToken);
}
}
private async Task ResumeWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input, CancellationToken cancellationToken)
{
foreach (var result in results)
await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(result.WorkflowInstanceId, result.ActivityId, input), cancellationToken);
}
private readonly IMediator _mediator;
public CorrelatedWorkflowDefinitionJob(IMediator mediator) => _mediator = mediator;
public async Task ExecuteAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken = default) => await _mediator.Send(request, cancellationToken);
}
}

30
src/server/Elsa.Server.Hangfire/Jobs/WorkflowDefinitionJob.cs
View file

@ -1,36 +1,14 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Elsa.Models;
using Elsa.Services;
using Microsoft.Extensions.Logging;
using MediatR;
namespace Elsa.Server.Hangfire.Jobs
{
public class WorkflowDefinitionJob
{
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IStartsWorkflow _startsWorkflow;
private readonly ILogger<WorkflowDefinitionJob> _logger;
public WorkflowDefinitionJob(IWorkflowRegistry workflowRegistry, IStartsWorkflow startsWorkflow, ILogger<WorkflowDefinitionJob> logger)
{
_workflowRegistry = workflowRegistry;
_startsWorkflow = startsWorkflow;
_logger = logger;
}
public async Task ExecuteAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default)
{
var workflowBlueprint = await _workflowRegistry.GetWorkflowAsync(request.WorkflowDefinitionId, request.TenantId, VersionOptions.Published, cancellationToken);
if (workflowBlueprint == null)
{
_logger.LogWarning("No published workflow definition {WorkflowDefinitionId} found", request.WorkflowDefinitionId);
return;
}
await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken);
}
private readonly IMediator _mediator;
public WorkflowDefinitionJob(IMediator mediator) => _mediator = mediator;
public async Task ExecuteAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) => await _mediator.Send(request, cancellationToken);
}
}

27
src/server/Elsa.Server.Hangfire/Jobs/WorkflowInstanceJob.cs
View file

@ -3,34 +3,15 @@ using System.Threading.Tasks;
using Elsa.Dispatch;
using Elsa.Persistence;
using Elsa.Services;
using MediatR;
using Microsoft.Extensions.Logging;
namespace Elsa.Server.Hangfire.Jobs
{
public class WorkflowInstanceJob
{
private readonly IWorkflowInstanceStore _store;
private readonly IResumesWorkflow _workflowRunner;
private readonly ILogger<WorkflowInstanceJob> _logger;
public WorkflowInstanceJob(IWorkflowInstanceStore store, IResumesWorkflow workflowRunner, ILogger<WorkflowInstanceJob> logger)
{
_store = store;
_workflowRunner = workflowRunner;
_logger = logger;
}
public async Task ExecuteAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
{
var workflowInstance = await _store.FindByIdAsync(request.WorkflowInstanceId, cancellationToken);
if(workflowInstance == null)
{
_logger.LogWarning("Workflow instance {WorkflowInstanceId} not found", request.WorkflowInstanceId);
return;
}
await _workflowRunner.ResumeWorkflowAsync(workflowInstance, request.ActivityId, request.Input, cancellationToken);
}
private readonly IMediator _mediator;
public WorkflowInstanceJob(IMediator mediator) => _mediator = mediator;
public async Task ExecuteAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default) => await _mediator.Send(request, cancellationToken);
}
}

72
src/server/Elsa.Server.Orleans/Grains/CorrelatedWorkflowDefinitionGrain.cs
View file

@ -1,15 +1,8 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Bookmarks;
using Elsa.Dispatch;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
using Elsa.Server.Orleans.Grains.Contracts;
using Elsa.Triggers;
using Microsoft.Extensions.Logging;
using Open.Linq.AsyncExtensions;
using MediatR;
using Orleans;
using Orleans.Concurrency;
@ -18,63 +11,8 @@ namespace Elsa.Server.Orleans.Grains
[StatelessWorker]
public class CorrelatedWorkflowDefinitionGrain : Grain, ICorrelatedWorkflowGrain
{
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IBookmarkFinder _bookmarkFinder;
private readonly ITriggerFinder _triggerFinder;
private readonly IWorkflowDefinitionDispatcher _workflowDefinitionDispatcher;
private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher;
private readonly ILogger<CorrelatedWorkflowDefinitionGrain> _logger;
public CorrelatedWorkflowDefinitionGrain(
IWorkflowInstanceStore workflowInstanceStore,
IBookmarkFinder bookmarkFinder,
ITriggerFinder triggerFinder,
IWorkflowDefinitionDispatcher workflowDefinitionDispatcher,
IWorkflowInstanceDispatcher workflowInstanceDispatcher,
ILogger<CorrelatedWorkflowDefinitionGrain> logger)
{
_workflowInstanceStore = workflowInstanceStore;
_bookmarkFinder = bookmarkFinder;
_triggerFinder = triggerFinder;
_workflowDefinitionDispatcher = workflowDefinitionDispatcher;
_workflowInstanceDispatcher = workflowInstanceDispatcher;
_logger = logger;
}
public async Task ExecutedCorrelatedWorkflowAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken = default)
{
var correlationId = request.CorrelationId;
var correlatedWorkflowInstanceCount = await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification<WorkflowInstance>(correlationId), cancellationToken);
if (correlatedWorkflowInstanceCount > 0)
{
_logger.LogDebug("{WorkflowInstanceCount} existing workflows found with correlation ID '{CorrelationId}' will be queued for execution", correlatedWorkflowInstanceCount, correlationId);
var existingWorkflows = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, request.TenantId, cancellationToken).ToList();
await ResumeWorkflowsAsync(existingWorkflows, request.Input, cancellationToken);
}
else
{
_logger.LogDebug("No existing workflows found with correlation ID '{CorrelationId}'. Starting new workflow", correlationId);
await StartWorkflowsAsync(request, cancellationToken);
}
}
private async Task StartWorkflowsAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken)
{
var filter = request.Trigger;
var triggers = await _triggerFinder.FindTriggersAsync(request.ActivityType, filter, request.TenantId, cancellationToken);
foreach (var trigger in triggers)
{
var workflowBlueprint = trigger.WorkflowBlueprint;
await _workflowDefinitionDispatcher.DispatchAsync(new ExecuteWorkflowDefinitionRequest(workflowBlueprint.Id, trigger.ActivityId, request.Input, request.CorrelationId, request.ContextId, workflowBlueprint.TenantId), cancellationToken);
}
}
private async Task ResumeWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input, CancellationToken cancellationToken)
{
foreach (var result in results)
await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(result.WorkflowInstanceId, result.ActivityId, input), cancellationToken);
}
private readonly IMediator _mediator;
public CorrelatedWorkflowDefinitionGrain(IMediator mediator) => _mediator = mediator;
public async Task ExecutedCorrelatedWorkflowAsync(TriggerWorkflowsRequest request, CancellationToken cancellationToken = default) => await _mediator.Send(request, cancellationToken);
}
}

30
src/server/Elsa.Server.Orleans/Grains/WorkflowDefinitionGrain.cs
View file

@ -1,10 +1,8 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Elsa.Models;
using Elsa.Server.Orleans.Grains.Contracts;
using Elsa.Services;
using Microsoft.Extensions.Logging;
using MediatR;
using Orleans;
using Orleans.Concurrency;
@ -13,28 +11,8 @@ namespace Elsa.Server.Orleans.Grains
[StatelessWorker]
public class WorkflowDefinitionGrain : Grain, IWorkflowDefinitionGrain
{
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IStartsWorkflow _startsWorkflow;
private readonly ILogger<WorkflowDefinitionGrain> _logger;
public WorkflowDefinitionGrain(IWorkflowRegistry workflowRegistry, IStartsWorkflow startsWorkflow, ILogger<WorkflowDefinitionGrain> logger)
{
_workflowRegistry = workflowRegistry;
_startsWorkflow = startsWorkflow;
_logger = logger;
}
public async Task ExecuteWorkflowAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default)
{
var workflowBlueprint = await _workflowRegistry.GetWorkflowAsync(request.WorkflowDefinitionId, request.TenantId, VersionOptions.Published, cancellationToken);
if (workflowBlueprint == null)
{
_logger.LogWarning("No published workflow definition {WorkflowDefinitionId} found", request.WorkflowDefinitionId);
return;
}
await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken);
}
private readonly IMediator _mediator;
public WorkflowDefinitionGrain(IMediator mediator) => _mediator = mediator;
public async Task ExecuteWorkflowAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) => await _mediator.Send(request, cancellationToken);
}
}

30
src/server/Elsa.Server.Orleans/Grains/WorkflowInstanceGrain.cs
View file

@ -1,10 +1,8 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Elsa.Persistence;
using Elsa.Server.Orleans.Grains.Contracts;
using Elsa.Services;
using Microsoft.Extensions.Logging;
using MediatR;
using Orleans;
using Orleans.Concurrency;
@ -13,28 +11,8 @@ namespace Elsa.Server.Orleans.Grains
[StatelessWorker]
public class WorkflowInstanceGrain : Grain, IWorkflowInstanceGrain
{
private readonly IWorkflowInstanceStore _store;
private readonly IResumesWorkflow _workflowRunner;
private readonly ILogger<WorkflowInstanceGrain> _logger;
public WorkflowInstanceGrain(IWorkflowInstanceStore store, IResumesWorkflow workflowRunner, ILogger<WorkflowInstanceGrain> logger)
{
_store = store;
_workflowRunner = workflowRunner;
_logger = logger;
}
public async Task ExecuteWorkflowAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
{
var workflowInstance = await _store.FindByIdAsync(request.WorkflowInstanceId, cancellationToken);
if(workflowInstance == null)
{
_logger.LogWarning("Workflow instance {WorkflowInstanceId} not found", request.WorkflowInstanceId);
return;
}
await _workflowRunner.ResumeWorkflowAsync(workflowInstance, request.ActivityId, request.Input, cancellationToken);
}
private readonly IMediator _mediator;
public WorkflowInstanceGrain(IMediator mediator) => _mediator = mediator;
public async Task ExecuteWorkflowAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default) => await _mediator.Send(request, cancellationToken);
}
}