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

Prevent WorkflowInbox from triggering new workflows

Browse source 
when targeting a specific workflow instance
This commit is contained in:
Sipke Schoorstra 2023-08-12 21:08:14 +02:00
parent cf15360c0d
commit 6b68bd8f6d
4 changed files with 36 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs
View file

@ -146,7 +146,7 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default)
public async Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default)
{
var request = new ResumeWorkflowRequest
{

2
src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs
View file

@ -47,7 +47,7 @@ public interface IWorkflowRuntime
/// <param name="workflowInstanceId">The ID of the workflow instance to resume.</param>
/// <param name="options"></param>
/// <param name="cancellationToken"></param>
Task<WorkflowExecutionResult> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default);
Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default);
/// <summary>
/// Resumes all workflows that are bookmarked on the specified activity type.

50
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs
View file

@ -25,11 +25,11 @@ public class DefaultWorkflowInbox : IWorkflowInbox
/// Initializes a new instance of the <see cref="DefaultWorkflowInbox"/> class.
/// </summary>
public DefaultWorkflowInbox(
IWorkflowRuntime workflowRuntime,
IWorkflowInboxStore store,
INotificationSender notificationSender,
ISystemClock systemClock,
IIdentityGenerator identityGenerator,
IWorkflowRuntime workflowRuntime,
IWorkflowInboxStore store,
INotificationSender notificationSender,
ISystemClock systemClock,
IIdentityGenerator identityGenerator,
IBookmarkHasher bookmarkHasher)
{
_workflowRuntime = workflowRuntime;
@ -51,7 +51,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox
public async ValueTask<SubmitWorkflowInboxMessageResult> SubmitAsync(NewWorkflowInboxMessage newMessage, WorkflowInboxMessageDeliveryOptions options, CancellationToken cancellationToken = default)
{
var now = _systemClock.UtcNow;
// Create a new message.
var message = new WorkflowInboxMessage
{
@ -66,22 +66,38 @@ public class DefaultWorkflowInbox : IWorkflowInbox
Input = newMessage.Input,
Hash = _bookmarkHasher.Hash(newMessage.ActivityTypeName, newMessage.BookmarkPayload),
};
// Store the message.
await _store.SaveAsync(message, cancellationToken);
// Send a notification.
var strategy = options.EventPublishingStrategy;
var workflowExecutionResults = new List<WorkflowExecutionResult>();
var notification = new WorkflowInboxMessageReceived(message, workflowExecutionResults);
await _notificationSender.SendAsync(notification, strategy, cancellationToken);
// Return the result.
return new SubmitWorkflowInboxMessageResult(message, workflowExecutionResults);
}
/// <inheritdoc />
public async ValueTask<DeliverWorkflowInboxMessageResult> DeliverAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default)
{
var triggeredWorkflows = await TriggerWorkflowsAsync(message, cancellationToken);
message.IsHandled = triggeredWorkflows.Any();
if (message.IsHandled)
{
message.AffectedWorkflowInstancesIds = triggeredWorkflows.Select(x => x.WorkflowInstanceId).ToList();
message.HandledAt = _systemClock.UtcNow;
await _store.SaveAsync(message, cancellationToken);
}
return new DeliverWorkflowInboxMessageResult(triggeredWorkflows);
}
private async Task<ICollection<WorkflowExecutionResult>> TriggerWorkflowsAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default)
{
var activityTypeName = message.ActivityTypeName;
var correlationId = message.CorrelationId;
@ -90,18 +106,12 @@ public class DefaultWorkflowInbox : IWorkflowInbox
var bookmarkPayload = message.BookmarkPayload;
var input = message.Input;
var options = new TriggerWorkflowsRuntimeOptions(correlationId, workflowInstanceId, activityInstanceId, input);
var result = await _workflowRuntime.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options, cancellationToken);
message.IsHandled = result.TriggeredWorkflows.Any();
if (message.IsHandled)
{
message.AffectedWorkflowInstancesIds = result.TriggeredWorkflows.Select(x => x.WorkflowInstanceId).ToList();
message.HandledAt = _systemClock.UtcNow;
await _store.SaveAsync(message, cancellationToken);
}
return new DeliverWorkflowInboxMessageResult(result.TriggeredWorkflows);
if (workflowInstanceId != null)
return await _workflowRuntime.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options, cancellationToken);
var result = await _workflowRuntime.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options, cancellationToken);
return result.TriggeredWorkflows;
}
/// <inheritdoc />

7
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs
View file

@ -119,14 +119,14 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default)
public async Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default)
{
await using (await _distributedLockProvider.AcquireLockAsync(workflowInstanceId, TimeSpan.FromMinutes(1), cancellationToken))
{
var workflowState = await _workflowStateStore.FindAsync(workflowInstanceId, cancellationToken);
if (workflowState == null)
throw new Exception($"Workflow instance {workflowInstanceId} not found");
return null;
var definitionId = workflowState.DefinitionId;
var version = workflowState.DefinitionVersion;
@ -309,7 +309,8 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
},
cancellationToken);
resumedWorkflows.Add(new WorkflowExecutionResult(workflowInstanceId, resumeResult.Status, resumeResult.SubStatus, resumeResult.Bookmarks));
if (resumeResult != null)
resumedWorkflows.Add(new WorkflowExecutionResult(workflowInstanceId, resumeResult.Status, resumeResult.SubStatus, resumeResult.Bookmarks));
}
return resumedWorkflows;