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

Update Azure Service Bus workers to use new dispatcher APIs

Browse source
This commit is contained in:
Sipke Schoorstra 2021-03-30 09:58:00 +02:00
parent bd87f5b2df
commit f91caf14bc
13 changed files with 168 additions and 607 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

97
src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs
View file

@ -1,110 +1,27 @@
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.AzureServiceBus.Bookmarks;
using Elsa.Activities.AzureServiceBus.Models;
using Elsa.Activities.AzureServiceBus.Bookmarks;
using Elsa.Activities.AzureServiceBus.Options;
using Elsa.Bookmarks;
using Elsa.Dispatch;
using Elsa.DistributedLock;
using Elsa.DistributedLocking;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
using Elsa.Services;
using Elsa.Services.Models;
using Elsa.Triggers;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.ServiceBus.Core;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Open.Linq.AsyncExtensions;
namespace Elsa.Activities.AzureServiceBus.Services
{
public class QueueWorker : IAsyncDisposable
public class QueueWorker : WorkerBase
{
// TODO: Figure out how to start jobs across multiple tenants / how to get a list of all tenants.
private const string TenantId = default;
private readonly IMessageReceiver _messageReceiver;
private readonly ICorrelatingWorkflowDispatcher _workflowDispatcher;
private readonly ILogger _logger;
public QueueWorker(
IMessageReceiver messageReceiver,
IReceiverClient messageReceiver,
ICorrelatingWorkflowDispatcher workflowDispatcher,
IServiceScopeFactory serviceScopeFactory,
IOptions<AzureServiceBusOptions> options,
ILogger<QueueWorker> logger)
ILogger<QueueWorker> logger) : base(messageReceiver, workflowDispatcher, options, logger)
{
_messageReceiver = messageReceiver;
_workflowDispatcher = workflowDispatcher;
_logger = logger;
_messageReceiver.RegisterMessageHandler(OnMessageReceived, new MessageHandlerOptions(ExceptionReceivedHandler)
{
AutoComplete = false,
MaxConcurrentCalls = options.Value.MaxConcurrentCalls
});
}
public async ValueTask DisposeAsync() => await _messageReceiver.CloseAsync();
protected override string ActivityType => nameof(AzureServiceBusQueueMessageReceived);
private async Task OnMessageReceived(Message message, CancellationToken cancellationToken)
{
_logger.LogDebug("Message received with ID {MessageId}", message.MessageId);
await TriggerWorkflowsAsync(message, cancellationToken);
await _messageReceiver.CompleteAsync(message.SystemProperties.LockToken);
}
private async Task TriggerWorkflowsAsync(Message message, CancellationToken cancellationToken)
{
var queueName = _messageReceiver.Path;
var correlationId = message.CorrelationId;
var model = new MessageModel
{
Body = message.Body,
CorrelationId = message.CorrelationId,
ContentType = message.ContentType,
Label = message.Label,
To = message.To,
MessageId = message.MessageId,
PartitionKey = message.PartitionKey,
ViaPartitionKey = message.ViaPartitionKey,
ReplyTo = message.ReplyTo,
SessionId = message.SessionId,
ExpiresAtUtc = message.ExpiresAtUtc,
TimeToLive = message.TimeToLive,
ReplyToSessionId = message.ReplyToSessionId,
ScheduledEnqueueTimeUtc = message.ScheduledEnqueueTimeUtc
};
var bookmark = new QueueMessageReceivedBookmark(queueName, correlationId);
var trigger = new QueueMessageReceivedBookmark(queueName);
var activityType = nameof(AzureServiceBusQueueMessageReceived);
await _workflowDispatcher.DispatchAsync(new ExecuteCorrelatedWorkflowRequest(correlationId, bookmark, trigger, activityType, model, TenantId: TenantId), cancellationToken);
}
private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e)
{
switch (e.Exception)
{
case MessageLockLostException:
_logger.LogDebug(e.Exception, "Message lock lost");
break;
case ServiceBusCommunicationException:
_logger.LogDebug(e.Exception, "Lost service bus communication");
break;
default:
_logger.LogError(e.Exception, "Unhandled exception");
break;
}
return Task.CompletedTask;
}
protected override IBookmark CreateBookmark(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path, message.CorrelationId);
protected override IBookmark CreateTrigger(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path);
}
}

156
src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs
View file

