From 652a617a3b49fef4be63347229cb346b89e84cd8 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Sat, 6 Jan 2024 12:04:51 +0100 Subject: [PATCH] Add option for synchronous broadcast in WorkflowInbox This update introduces a new option to control synchronicity when broadcasting messages in the WorkflowInbox. The 'BroadcastWorkflowInboxMessageOptions' class allows the developer to specify whether the broadcasting will occur synchronously or asynchronously. The update also includes a new Endpoint and Workflow for demonstration and testing of this functionality. --- src/bundles/Elsa.Server.Web/MyEndpoint.cs | 27 ++++++++ .../Elsa.Server.Web/MyEventWorkflow.cs | 39 +++++++++++ .../Contracts/IWorkflowInbox.cs | 5 +- .../Handlers/ReadWorkflowInboxMessage.cs | 11 ++- .../BroadcastWorkflowInboxMessageOptions.cs | 15 ++++ .../WorkflowInboxMessageReceived.cs | 7 +- .../WorkflowInboxMessageDeliveryOptions.cs | 8 +-- .../DeliverWorkflowInboxMessageResult.cs | 2 +- .../Services/DefaultWorkflowInbox.cs | 69 ++++++++++++++----- .../Services/EventPublisher.cs | 8 +-- 10 files changed, 159 insertions(+), 32 deletions(-) create mode 100644 src/bundles/Elsa.Server.Web/MyEndpoint.cs create mode 100644 src/bundles/Elsa.Server.Web/MyEventWorkflow.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Models/BroadcastWorkflowInboxMessageOptions.cs diff --git a/src/bundles/Elsa.Server.Web/MyEndpoint.cs b/src/bundles/Elsa.Server.Web/MyEndpoint.cs new file mode 100644 index 000000000..12ff777c2 --- /dev/null +++ b/src/bundles/Elsa.Server.Web/MyEndpoint.cs @@ -0,0 +1,27 @@ +using Elsa.Abstractions; +using Elsa.Workflows.Runtime.Contracts; + +namespace Elsa.Server.Web; + +public class MyEndpoint : ElsaEndpointWithoutRequest +{ + private readonly IEventPublisher _eventPublisher; + + public MyEndpoint(IEventPublisher eventPublisher) + { + _eventPublisher = eventPublisher; + } + + public override void Configure() + { + Get("/my-event-workflow"); + AllowAnonymous(); + } + + public override async Task HandleAsync(CancellationToken ct) + { + Console.WriteLine("Publishing MyEvent"); + var results = await _eventPublisher.PublishAsync("MyEvent", cancellationToken: ct); + Console.WriteLine($"Affected workflows: {results.Count}"); + } +} \ No newline at end of file diff --git a/src/bundles/Elsa.Server.Web/MyEventWorkflow.cs b/src/bundles/Elsa.Server.Web/MyEventWorkflow.cs new file mode 100644 index 000000000..2439e537c --- /dev/null +++ b/src/bundles/Elsa.Server.Web/MyEventWorkflow.cs @@ -0,0 +1,39 @@ +using Elsa.Workflows; +using Elsa.Workflows.Activities; +using Elsa.Workflows.Contracts; +using Elsa.Workflows.Runtime.Activities; + +namespace Elsa.Server.Web; + +public class OnMyEventWorkflow : WorkflowBase +{ + protected override void Build(IWorkflowBuilder builder) + { + builder.Version = 1; + builder.Id = "OnMyEventWorkflow"; + builder.Root = new Sequence + { + Activities = + { + new Event("MyEvent") + { + CanStartWorkflow = true + }, + new Inline(async () => + { + // IEventPublisher.PublishAsync returns before this executes + await SomeCallAsync(); + }), + new WriteLine("End of workflow"), + new Finish() + } + }; + } + + private async Task SomeCallAsync() + { + Console.WriteLine("Hello from OnMyEventWorkflow"); + await Task.Delay(1000); + Console.WriteLine("Goodbye from OnMyEventWorkflow"); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowInbox.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowInbox.cs index 949c7cf62..52245a1fe 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowInbox.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowInbox.cs @@ -32,13 +32,14 @@ public interface IWorkflowInbox /// The message to deliver. /// An optional cancellation token. ValueTask DeliverAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default); - + /// /// Broadcasts the specified message, which may trigger new workflows and resume existing ones. /// /// The message to broadcast. + /// An optional set of delivery options. /// An optional cancellation token. - ValueTask BroadcastAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default); + ValueTask BroadcastAsync(WorkflowInboxMessage message, BroadcastWorkflowInboxMessageOptions? options, CancellationToken cancellationToken = default); /// /// Finds all messages matching the specified filter. diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ReadWorkflowInboxMessage.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ReadWorkflowInboxMessage.cs index 59e115967..ecf6cb080 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ReadWorkflowInboxMessage.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ReadWorkflowInboxMessage.cs @@ -1,5 +1,7 @@ +using Elsa.Extensions; using Elsa.Mediator.Contracts; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Notifications; namespace Elsa.Workflows.Runtime.Handlers; @@ -19,11 +21,16 @@ public class ReadWorkflowInboxMessage : INotificationHandler public async Task HandleAsync(WorkflowInboxMessageReceived notification, CancellationToken cancellationToken) { var message = notification.InboxMessage; - await _workflowInbox.BroadcastAsync(message, cancellationToken); + var options = new BroadcastWorkflowInboxMessageOptions + { + DispatchAsynchronously = notification.Options.DispatchAsynchronously + }; + var result = await _workflowInbox.BroadcastAsync(message, options, cancellationToken); + notification.WorkflowExecutionResults.AddRange(result.WorkflowExecutionResults); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Models/BroadcastWorkflowInboxMessageOptions.cs b/src/modules/Elsa.Workflows.Runtime/Models/BroadcastWorkflowInboxMessageOptions.cs new file mode 100644 index 000000000..e43cb2e98 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Models/BroadcastWorkflowInboxMessageOptions.cs @@ -0,0 +1,15 @@ +namespace Elsa.Workflows.Runtime.Models; + +/// +/// Represents the options for broadcasting a workflow inbox message. +/// +public class BroadcastWorkflowInboxMessageOptions +{ + /// + /// Gets or sets a value indicating whether the dispatch should be executed asynchronously. + /// + /// + /// true if the dispatch should be executed asynchronously; otherwise, false. + /// + public bool DispatchAsynchronously { get; set; } = true; +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowInboxMessageReceived.cs b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowInboxMessageReceived.cs index 075a960e2..823fc7cd4 100644 --- a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowInboxMessageReceived.cs +++ b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowInboxMessageReceived.cs @@ -1,5 +1,7 @@ using Elsa.Mediator.Contracts; using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.Runtime.Results; namespace Elsa.Workflows.Runtime.Notifications; @@ -7,4 +9,7 @@ namespace Elsa.Workflows.Runtime.Notifications; /// A notification that is sent when a workflow inbox message is received. /// /// The inbox message that was received. -public record WorkflowInboxMessageReceived(WorkflowInboxMessage InboxMessage) : INotification; \ No newline at end of file +public record WorkflowInboxMessageReceived( + WorkflowInboxMessage InboxMessage, + WorkflowInboxMessageDeliveryOptions Options, + ICollection WorkflowExecutionResults) : INotification; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Options/WorkflowInboxMessageDeliveryOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowInboxMessageDeliveryOptions.cs index 815de54dc..4b874e210 100644 --- a/src/modules/Elsa.Workflows.Runtime/Options/WorkflowInboxMessageDeliveryOptions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowInboxMessageDeliveryOptions.cs @@ -1,7 +1,3 @@ -using Elsa.Mediator; -using Elsa.Mediator.Contracts; -using Elsa.Workflows.Runtime.Notifications; - namespace Elsa.Workflows.Runtime.Options; /// @@ -10,7 +6,7 @@ namespace Elsa.Workflows.Runtime.Options; public class WorkflowInboxMessageDeliveryOptions { /// - /// The strategy to use when publishing the notification. + /// Whether to dispatch the message to the workflow dispatcher or send immediately. /// - public IEventPublishingStrategy EventPublishingStrategy { get; set; } = NotificationStrategy.Background; + public bool DispatchAsynchronously { get; set; } = true; } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Results/DeliverWorkflowInboxMessageResult.cs b/src/modules/Elsa.Workflows.Runtime/Results/DeliverWorkflowInboxMessageResult.cs index a05749e6a..5a786261f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Results/DeliverWorkflowInboxMessageResult.cs +++ b/src/modules/Elsa.Workflows.Runtime/Results/DeliverWorkflowInboxMessageResult.cs @@ -3,4 +3,4 @@ namespace Elsa.Workflows.Runtime.Results; /// /// Result of delivering a workflow inbox message. /// -public record DeliverWorkflowInboxMessageResult; \ No newline at end of file +public record DeliverWorkflowInboxMessageResult(ICollection WorkflowExecutionResults); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs index 7f2fb4dda..1b9259f69 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs @@ -1,4 +1,5 @@ using Elsa.Common.Contracts; +using Elsa.Mediator; using Elsa.Mediator.Contracts; using Elsa.Workflows.Contracts; using Elsa.Workflows.Runtime.Contracts; @@ -16,6 +17,7 @@ namespace Elsa.Workflows.Runtime.Services; public class DefaultWorkflowInbox : IWorkflowInbox { private readonly IWorkflowDispatcher _workflowDispatcher; + private readonly IWorkflowRuntime _workflowRuntime; private readonly IWorkflowInboxMessageStore _messageStore; private readonly INotificationSender _notificationSender; private readonly ISystemClock _systemClock; @@ -27,6 +29,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox /// public DefaultWorkflowInbox( IWorkflowDispatcher workflowDispatcher, + IWorkflowRuntime workflowRuntime, IWorkflowInboxMessageStore messageStore, INotificationSender notificationSender, ISystemClock systemClock, @@ -34,6 +37,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox IBookmarkHasher bookmarkHasher) { _workflowDispatcher = workflowDispatcher; + _workflowRuntime = workflowRuntime; _messageStore = messageStore; _notificationSender = notificationSender; _systemClock = systemClock; @@ -72,10 +76,9 @@ public class DefaultWorkflowInbox : IWorkflowInbox await _messageStore.SaveAsync(message, cancellationToken); // Send a notification. - var strategy = options.EventPublishingStrategy; var workflowExecutionResults = new List(); - var notification = new WorkflowInboxMessageReceived(message); - await _notificationSender.SendAsync(notification, strategy, cancellationToken); + var notification = new WorkflowInboxMessageReceived(message, options, workflowExecutionResults); + await _notificationSender.SendAsync(notification, NotificationStrategy.Sequential, cancellationToken); // Return the result. return new SubmitWorkflowInboxMessageResult(message, workflowExecutionResults); @@ -84,19 +87,12 @@ public class DefaultWorkflowInbox : IWorkflowInbox /// public async ValueTask DeliverAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) { - await ResumeWorkflowsAsync(message, cancellationToken); - return new DeliverWorkflowInboxMessageResult(); + await ResumeWorkflowsAsynchronouslyAsync(message, cancellationToken); + return new DeliverWorkflowInboxMessageResult(new List()); } /// - public async ValueTask BroadcastAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) - { - await TriggerWorkflowsAsync(message, cancellationToken); - - return new DeliverWorkflowInboxMessageResult(); - } - - private async Task TriggerWorkflowsAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) + public async ValueTask BroadcastAsync(WorkflowInboxMessage message, BroadcastWorkflowInboxMessageOptions? options, CancellationToken cancellationToken = default) { var activityTypeName = message.ActivityTypeName; var correlationId = message.CorrelationId; @@ -107,8 +103,28 @@ public class DefaultWorkflowInbox : IWorkflowInbox if (workflowInstanceId != null) { - await ResumeWorkflowsAsync(message, cancellationToken); - return; + if (options?.DispatchAsynchronously == true) + { + await ResumeWorkflowsAsynchronouslyAsync(message, cancellationToken); + return new DeliverWorkflowInboxMessageResult(new List()); + } + + var results = await ResumeWorkflowsSynchronouslyAsync(message, cancellationToken); + return new DeliverWorkflowInboxMessageResult(results.ToList()); + } + + if (options?.DispatchAsynchronously == false) + { + var results = await _workflowRuntime.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, new TriggerWorkflowsOptions + { + CorrelationId = correlationId, + WorkflowInstanceId = workflowInstanceId, + ActivityInstanceId = activityInstanceId, + Input = input, + CancellationTokens = cancellationToken + }); + + return new DeliverWorkflowInboxMessageResult(results.TriggeredWorkflows); } await _workflowDispatcher.DispatchAsync(new DispatchTriggerWorkflowsRequest(activityTypeName, bookmarkPayload) @@ -118,9 +134,11 @@ public class DefaultWorkflowInbox : IWorkflowInbox ActivityInstanceId = activityInstanceId, Input = input }, cancellationToken); + + return new DeliverWorkflowInboxMessageResult(new List()); } - private async Task ResumeWorkflowsAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) + private async Task ResumeWorkflowsAsynchronouslyAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) { var activityTypeName = message.ActivityTypeName; var correlationId = message.CorrelationId; @@ -138,6 +156,25 @@ public class DefaultWorkflowInbox : IWorkflowInbox }, cancellationToken); } + private async Task> ResumeWorkflowsSynchronouslyAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) + { + var activityTypeName = message.ActivityTypeName; + var correlationId = message.CorrelationId; + var workflowInstanceId = message.WorkflowInstanceId; + var activityInstanceId = message.ActivityInstanceId; + var bookmarkPayload = message.BookmarkPayload; + var input = message.Input; + + return await _workflowRuntime.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, new TriggerWorkflowsOptions + { + CorrelationId = correlationId, + WorkflowInstanceId = workflowInstanceId, + ActivityInstanceId = activityInstanceId, + Input = input, + CancellationTokens = cancellationToken + }); + } + /// public async ValueTask> FindManyAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default) { diff --git a/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs b/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs index ff6d59e0c..1acd1159f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs @@ -31,7 +31,7 @@ public class EventPublisher : IEventPublisher IDictionary? input = default, CancellationToken cancellationToken = default) { - return await PublishInternalAsync(eventName, NotificationStrategy.Sequential, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken); + return await PublishInternalAsync(eventName, false, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken); } /// @@ -43,12 +43,12 @@ public class EventPublisher : IEventPublisher IDictionary? input = default, CancellationToken cancellationToken = default) { - await PublishInternalAsync(eventName, NotificationStrategy.FireAndForget, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken); + await PublishInternalAsync(eventName, true, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken); } private async Task> PublishInternalAsync( string eventName, - IEventPublishingStrategy publishingStrategy, + bool dispatchAsynchronously, string? correlationId = default, string? workflowInstanceId = default, string? activityInstanceId = default, @@ -59,7 +59,7 @@ public class EventPublisher : IEventPublisher var message = NewWorkflowInboxMessage.For(eventBookmark, workflowInstanceId, correlationId, activityInstanceId, input); var options = new WorkflowInboxMessageDeliveryOptions { - EventPublishingStrategy = publishingStrategy, + DispatchAsynchronously = dispatchAsynchronously }; var result = await _workflowInbox.SubmitAsync(message, options, cancellationToken);