diff --git a/src/core/Elsa.Abstractions/Dispatch/Models.cs b/src/core/Elsa.Abstractions/Dispatch/Models.cs index fccfb1bd5..8a6cf3645 100644 --- a/src/core/Elsa.Abstractions/Dispatch/Models.cs +++ b/src/core/Elsa.Abstractions/Dispatch/Models.cs @@ -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; + public record ExecuteWorkflowDefinitionRequest(string WorkflowDefinitionId, string? ActivityId = default, object? Input = default, string? CorrelationId = default, string? ContextId = default, string? TenantId = default) : IRequest; + public record ExecuteWorkflowInstanceRequest(string WorkflowInstanceId, string ActivityId, object? Input = default) : IRequest; } \ No newline at end of file diff --git a/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowDefinitionRequestConsumer.cs b/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowDefinitionRequestConsumer.cs index 00ee3e27d..351555a4b 100644 --- a/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowDefinitionRequestConsumer.cs +++ b/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowDefinitionRequestConsumer.cs @@ -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 { - 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 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 GetWorkflowIsAlreadyExecutingAsync(string? tenantId, string workflowDefinitionId) - { - var specification = new TenantSpecification(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); } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowInstanceRequestConsumer.cs b/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowInstanceRequestConsumer.cs index 282798665..25a871617 100644 --- a/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowInstanceRequestConsumer.cs +++ b/src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowInstanceRequestConsumer.cs @@ -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 { - 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 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; - } } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Dispatch/Consumers/TriggerWorkflowsRequestConsumer.cs b/src/core/Elsa.Core/Dispatch/Consumers/TriggerWorkflowsRequestConsumer.cs index a110f889e..4be6dbc23 100644 --- a/src/core/Elsa.Core/Dispatch/Consumers/TriggerWorkflowsRequestConsumer.cs +++ b/src/core/Elsa.Core/Dispatch/Consumers/TriggerWorkflowsRequestConsumer.cs @@ -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 { - 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 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(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 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); } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowDefinition.cs b/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowDefinition.cs new file mode 100644 index 000000000..b46d7dde8 --- /dev/null +++ b/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowDefinition.cs @@ -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 + { + 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 logger) + { + _startsWorkflow = startsWorkflow; + _workflowRegistry = workflowRegistry; + _workflowInstanceStore = workflowInstanceStore; + _logger = logger; + } + + public async Task 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 GetWorkflowIsAlreadyExecutingAsync(string? tenantId, string workflowDefinitionId) + { + var specification = new TenantSpecification(tenantId).WithWorkflowDefinition(workflowDefinitionId).And(new WorkflowIsAlreadyExecutingSpecification()); + return await _workflowInstanceStore.FindAsync(specification) != null; + } + } +} \ No newline at end of file diff --git a/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowInstance.cs b/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowInstance.cs new file mode 100644 index 000000000..53204f905 --- /dev/null +++ b/src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowInstance.cs @@ -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 + { + private readonly IResumesWorkflow _workflowRunner; + private readonly IWorkflowInstanceStore _workflowInstanceStore; + private readonly ILogger _logger; + + public ExecuteWorkflowInstance(IResumesWorkflow workflowRunner, IWorkflowInstanceStore workflowInstanceStore, ILogger logger) + { + _workflowRunner = workflowRunner; + _workflowInstanceStore = workflowInstanceStore; + _logger = logger; + } + + public async Task 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; + } + } +} \ No newline at end of file diff --git a/src/core/Elsa.Core/Dispatch/Handlers/TriggerWorkflows.cs b/src/core/Elsa.Core/Dispatch/Handlers/TriggerWorkflows.cs new file mode 100644 index 000000000..e81795f45 --- /dev/null +++ b/src/core/Elsa.Core/Dispatch/Handlers/TriggerWorkflows.cs @@ -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 + { + private readonly IWorkflowInstanceStore _workflowInstanceStore; + private readonly IBookmarkFinder _bookmarkFinder; + private readonly ITriggerFinder _triggerFinder; + private readonly IWorkflowDefinitionDispatcher _workflowDefinitionDispatcher; + private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher; + private readonly ILogger _logger; + + public TriggerWorkflows( + IWorkflowInstanceStore workflowInstanceStore, + IBookmarkFinder bookmarkFinder, + ITriggerFinder triggerFinder, + IWorkflowDefinitionDispatcher workflowDefinitionDispatcher, + IWorkflowInstanceDispatcher workflowInstanceDispatcher, + ILogger logger) + { + _workflowInstanceStore = workflowInstanceStore; + _bookmarkFinder = bookmarkFinder; + _triggerFinder = triggerFinder; + _workflowDefinitionDispatcher = workflowDefinitionDispatcher; + _workflowInstanceDispatcher = workflowInstanceDispatcher; + _logger = logger; + } + + public async Task Handle(TriggerWorkflowsRequest request, CancellationToken cancellationToken) + { + var correlationId = request.CorrelationId; + var correlatedWorkflowInstanceCount = await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification(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 results, object? input, CancellationToken cancellationToken) + { + foreach (var result in results) + await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(result.WorkflowInstanceId, result.ActivityId, input), cancellationToken); + } + } +} \ No newline at end of file diff --git a/src/server/Elsa.Server.Hangfire/Dispatch/HangfireWorkflowDispatcher.cs b/src/server/Elsa.Server.Hangfire/Dispatch/HangfireWorkflowDispatcher.cs index 6ff52cc59..8b10efef4 100644 --- a/src/server/Elsa.Server.Hangfire/Dispatch/HangfireWorkflowDispatcher.cs +++ b/src/server/Elsa.Server.Hangfire/Dispatch/HangfireWorkflowDispatcher.cs @@ -6,7 +6,6 @@ using Elsa.Dispatch; using Elsa.Server.Hangfire.Jobs; using Hangfire; using Hangfire.States; -using Humanizer; namespace Elsa.Server.Hangfire.Dispatch { diff --git a/src/server/Elsa.Server.Hangfire/Jobs/CorrelatedWorkflowDefinitionJob.cs b/src/server/Elsa.Server.Hangfire/Jobs/CorrelatedWorkflowDefinitionJob.cs index bd834a6b6..0b9de5526 100644 --- a/src/server/Elsa.Server.Hangfire/Jobs/CorrelatedWorkflowDefinitionJob.cs +++ b/src/server/Elsa.Server.Hangfire/Jobs/CorrelatedWorkflowDefinitionJob.cs @@ -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 _logger; - - public CorrelatedWorkflowDefinitionJob( - IWorkflowInstanceStore workflowInstanceStore, - IBookmarkFinder bookmarkFinder, - ITriggerFinder triggerFinder, - IWorkflowDefinitionDispatcher workflowDefinitionDispatcher, - IWorkflowInstanceDispatcher workflowInstanceDispatcher, - ILogger 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(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 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); } } \ No newline at end of file diff --git a/src/server/Elsa.Server.Hangfire/Jobs/WorkflowDefinitionJob.cs b/src/server/Elsa.Server.Hangfire/Jobs/WorkflowDefinitionJob.cs index 402f2831b..26897ad72 100644 --- a/src/server/Elsa.Server.Hangfire/Jobs/WorkflowDefinitionJob.cs +++ b/src/server/Elsa.Server.Hangfire/Jobs/WorkflowDefinitionJob.cs @@ -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 _logger; - - public WorkflowDefinitionJob(IWorkflowRegistry workflowRegistry, IStartsWorkflow startsWorkflow, ILogger 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); } } \ No newline at end of file diff --git a/src/server/Elsa.Server.Hangfire/Jobs/WorkflowInstanceJob.cs b/src/server/Elsa.Server.Hangfire/Jobs/WorkflowInstanceJob.cs index 8a2532252..5267d748e 100644 --- a/src/server/Elsa.Server.Hangfire/Jobs/WorkflowInstanceJob.cs +++ b/src/server/Elsa.Server.Hangfire/Jobs/WorkflowInstanceJob.cs @@ -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 _logger; - - public WorkflowInstanceJob(IWorkflowInstanceStore store, IResumesWorkflow workflowRunner, ILogger 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); } } \ No newline at end of file diff --git a/src/server/Elsa.Server.Orleans/Grains/CorrelatedWorkflowDefinitionGrain.cs b/src/server/Elsa.Server.Orleans/Grains/CorrelatedWorkflowDefinitionGrain.cs index b0bb81a3f..74eeec90e 100644 --- a/src/server/Elsa.Server.Orleans/Grains/CorrelatedWorkflowDefinitionGrain.cs +++ b/src/server/Elsa.Server.Orleans/Grains/CorrelatedWorkflowDefinitionGrain.cs @@ -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 _logger; - - public CorrelatedWorkflowDefinitionGrain( - IWorkflowInstanceStore workflowInstanceStore, - IBookmarkFinder bookmarkFinder, - ITriggerFinder triggerFinder, - IWorkflowDefinitionDispatcher workflowDefinitionDispatcher, - IWorkflowInstanceDispatcher workflowInstanceDispatcher, - ILogger 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(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 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); } } \ No newline at end of file diff --git a/src/server/Elsa.Server.Orleans/Grains/WorkflowDefinitionGrain.cs b/src/server/Elsa.Server.Orleans/Grains/WorkflowDefinitionGrain.cs index 4d151ee3f..8d8ca0d8c 100644 --- a/src/server/Elsa.Server.Orleans/Grains/WorkflowDefinitionGrain.cs +++ b/src/server/Elsa.Server.Orleans/Grains/WorkflowDefinitionGrain.cs @@ -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 _logger; - - public WorkflowDefinitionGrain(IWorkflowRegistry workflowRegistry, IStartsWorkflow startsWorkflow, ILogger 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); } } \ No newline at end of file diff --git a/src/server/Elsa.Server.Orleans/Grains/WorkflowInstanceGrain.cs b/src/server/Elsa.Server.Orleans/Grains/WorkflowInstanceGrain.cs index 778448c0f..9744cfe3d 100644 --- a/src/server/Elsa.Server.Orleans/Grains/WorkflowInstanceGrain.cs +++ b/src/server/Elsa.Server.Orleans/Grains/WorkflowInstanceGrain.cs @@ -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 _logger; - - public WorkflowInstanceGrain(IWorkflowInstanceStore store, IResumesWorkflow workflowRunner, ILogger 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); } } \ No newline at end of file