@ -1,167 +1,43 @@
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.AzureServiceBus.Bookmarks;
using Elsa.Activities.AzureServiceBus.Models;
using Elsa.Activities.AzureServiceBus.Options;
using Elsa.Bookmarks;
using Elsa.DistributedLock;
using Elsa.DistributedLocking;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
using Elsa.Services;
using Elsa.Triggers;
using Elsa.Dispatch;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.ServiceBus.Core;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Open.Linq.AsyncExtensions;
namespace Elsa.Activities.AzureServiceBus.Services
{
public class TopicWorker : IAsyncDisposable
public class TopicWorker : WorkerBase
{
// TODO: Figure out how to start jobs across multiple tenants / how to get a list of all tenants.
private const string TenantId = default;
private readonly IReceiverClient _messageReceiver;
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly IDistributedLockProvider _distributedLockProvider;
private readonly ILogger _logger;
public TopicWorker(
IReceiverClient messageReceiver,
IServiceScopeFactory serviceScopeFactory,
IDistributedLockProvider distributedLockProvider,
IReceiverClient receiverClient,
ICorrelatingWorkflowDispatcher correlatingWorkflowDispatcher,
IOptions<AzureServiceBusOptions> options,
ILogger<TopicWorker> logger)
ILogger<TopicWorker> logger) : base(receiverClient, correlatingWorkflowDispatcher, options, logger)
{
_messageReceiver = messageReceiver;
_serviceScopeFactory = serviceScopeFactory;
_distributedLockProvider = distributedLockProvider;
_logger = logger;
_messageReceiver.RegisterMessageHandler(OnMessageReceived, new MessageHandlerOptions(ExceptionReceivedHandler)
{
AutoComplete = false,
MaxConcurrentCalls = options.Value.MaxConcurrentCalls
});
}
public async ValueTask DisposeAsync() => await _messageReceiver.CloseAsync();
protected override string ActivityType => nameof(AzureServiceBusTopicMessageReceived);
private async Task OnMessageReceived(Message message, CancellationToken cancellationToken)
protected override IBookmark CreateBookmark(Message message)
{
_logger.LogDebug("Message received with ID {MessageId}", message.MessageId);
await TriggerWorkflowsAsync(message, cancellationToken);
await _messageReceiver.CompleteAsync(message.SystemProperties.LockToken);
GetTopicAndSubscription(out var topicName, out var subscriptionName);
return new TopicMessageReceivedBookmark(topicName, subscriptionName, message.CorrelationId);
}
private async Task TriggerWorkflowsAsync(Message message, CancellationToken cancellationToken)
protected override IBookmark CreateTrigger(Message message)
{
using var scope = _serviceScopeFactory.CreateScope();
var workflowQueue = scope.ServiceProvider.GetRequiredService<IWorkflowQueue>();
var topicName = _messageReceiver.Path.Split('/')[0];
var subscriptionName = _messageReceiver.Path.Split('/')[2];
var correlationId = message.CorrelationId;
var triggerFinder = scope.ServiceProvider.GetRequiredService<ITriggerFinder>();
var model = new MessageModel
{
Body = message.Body,
CorrelationId = message.CorrelationId,
ContentType = message.ContentType,
Label = message.Label,
To = message.To,
MessageId = message.MessageId,
PartitionKey = message.PartitionKey,
ViaPartitionKey = message.ViaPartitionKey,
ReplyTo = message.ReplyTo,
SessionId = message.SessionId,
ExpiresAtUtc = message.ExpiresAtUtc,
TimeToLive = message.TimeToLive,
ReplyToSessionId = message.ReplyToSessionId,
ScheduledEnqueueTimeUtc = message.ScheduledEnqueueTimeUtc
};
async Task TriggerNewWorkflowAsync()
{
var bookmark = new TopicMessageReceivedBookmark(topicName, subscriptionName);
var triggers = await triggerFinder.FindTriggersAsync<AzureServiceBusTopicMessageReceived>(bookmark, TenantId, cancellationToken);
foreach (var trigger in triggers)
{
var workflowBlueprint = trigger.WorkflowBlueprint;
await workflowQueue.EnqueueWorkflowDefinition(workflowBlueprint.Id, workflowBlueprint.TenantId, trigger.ActivityId, model, correlationId, null, cancellationToken);
}
}
if (string.IsNullOrWhiteSpace(correlationId))
{
await TriggerNewWorkflowAsync();
return;
}
var lockKey = $"azure-service-bus:{topicName}:{subscriptionName}:correlation-{correlationId}";
var stopwatch = new Stopwatch();
_logger.LogDebug("Acquiring lock {LockKey}", lockKey);
stopwatch.Start();
if (!await _distributedLockProvider.AcquireLockAsync(lockKey, cancellationToken))
{
_logger.LogDebug("Lock {LockKey} already taken", lockKey);
return;
}
try
{
var bookmarkFinder = scope.ServiceProvider.GetRequiredService<IBookmarkFinder>();
var workflowInstanceStore = scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
var correlatedWorkflowInstanceCount = await workflowInstanceStore.CountAsync(new CorrelationIdSpecification<WorkflowInstance>(model.CorrelationId), cancellationToken);
if (correlatedWorkflowInstanceCount > 0)
{
// Trigger existing workflows (if blocked on this message).
_logger.LogDebug("{WorkflowInstanceCount} existing workflows found with correlation ID '{CorrelationId}'. Resuming them", correlatedWorkflowInstanceCount, correlationId);
var bookmark = new TopicMessageReceivedBookmark(topicName, subscriptionName, correlationId);
var existingWorkflows = await bookmarkFinder.FindBookmarksAsync<AzureServiceBusTopicMessageReceived>(bookmark, TenantId, cancellationToken).ToList();
await workflowQueue.EnqueueWorkflowsAsync(existingWorkflows, model, model.CorrelationId, cancellationToken: cancellationToken);
}
else
{
// Trigger new workflow.
_logger.LogDebug("No existing workflows found with correlation ID '{CorrelationId}'. Starting new workflow", correlationId);
await TriggerNewWorkflowAsync();
}
}
finally
{
await _distributedLockProvider.ReleaseLockAsync(lockKey, cancellationToken);
stopwatch.Stop();
_logger.LogDebug("Lock held for {ElapseTime}", stopwatch.Elapsed);
}
GetTopicAndSubscription(out var topicName, out var subscriptionName);
return new TopicMessageReceivedBookmark(topicName, subscriptionName);
}
private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e)
private void GetTopicAndSubscription(out string topicName, out string subscriptionName)
{
switch (e.Exception)
{
case MessageLockLostException:
_logger.LogDebug(e.Exception, "Message lock lost");
break;
case ServiceBusCommunicationException:
_logger.LogDebug(e.Exception, "Lost service bus communication");
break;
default:
_logger.LogError(e.Exception, "Unhandled exception");
break;
}
return Task.CompletedTask;
var segments = ReceiverClient.Path.Split('/');
topicName = segments[0];
subscriptionName = segments[2];
}
}
}

101
src/activities/Elsa.Activities.AzureServiceBus/Services/WorkerBase.cs Normal file
View file

@ -0,0 +1,101 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Activities.AzureServiceBus.Models;
using Elsa.Activities.AzureServiceBus.Options;
using Elsa.Bookmarks;
using Elsa.Dispatch;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.ServiceBus.Core;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace Elsa.Activities.AzureServiceBus.Services
{
public abstract class WorkerBase : IAsyncDisposable
{
// TODO: Design multi-tenancy.
private const string TenantId = default;
private readonly ICorrelatingWorkflowDispatcher _workflowDispatcher;
private readonly ILogger _logger;
protected WorkerBase(
IReceiverClient receiverClient,
ICorrelatingWorkflowDispatcher workflowDispatcher,
IOptions<AzureServiceBusOptions> options,
ILogger logger)
{
ReceiverClient = receiverClient;
_workflowDispatcher = workflowDispatcher;
_logger = logger;
ReceiverClient.RegisterMessageHandler(OnMessageReceived, new MessageHandlerOptions(ExceptionReceivedHandler)
{
AutoComplete = false,
MaxConcurrentCalls = options.Value.MaxConcurrentCalls
});
}
protected IReceiverClient ReceiverClient { get; }
protected abstract string ActivityType { get; }
public async ValueTask DisposeAsync() => await ReceiverClient.CloseAsync();
protected abstract IBookmark CreateBookmark(Message message);
protected abstract IBookmark CreateTrigger(Message message);
private async Task TriggerWorkflowsAsync(Message message, CancellationToken cancellationToken)
{
var correlationId = message.CorrelationId;
var model = new MessageModel
{
Body = message.Body,
CorrelationId = message.CorrelationId,
ContentType = message.ContentType,
Label = message.Label,
To = message.To,
MessageId = message.MessageId,
PartitionKey = message.PartitionKey,
ViaPartitionKey = message.ViaPartitionKey,
ReplyTo = message.ReplyTo,
SessionId = message.SessionId,
ExpiresAtUtc = message.ExpiresAtUtc,
TimeToLive = message.TimeToLive,
ReplyToSessionId = message.ReplyToSessionId,
ScheduledEnqueueTimeUtc = message.ScheduledEnqueueTimeUtc
};
var queueName = ReceiverClient.Path;
var bookmark = CreateBookmark(message);
var trigger = CreateTrigger(message);
await _workflowDispatcher.DispatchAsync(new ExecuteCorrelatedWorkflowRequest(correlationId, bookmark, trigger, ActivityType, model, TenantId: TenantId), cancellationToken);
}
private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e)
{
switch (e.Exception)
{
case MessageLockLostException:
_logger.LogDebug(e.Exception, "Message lock lost");
break;
case ServiceBusCommunicationException:
_logger.LogDebug(e.Exception, "Lost service bus communication");
break;
default:
_logger.LogError(e.Exception, "Unhandled exception");
break;
}
return Task.CompletedTask;
}
private async Task OnMessageReceived(Message message, CancellationToken cancellationToken)
{
_logger.LogDebug("Message received with ID {MessageId}", message.MessageId);
await TriggerWorkflowsAsync(message, cancellationToken);
await ReceiverClient.CompleteAsync(message.SystemProperties.LockToken);
}
}
}

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

@ -17,7 +17,7 @@ namespace Elsa.Activities.Http.Middleware
{
public class HttpEndpointMiddleware
{
// TODO: Figure out how to start jobs across multiple tenants / how to get a list of all tenants.
// TODO: Design multi-tenancy.
private const string TenantId = default;
private readonly RequestDelegate _next;

35
src/activities/Elsa.Activities.Temporal.Hangfire/Jobs/RunHangfireWorkflowJob.cs
View file

@ -1,5 +1,6 @@
using System.Threading.Tasks;
using Elsa.Activities.Temporal.Hangfire.Models;
using Elsa.Dispatch;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
@ -10,41 +11,21 @@ namespace Elsa.Activities.Temporal.Hangfire.Jobs
{
public class RunHangfireWorkflowJob
{
private readonly IWorkflowRunner _workflowRunner;
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IWorkflowQueue _workflowQueue;
private readonly IWorkflowDefinitionDispatcher _workflowDefinitionDispatcher;
private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher;
public RunHangfireWorkflowJob(IWorkflowRunner workflowRunner, IWorkflowRegistry workflowRegistry, IWorkflowInstanceStore workflowInstanceStore, IWorkflowQueue workflowQueue)
public RunHangfireWorkflowJob(IWorkflowDefinitionDispatcher workflowDefinitionDispatcher, IWorkflowInstanceDispatcher workflowInstanceDispatcher)
{
_workflowRunner = workflowRunner;
_workflowRegistry = workflowRegistry;
_workflowInstanceStore = workflowInstanceStore;
_workflowQueue = workflowQueue;
_workflowDefinitionDispatcher = workflowDefinitionDispatcher;
_workflowInstanceDispatcher = workflowInstanceDispatcher;
}
public async Task ExecuteAsync(RunHangfireWorkflowJobModel data)
{
var workflowBlueprint = (await _workflowRegistry.GetAsync(data.WorkflowDefinitionId, data.TenantId, VersionOptions.Published));
if(workflowBlueprint == null)
return;
if (data.WorkflowInstanceId == null)
{
if (workflowBlueprint.IsSingleton == false || await GetWorkflowIsAlreadyExecutingAsync(data.TenantId, data.WorkflowDefinitionId) == false)
await _workflowRunner.RunWorkflowAsync(workflowBlueprint, data.ActivityId);
}
await _workflowDefinitionDispatcher.DispatchAsync(new ExecuteWorkflowDefinitionRequest(data.WorkflowDefinitionId!, data.ActivityId, TenantId: data.TenantId));
else
{
await _workflowQueue.EnqueueWorkflowInstance(data.WorkflowInstanceId, data.ActivityId, default);
}
}
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;
await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(data.WorkflowInstanceId, data.ActivityId));
}
}
}

48
src/core/Elsa.Abstractions/Services/IWorkflowQueue.cs
View file

@ -1,48 +0,0 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Bookmarks;
using Elsa.Builders;
namespace Elsa.Services
{
public interface IWorkflowQueue
{
/// <summary>
/// Selects workflows and workflow instances based on the specified trigger predicate and enqueues the results for execution.
/// </summary>
Task EnqueueWorkflowsAsync(
string activityType,
IBookmark bookmark,
string? tenantId,
object? input = default,
string? correlationId = default,
string? contextId = default,
CancellationToken cancellationToken = default);
/// <summary>
/// Enqueues the specified workflows for execution.
/// </summary>
Task EnqueueWorkflowsAsync(
IEnumerable<BookmarkFinderResult> results,
object? input = default,
string? correlationId = default,
string? contextId = default,
CancellationToken cancellationToken = default);
/// <summary>
/// Enqueues the specified workflow instance and activity for execution.
/// </summary>
Task EnqueueWorkflowInstance(string workflowInstanceId, string activityId, object? input, CancellationToken cancellationToken = default);
/// <summary>
/// Enqueues the specified workflow definition and activity for execution.
/// </summary>
Task EnqueueWorkflowDefinition<T>(string? tenantId, string activityId, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken) where T : IWorkflow;
/// <summary>
/// Enqueues the specified workflow definition and activity for execution.
/// </summary>
Task EnqueueWorkflowDefinition(string workflowDefinitionId, string? tenantId, string activityId, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken = default);
}
}

106
src/core/Elsa.Core/Consumers/RunWorkflowDefinitionConsumer.cs
View file

@ -1,106 +0,0 @@
using System;
using System.Threading.Tasks;
using Elsa.DistributedLock;
using Elsa.DistributedLocking;
using Elsa.Messages;
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 Rebus.Handlers;
namespace Elsa.Consumers
{
[Obsolete]
public class RunWorkflowDefinitionConsumer : IHandleMessages<RunWorkflowDefinition>
{
private readonly IWorkflowRunner _workflowRunner;
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IDistributedLockProvider _distributedLockProvider;
private readonly ILogger _logger;
private IWorkflowFactory _workflowFactory;
public RunWorkflowDefinitionConsumer(
IWorkflowRunner workflowRunner,
IWorkflowRegistry workflowRegistry,
IWorkflowInstanceStore workflowInstanceStore,
IDistributedLockProvider distributedLockProvider,
IWorkflowFactory workflowFactory,
ILogger<RunWorkflowDefinitionConsumer> logger)
{
_workflowRunner = workflowRunner;
_workflowRegistry = workflowRegistry;
_workflowInstanceStore = workflowInstanceStore;
_distributedLockProvider = distributedLockProvider;
_logger = logger;
_workflowFactory = workflowFactory;
}
public async Task Handle(RunWorkflowDefinition message)
{
var workflowDefinitionId = message.WorkflowDefinitionId;
var tenantId = message.TenantId;
var workflowBlueprint = await _workflowRegistry.GetAsync(workflowDefinitionId, tenantId, VersionOptions.Published);
if (!ValidatePreconditions(workflowDefinitionId, workflowBlueprint))
return;
var correlationId = message.CorrelationId;
if (!string.IsNullOrWhiteSpace(message.CorrelationId))
{
var lockKey = $"{nameof(RunWorkflowDefinitionConsumer)}:workflow-definition-{message.WorkflowDefinitionId}:correlation-{message.CorrelationId}";
if (!await _distributedLockProvider.AcquireLockAsync(lockKey))
{
_logger.LogDebug("Lock {LockKey} already taken", lockKey);
return;
}
try
{
var correlatedWorkflowInstanceCount = await _workflowInstanceStore.CountAsync(new WorkflowDefinitionIdSpecification(workflowDefinitionId).And(new CorrelationIdSpecification<WorkflowInstance>(correlationId)));
if (correlatedWorkflowInstanceCount > 0)
{
// Do not create a new workflow instance.
_logger.LogWarning("There's already a workflow with correlation ID '{CorrelationId}'", correlationId);
return;
}
_logger.LogDebug("No existing workflows found with correlation ID '{CorrelationId}'. Starting new workflow", correlationId);
// Persist workflow immediately before leaving the lock to prevent scenarios where the workflow takes a long time before the first time it gets persisted, while a new message comes in with the same workflow and correlation.
var workflowInstance = await _workflowFactory.InstantiateAsync(workflowBlueprint!, correlationId, message.ContextId);
await _workflowInstanceStore.SaveAsync(workflowInstance);
// Run the workflow instance.
await _workflowRunner.RunWorkflowAsync(workflowBlueprint!, workflowInstance, message.ActivityId, message.Input);
return;
}
finally
{
await _distributedLockProvider.ReleaseLockAsync(lockKey);
}
}
await _workflowRunner.RunWorkflowAsync(workflowBlueprint!, message.ActivityId, message.Input, message.CorrelationId, message.ContextId);
}
private bool ValidatePreconditions(string? workflowDefinitionId, IWorkflowBlueprint? workflowBlueprint)
{
if (workflowBlueprint == null)
{
_logger.LogError("Could not run workflow with ID {WorkflowDefinitionId} because it does not exist", workflowDefinitionId);
return false;
}
return true;
}
}
}

102
src/core/Elsa.Core/Consumers/RunWorkflowInstanceConsumer.cs
View file

@ -1,102 +0,0 @@
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;
using Elsa.DistributedLock;
using Elsa.DistributedLocking;
using Elsa.Messages;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Services;
using Microsoft.Extensions.Logging;
using Rebus.Handlers;
namespace Elsa.Consumers
{
public class RunWorkflowInstanceConsumer : IHandleMessages<RunWorkflowInstance>
{
private readonly IWorkflowRunner _workflowRunner;
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IDistributedLockProvider _distributedLockProvider;
private readonly ICommandSender _commandSender;
private readonly ILogger _logger;
private readonly Stopwatch _stopwatch = new();
public RunWorkflowInstanceConsumer(
IWorkflowRunner workflowRunner,
IWorkflowInstanceStore workflowInstanceStore,
IDistributedLockProvider distributedLockProvider,
ICommandSender commandSender,
ILogger<RunWorkflowInstanceConsumer> logger)
{
_workflowRunner = workflowRunner;
_workflowInstanceStore = workflowInstanceStore;
_distributedLockProvider = distributedLockProvider;
_commandSender = commandSender;
_logger = logger;
}
public async Task Handle(RunWorkflowInstance message)
{
var workflowInstanceId = message.WorkflowInstanceId;
var lockKey = workflowInstanceId;
_logger.LogDebug("Acquiring lock on workflow instance {WorkflowInstanceId}", workflowInstanceId);
_stopwatch.Restart();
if (!await _distributedLockProvider.AcquireLockAsync(lockKey))
{
_logger.LogDebug("Failed to acquire lock on workflow instance {WorkflowInstanceId}. Re-queueing message", workflowInstanceId);
await _commandSender.SendAsync(message);
return;
}
try
{
var workflowInstance = await _workflowInstanceStore.FindByIdAsync(message.WorkflowInstanceId);
if (!ValidatePreconditions(workflowInstanceId, workflowInstance, message.ActivityId))
return;
await _workflowRunner.RunWorkflowAsync(
workflowInstance!,
message.ActivityId,
message.Input);
}
finally
{
await _distributedLockProvider.ReleaseLockAsync(lockKey);
_stopwatch.Stop();
_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;
}
}
}

6
src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowRequestConsumer.cs → src/core/Elsa.Core/Dispatch/Consumers/ExecuteWorkflowInstanceRequestConsumer.cs
View file

@ -10,7 +10,7 @@ using Rebus.Handlers;
namespace Elsa.Dispatch.Consumers
{
public class ExecuteWorkflowRequestConsumer : IHandleMessages<ExecuteWorkflowInstanceRequest>
public class ExecuteWorkflowInstanceRequestConsumer : IHandleMessages<ExecuteWorkflowInstanceRequest>
{
private readonly IWorkflowRunner _workflowRunner;
private readonly IWorkflowInstanceStore _workflowInstanceStore;
@ -19,12 +19,12 @@ namespace Elsa.Dispatch.Consumers
private readonly ILogger _logger;
private readonly Stopwatch _stopwatch = new();
public ExecuteWorkflowRequestConsumer(
public ExecuteWorkflowInstanceRequestConsumer(
IWorkflowRunner workflowRunner,
IWorkflowInstanceStore workflowInstanceStore,
IDistributedLockProvider distributedLockProvider,
ICommandSender commandSender,
ILogger<ExecuteWorkflowRequestConsumer> logger)
ILogger<ExecuteWorkflowInstanceRequestConsumer> logger)
{
_workflowRunner = workflowRunner;
_workflowInstanceStore = workflowInstanceStore;

12
src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs
View file

@ -7,13 +7,13 @@ using Elsa.ActivityProviders;
using Elsa.ActivityTypeProviders;
using Elsa.Bookmarks;
using Elsa.Builders;
using Elsa.Consumers;
using Elsa.Decorators;
using Elsa.Dispatch;
using Elsa.Dispatch.Consumers;
using Elsa.Expressions;
using Elsa.Handlers;
using Elsa.HostedServices;
using Elsa.Mapping;
using Elsa.Messages;
using Elsa.Metadata;
using Elsa.Persistence;
using Elsa.Persistence.Decorators;
@ -59,8 +59,6 @@ namespace Microsoft.Extensions.DependencyInjection
.AddCoreActivities();
options.AddAutoMapper();
options.AddConsumer<RunWorkflowDefinitionConsumer, RunWorkflowDefinition>();
options.AddConsumer<RunWorkflowInstanceConsumer, RunWorkflowInstance>();
services.Decorate<IWorkflowDefinitionStore, InitializingWorkflowDefinitionStore>();
services.Decorate<IWorkflowDefinitionStore, EventPublishingWorkflowDefinitionStore>();
@ -160,15 +158,15 @@ namespace Microsoft.Extensions.DependencyInjection
// Service Bus.
services
.AddScoped<IWorkflowQueue, WorkflowQueue>()
.AddSingleton<ServiceBusFactory>()
.AddSingleton<IServiceBusFactory, ServiceBusFactory>()
.AddSingleton<ICommandSender, CommandSender>()
.AddSingleton<IEventPublisher, EventPublisher>();
options
.AddConsumer<RunWorkflowDefinitionConsumer, RunWorkflowDefinition>()
.AddConsumer<RunWorkflowInstanceConsumer, RunWorkflowInstance>();
.AddConsumer<ExecuteCorrelatedWorkflowRequestConsumer, ExecuteCorrelatedWorkflowRequest>()
.AddConsumer<ExecuteWorkflowDefinitionRequestConsumer, ExecuteWorkflowDefinitionRequest>()
.AddConsumer<ExecuteWorkflowInstanceRequestConsumer, ExecuteWorkflowInstanceRequest>();
// AutoMapper.
services

57
src/core/Elsa.Core/Services/WorkflowQueue.cs
View file

@ -1,57 +0,0 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Bookmarks;
using Elsa.Builders;
using Elsa.Messages;
using Open.Linq.AsyncExtensions;
namespace Elsa.Services
{
public class WorkflowQueue : IWorkflowQueue
{
private readonly IBookmarkFinder _bookmarkFinder;
private readonly ICommandSender _commandSender;
public WorkflowQueue(IBookmarkFinder bookmarkFinder, ICommandSender commandSender)
{
_bookmarkFinder = bookmarkFinder;
_commandSender = commandSender;
}
public async Task EnqueueWorkflowsAsync(
string activityType,
IBookmark bookmark,
string? tenantId,
object? input = default,
string? correlationId = default,
string? contextId = default,
CancellationToken cancellationToken = default)
{
var results = await _bookmarkFinder.FindBookmarksAsync(activityType, bookmark, tenantId, cancellationToken).ToList();
await EnqueueWorkflowsAsync(results, input, correlationId, contextId, cancellationToken);
}
public async Task EnqueueWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken)
{
foreach (var result in results)
await EnqueueWorkflowInstance(result.WorkflowInstanceId, result.ActivityId, input, cancellationToken);
}
public async Task EnqueueWorkflowInstance(string workflowInstanceId, string activityId, object? input, CancellationToken cancellationToken = default)
{
await _commandSender.SendAsync(new RunWorkflowInstance(workflowInstanceId, activityId, input));
}
public async Task EnqueueWorkflowDefinition<T>(string? tenantId, string activityId, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken) where T : IWorkflow
{
var workflowDefinitionId = typeof(T).Name;
await EnqueueWorkflowDefinition(workflowDefinitionId, tenantId, activityId, input, correlationId, contextId, cancellationToken);
}
public async Task EnqueueWorkflowDefinition(string workflowDefinitionId, string? tenantId, string activityId, object? input, string? correlationId, string? contextId, CancellationToken cancellationToken)
{
await _commandSender.SendAsync(new RunWorkflowDefinition(workflowDefinitionId, tenantId, activityId, input, correlationId, contextId));
}
}
}

34
src/core/Elsa.Core/Services/WorkflowReviver.cs
View file

@ -2,6 +2,7 @@
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Elsa.Exceptions;
using Elsa.Models;
using Elsa.Persistence;
@ -11,24 +12,25 @@ namespace Elsa.Services
public class WorkflowReviver : IWorkflowReviver
{
private readonly IWorkflowRunner _workflowRunner;
private readonly IWorkflowQueue _workflowQueue;
private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher;
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IGetsStartActivitiesForCompositeActivityBlueprint startingActivitiesProvider;
private readonly IGetsStartActivitiesForCompositeActivityBlueprint _startingActivitiesProvider;
public WorkflowReviver(IWorkflowRunner workflowRunner,
IWorkflowQueue workflowQueue,
IWorkflowRegistry workflowRegistry,
IWorkflowInstanceStore workflowInstanceStore,
IGetsStartActivitiesForCompositeActivityBlueprint startingActivitiesProvider)
public WorkflowReviver(
IWorkflowRunner workflowRunner,
IWorkflowInstanceDispatcher workflowInstanceDispatcher,
IWorkflowRegistry workflowRegistry,
IWorkflowInstanceStore workflowInstanceStore,
IGetsStartActivitiesForCompositeActivityBlueprint startingActivitiesProvider)
{
_workflowRunner = workflowRunner;
_workflowQueue = workflowQueue;
_workflowInstanceDispatcher = workflowInstanceDispatcher;
_workflowRegistry = workflowRegistry;
_workflowInstanceStore = workflowInstanceStore;
this.startingActivitiesProvider = startingActivitiesProvider ?? throw new ArgumentNullException(nameof(startingActivitiesProvider));
_startingActivitiesProvider = startingActivitiesProvider ?? throw new ArgumentNullException(nameof(startingActivitiesProvider));
}
public async Task<WorkflowInstance> ReviveAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken)
{
if (workflowInstance.WorkflowStatus != WorkflowStatus.Faulted)
@ -38,7 +40,7 @@ namespace Elsa.Services
if (fault == null)
throw new WorkflowException("Cannot revive a workflow with no fault");
var faultedActivityId = fault.FaultedActivityId;
if (faultedActivityId == null)
@ -66,12 +68,12 @@ namespace Elsa.Services
workflowInstance = await ReviveAsync(workflowInstance, cancellationToken);
return await _workflowRunner.RunWorkflowAsync(workflowInstance, null, null, cancellationToken);
}
public async Task<WorkflowInstance> ReviveAndQueueAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken)
{
workflowInstance = await ReviveAsync(workflowInstance, cancellationToken);
var currentActivity = await GetActivityToScheduleAsync(workflowInstance, cancellationToken);
await _workflowQueue.EnqueueWorkflowInstance(workflowInstance.Id, currentActivity.ActivityId, currentActivity.Input, cancellationToken);
var currentActivity = await GetActivityToScheduleAsync(workflowInstance, cancellationToken);
await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(workflowInstance.Id, currentActivity.ActivityId, currentActivity.Input), cancellationToken);
return workflowInstance;
}
@ -86,8 +88,8 @@ namespace Elsa.Services
if (workflowBlueprint == null)
throw new WorkflowException($"Could not find associated workflow definition {workflowInstance.DefinitionId} with version {workflowInstance.Version}");
var startActivity = startingActivitiesProvider.GetStartActivities(workflowBlueprint).FirstOrDefault();
var startActivity = _startingActivitiesProvider.GetStartActivities(workflowBlueprint).FirstOrDefault();
if (startActivity == null)
throw new WorkflowException($"Cannot revive workflow {workflowInstance.Id} because it has no start activities");

19
src/core/Elsa.Core/StartupTasks/ContinueRunningWorkflows.cs
View file

@ -1,6 +1,7 @@
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Elsa.DistributedLock;
using Elsa.DistributedLocking;
using Elsa.Models;
@ -19,18 +20,18 @@ namespace Elsa.StartupTasks
public class ContinueRunningWorkflows : IStartupTask
{
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IWorkflowQueue _workflowQueue;
private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher;
private readonly IDistributedLockProvider _distributedLockProvider;
private readonly ILogger _logger;
public ContinueRunningWorkflows(
IWorkflowInstanceStore workflowInstanceStore,
IWorkflowQueue workflowQueue,
IWorkflowInstanceDispatcher workflowInstanceDispatcher,
IDistributedLockProvider distributedLockProvider,
ILogger<ContinueRunningWorkflows> logger)
{
_workflowInstanceStore = workflowInstanceStore;
_workflowQueue = workflowQueue;
_workflowInstanceDispatcher = workflowInstanceDispatcher;
_distributedLockProvider = distributedLockProvider;
_logger = logger;
}
@ -48,7 +49,7 @@ namespace Elsa.StartupTasks
{
var instances = await _workflowInstanceStore.FindManyAsync(new WorkflowStatusSpecification(WorkflowStatus.Running), cancellationToken: cancellationToken).ToList();
if(instances.Any())
if (instances.Any())
_logger.LogInformation("Found {WorkflowInstanceCount} workflows with status 'Running'. Resuming each one of them", instances.Count);
else
_logger.LogInformation("Found no workflows with status 'Running'. Nothing to resume");
@ -62,7 +63,9 @@ namespace Elsa.StartupTasks
{
if (instance.BlockingActivities.Any())
{
_logger.LogWarning("Workflow '{WorkflowInstanceId}' was in the Running state, but has no scheduled activities not has a currently executing one. However, it does have blocking activities, so switching to Suspended status", instance.Id);
_logger.LogWarning(
"Workflow '{WorkflowInstanceId}' was in the Running state, but has no scheduled activities not has a currently executing one. However, it does have blocking activities, so switching to Suspended status",
instance.Id);
instance.WorkflowStatus = WorkflowStatus.Suspended;
await _workflowInstanceStore.SaveAsync(instance, cancellationToken);
continue;
@ -74,11 +77,7 @@ namespace Elsa.StartupTasks
var scheduledActivity = instance.CurrentActivity ?? instance.ScheduledActivities.Peek();
await _workflowQueue.EnqueueWorkflowInstance(
instance.Id,
scheduledActivity.ActivityId,
scheduledActivity.Input,
cancellationToken);
await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(instance.Id, scheduledActivity.ActivityId, scheduledActivity.Input), cancellationToken);
}
}
finally