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

Merge branch '3.0.1' into v3.0.1

Browse source
This commit is contained in:
Sipke Schoorstra 2024-01-06 14:07:21 +01:00
parent 7f1b5be9b9 3b48636ced
commit 91ac5e816a
36 changed files with 508 additions and 113 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
src/bundles/Elsa.Server.Web/MyEndpoint.cs Normal file
View file

@ -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}");
}
}

39
src/bundles/Elsa.Server.Web/MyEventWorkflow.cs Normal file
View file

@ -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");
}
}

2
src/bundles/Elsa.Server.Web/Program.cs
View file

@ -24,7 +24,7 @@ using Proto.Persistence.SqlServer;
const bool useMongoDb = false;
const bool useSqlServer = false;
const bool useDapper = false;
const bool useProtoActor = true;
const bool useProtoActor = false;
const bool useHangfire = false;
const bool useQuartz = true;
const bool useMassTransit = true;

29
src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs
View file

@ -52,7 +52,6 @@ public static class DependencyInjectionExtensions
{
var builderOptions = new ElsaClientBuilderOptions();
configureClient.Invoke(builderOptions);
builderOptions.ConfigureHttpClientBuilder += builder => builder.AddHttpMessageHandler(sp => (DelegatingHandler)sp.GetRequiredService(builderOptions.AuthenticationHandler));
services.AddScoped(builderOptions.AuthenticationHandler);
@ -63,7 +62,19 @@ public static class DependencyInjectionExtensions
options.ConfigureHttpClient = builderOptions.ConfigureHttpClient;
options.ApiKey = builderOptions.ApiKey;
});
var builderOptionsWithoutRetryPolicy = new ElsaClientBuilderOptions
{
ApiKey = builderOptions.ApiKey,
AuthenticationHandler = builderOptions.AuthenticationHandler,
BaseAddress = builderOptions.BaseAddress,
ConfigureHttpClient = builderOptions.ConfigureHttpClient,
ConfigureHttpClientBuilder = builderOptions.ConfigureHttpClientBuilder,
ConfigureRetryPolicy = null
};
services.AddApi<IWorkflowDefinitionsApi>(builderOptions);
services.AddApi<IExecuteWorkflowApi>(builderOptionsWithoutRetryPolicy);
services.AddApi<IWorkflowInstancesApi>(builderOptions);
services.AddApi<IActivityDescriptorsApi>(builderOptions);
services.AddApi<IActivityDescriptorOptionsApi>(builderOptions);
@ -89,8 +100,20 @@ public static class DependencyInjectionExtensions
public static void AddApi<T>(this IServiceCollection services, ElsaClientBuilderOptions? httpClientBuilderOptions = default) where T : class
{
var builder = services.AddRefitClient<T>(CreateRefitSettings, typeof(T).Name).ConfigureHttpClient(ConfigureElsaApiHttpClient);
httpClientBuilderOptions?.ConfigureHttpClientBuilder?.Invoke(builder);
builder.AddTransientHttpErrorPolicy(p => p.WaitAndRetryAsync(3, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt))));
httpClientBuilderOptions?.ConfigureHttpClientBuilder(builder);
httpClientBuilderOptions?.ConfigureRetryPolicy?.Invoke(builder);
}
/// <summary>
/// Adds a refit client for the specified API type.
/// </summary>
/// <param name="services">The service collection.</param>
/// <param name="httpClientBuilderOptions">An options object that can be used to configure the HTTP client builder.</param>
/// <typeparam name="T">The type representing the API.</typeparam>
public static void AddApiWithoutRetryPolicy<T>(this IServiceCollection services, ElsaClientBuilderOptions? httpClientBuilderOptions = default) where T : class
{
var builder = services.AddRefitClient<T>(CreateRefitSettings, typeof(T).Name).ConfigureHttpClient(ConfigureElsaApiHttpClient);
httpClientBuilderOptions?.ConfigureHttpClientBuilder(builder);
}
/// <summary>

6
src/clients/Elsa.Api.Client/Options/ElsaClientBuilderOptions.cs
View file

@ -1,5 +1,6 @@
using Elsa.Api.Client.HttpMessageHandlers;
using Microsoft.Extensions.DependencyInjection;
using Polly;
namespace Elsa.Api.Client.Options;
@ -33,4 +34,9 @@ public class ElsaClientBuilderOptions
/// Gets or sets a delegate that can be used to configure the HTTP client builder.
/// </summary>
public Action<IHttpClientBuilder> ConfigureHttpClientBuilder { get; set; } = _ => { };
/// <summary>
/// Gets or sets a delegate that can be used to configure the retry policy.
/// </summary>
public Action<IHttpClientBuilder>? ConfigureRetryPolicy { get; set; } = builder => builder.AddTransientHttpErrorPolicy(p => p.WaitAndRetryAsync(3, attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt))));
}

32
src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IExecuteWorkflowApi.cs Normal file
View file

@ -0,0 +1,32 @@
using Elsa.Api.Client.Resources.WorkflowDefinitions.Requests;
using JetBrains.Annotations;
using Refit;
namespace Elsa.Api.Client.Resources.WorkflowDefinitions.Contracts;
/// <summary>
/// Represents a client for the workflow definitions API.
/// </summary>
[PublicAPI]
public interface IExecuteWorkflowApi
{
/// <summary>
/// Executes a workflow definition.
/// </summary>
/// <param name="definitionId">The definition ID of the workflow definition to execute.</param>
/// <param name="request">An optional request containing options for executing the workflow definition.</param>
/// <param name="cancellationToken">An optional cancellation token.</param>
/// <returns>A response containing information about the workflow instance that was created.</returns>
[Post("/workflow-definitions/{definitionId}/execute")]
Task<HttpResponseMessage> ExecuteAsync(string definitionId, ExecuteWorkflowDefinitionRequest? request, CancellationToken cancellationToken = default);
/// <summary>
/// Dispatches a request to execute the specified workflow definition.
/// </summary>
/// <param name="definitionId">The definition ID of the workflow definition to dispatch request.</param>
/// <param name="request">An optional request containing options for dispatching a request to execute the specified workflow definition.</param>
/// <param name="cancellationToken">An optional cancellation token.</param>
/// <returns>A response containing information about the workflow instance that was created.</returns>
[Post("/workflow-definitions/{definitionId}/dispatch")]
Task<HttpResponseMessage> DispatchAsync(string definitionId, DispatchWorkflowDefinitionRequest? request, CancellationToken cancellationToken = default);
}

20
src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IWorkflowDefinitionsApi.cs
View file

@ -196,24 +196,4 @@ public interface IWorkflowDefinitionsApi
/// <param name="cancellationToken">An optional cancellation token.</param>
[Post("/workflow-definitions/{definitionId}/revert/{version}")]
Task RevertVersionAsync(string definitionId, int version, CancellationToken cancellationToken = default);
/// <summary>
/// Executes a workflow definition.
/// </summary>
/// <param name="definitionId">The definition ID of the workflow definition to execute.</param>
/// <param name="request">An optional request containing options for executing the workflow definition.</param>
/// <param name="cancellationToken">An optional cancellation token.</param>
/// <returns>A response containing information about the workflow instance that was created.</returns>
[Post("/workflow-definitions/{definitionId}/execute")]
Task<HttpResponseMessage> ExecuteAsync(string definitionId, ExecuteWorkflowDefinitionRequest? request, CancellationToken cancellationToken = default);
/// <summary>
/// Dispatches a request to execute the specified workflow definition.
/// </summary>
/// <param name="definitionId">The definition ID of the workflow definition to dispatch request.</param>
/// <param name="request">An optional request containing options for dispatching a request to execute the specified workflow definition.</param>
/// <param name="cancellationToken">An optional cancellation token.</param>
/// <returns>A response containing information about the workflow instance that was created.</returns>
[Post("/workflow-definitions/{definitionId}/dispatch")]
Task<HttpResponseMessage> DispatchAsync(string definitionId, DispatchWorkflowDefinitionRequest? request, CancellationToken cancellationToken = default);
}

6
src/common/Elsa.Api.Common/Elsa.Api.Common.csproj
View file

@ -19,9 +19,9 @@
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)' == 'net8.0'">
<PackageReference Include="FastEndpoints" Version="5.20.1.7-beta"/>
<PackageReference Include="FastEndpoints.Security" Version="5.20.1.7-beta"/>
<PackageReference Include="FastEndpoints.Swagger" Version="5.20.1.7-beta"/>
<PackageReference Include="FastEndpoints" Version="5.21.2"/>
<PackageReference Include="FastEndpoints.Security" Version="5.21.2"/>
<PackageReference Include="FastEndpoints.Swagger" Version="5.21.2"/>
</ItemGroup>
<ItemGroup>

2
src/modules/Elsa.Http/Activities/FlowSendHttpRequest.cs
View file

@ -43,8 +43,6 @@ public class FlowSendHttpRequest : SendHttpRequestBase, IActivityPropertyDefault
outcomes.Add(outcome);
outcomes.Add("Done");
context.JournalData["StatusCode"] = statusCode;
await context.CompleteActivityWithOutcomesAsync(outcomes.ToArray());
}

17
src/modules/Elsa.Http/Activities/SendHttpRequestBase.cs
View file

@ -77,12 +77,24 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
)]
public Input<HttpHeaders?> RequestHeaders { get; set; } = new(new HttpHeaders());
/// <summary>
/// The HTTP response status code
/// </summary>
[Output(Description = "The HTTP response status code")]
public Output<int> StatusCode { get; set; } = default!;
/// <summary>
/// The parsed content, if any.
/// </summary>
[Output(Description = "The parsed content, if any.")]
public Output<object?> ParsedContent { get; set; } = default!;
/// <summary>
/// The response headers that were received.
/// </summary>
[Output(Description = "The response headers that were received.")]
public Output<HttpHeaders?> ResponseHeaders { get; set; } = default!;
/// <inheritdoc />
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
@ -115,8 +127,13 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
{
var response = await httpClient.SendAsync(request, cancellationToken);
var parsedContent = await ParseContentAsync(context, response.Content);
var statusCode = (int)response.StatusCode;
var responseHeaders = new HttpHeaders(response.Headers);
context.Set(Result, response);
context.Set(ParsedContent, parsedContent);
context.Set(StatusCode, statusCode);
context.Set(ResponseHeaders, responseHeaders);
await HandleResponseAsync(context, response);
}

20
src/modules/Elsa.Http/Models/HttpHeaders.cs
View file

@ -1,3 +1,4 @@
using System.Net.Http.Headers;
using System.Text.Json.Serialization;
using Elsa.Extensions;
using Elsa.Http.Serialization;
@ -10,6 +11,25 @@ namespace Elsa.Http.Models;
[JsonConverter(typeof(HttpHeadersConverter))]
public class HttpHeaders : Dictionary<string, string[]>
{
/// <inheritdoc />
public HttpHeaders()
{
}
/// <inheritdoc />
public HttpHeaders(IDictionary<string, string[]> source)
{
foreach (var item in source)
Add(item.Key, item.Value);
}
/// <inheritdoc />
public HttpHeaders(HttpResponseHeaders source)
{
foreach (var item in source)
Add(item.Key, item.Value.ToArray());
}
/// <summary>
/// Gets the content type.
/// </summary>

1
src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs
View file

@ -46,6 +46,7 @@ public class MassTransitWorkflowDispatcher : IWorkflowDispatcher
ActivityInstanceId = request.ActivityInstanceId,
ActivityHash = request.ActivityHash,
Input = request.Input,
Properties = request.Properties,
CorrelationId = request.CorrelationId
}, cancellationToken);
return new();

2
src/modules/Elsa.WorkflowContexts/Middleware/WorkflowContextActivityExecutionMiddleware.cs
View file

@ -36,7 +36,7 @@ public class WorkflowContextActivityExecutionMiddleware : IActivityExecutionMidd
}
// Check if this is a background execution.
var isBackgroundExecution = context.TransientProperties.GetValueOrDefault<object, bool>(BackgroundActivityCollectorMiddleware.IsBackgroundExecution);
var isBackgroundExecution = context.GetIsBackgroundExecution();
// Is the activity configured to load the context?
foreach (var providerType in providerTypes)

23
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs
View file

@ -230,6 +230,29 @@ public class ActivityExecutionContext : IExecutionContext
/// <param name="options">The options used to schedule the activity.</param>
public async ValueTask ScheduleActivityAsync(ActivityNode? activityNode, ActivityExecutionContext? owner = default, ScheduleWorkOptions? options = default)
{
if (this.GetIsBackgroundExecution())
{
var scheduledActivity = new ScheduledActivity
{
ActivityNodeId = activityNode?.NodeId,
OwnerActivityInstanceId = owner?.Id,
Options = options != null ? new ScheduledActivityOptions
{
CompletionCallback = options?.CompletionCallback?.Method.Name,
Tag = options?.Tag,
ExistingActivityInstanceId = options?.ExistingActivityExecutionContext?.Id,
PreventDuplicateScheduling = options?.PreventDuplicateScheduling ?? false,
Variables = options?.Variables?.ToList(),
Input = options?.Input
} : default
};
var scheduledActivities = this.GetBackgroundScheduledActivities().ToList();
scheduledActivities.Add(scheduledActivity);
this.SetBackgroundScheduledActivities(scheduledActivities);
return;
}
var completionCallback = options?.CompletionCallback;
owner ??= this;

12
src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs
View file

@ -456,6 +456,16 @@ public static class ActivityExecutionContextExtensions
/// </summary>
public static async ValueTask CompleteActivityAsync(this ActivityExecutionContext context, object? result = default)
{
var outcomes = result as Outcomes;
// If the activity is executing in the background, simply capture the result and return.
if (context.GetIsBackgroundExecution())
{
if (outcomes != null)
context.SetBackgroundOutcomes(outcomes.Names);
return;
}
// If the activity is not running, do nothing.
if (context.Status != ActivityStatus.Running)
return;
@ -470,7 +480,7 @@ public static class ActivityExecutionContextExtensions
context.Status = ActivityStatus.Completed;
// Record the outcomes, if any.
if (result is Outcomes outcomes)
if (outcomes != null)
context.JournalData["Outcomes"] = outcomes.Names;
// Record the output, if any.

65
src/modules/Elsa.Workflows.Core/Extensions/BackgroundActivityExecutionContextExtensions.cs Normal file
View file

@ -0,0 +1,65 @@
using Elsa.Workflows.Models;
namespace Elsa.Workflows;
/// <summary>
/// Adds extension methods to <see cref="ActivityExecutionContext"/>.
/// </summary>
public static class BackgroundActivityExecutionContextExtensions
{
/// <summary>
/// A key into the activity execution context's transient properties that indicates whether the current activity is being executed in the background.
/// </summary>
public static readonly object IsBackgroundExecution = new();
/// <summary>
/// Configures the activity execution context to execute the current activity in the background.
/// </summary>
public static void SetIsBackgroundExecution(this ActivityExecutionContext activityExecutionContext, bool value = true)
{
activityExecutionContext.TransientProperties[IsBackgroundExecution] = value;
}
/// <summary>
/// Gets a value indicating whether the current activity is being executed in the background.
/// </summary>
public static bool GetIsBackgroundExecution(this ActivityExecutionContext activityExecutionContext)
{
return activityExecutionContext.TransientProperties.ContainsKey(IsBackgroundExecution);
}
/// <summary>
/// Sets the background outcomes.
/// </summary>
public static void SetBackgroundOutcomes(this ActivityExecutionContext activityExecutionContext, IEnumerable<string> outcomes)
{
var outcomesList = outcomes.ToList();
activityExecutionContext.SetProperty("BackgroundOutcomes", outcomesList);
}
/// <summary>
/// Gets the background outcomes.
/// </summary>
public static IEnumerable<string> GetBackgroundOutcomes(this ActivityExecutionContext activityExecutionContext)
{
return activityExecutionContext.GetProperty<IEnumerable<string>>("BackgroundOutcomes") ?? Enumerable.Empty<string>();
}
/// <summary>
/// Sets the background scheduled activities.
/// </summary>
public static void SetBackgroundScheduledActivities(this ActivityExecutionContext activityExecutionContext, IEnumerable<ScheduledActivity> scheduledActivities)
{
var scheduledActivitiesList = scheduledActivities.ToList();
activityExecutionContext.SetProperty("BackgroundScheduledActivities", scheduledActivitiesList);
}
/// <summary>
/// Gets the background scheduled activities.
/// </summary>
/// <param name="activityExecutionContext"></param>
public static IEnumerable<ScheduledActivity> GetBackgroundScheduledActivities(this ActivityExecutionContext activityExecutionContext)
{
return activityExecutionContext.GetProperty<IEnumerable<ScheduledActivity>>("BackgroundScheduledActivities") ?? Enumerable.Empty<ScheduledActivity>();
}
}

4
src/modules/Elsa.Workflows.Core/Extensions/ModuleExtensions.cs
View file

@ -16,6 +16,8 @@ public static class ModuleExtensions
public static IServiceCollection AddStorageDriver<T>(this IServiceCollection services) where T : class, IStorageDriver
{
return services.AddSingleton<IStorageDriver, T>();
return services
.AddSingleton<T>()
.AddSingleton<IStorageDriver, T>();
}
}

3
src/modules/Elsa.Workflows.Core/Models/BackgroundExecutionOutcome.cs Normal file
View file

@ -0,0 +1,3 @@
namespace Elsa.Workflows.Models;
public record BackgroundExecutionOutcome(string Name, object? Payload);

8
src/modules/Elsa.Workflows.Core/Models/BackgroundExecutionResult.cs Normal file
View file

@ -0,0 +1,8 @@
namespace Elsa.Workflows.Models;
public class BackgroundExecutionResult
{
public ICollection<BackgroundExecutionOutcome> Outcomes { get; set; } = new List<BackgroundExecutionOutcome>();
public ICollection<WorkflowExecutionLogEntry> ExecutionLog { get; set; } = new List<WorkflowExecutionLogEntry>();
public IDictionary<string, object?> JournalData { get; } = new Dictionary<string, object?>();
}

8
src/modules/Elsa.Workflows.Core/Models/ScheduledActivity.cs Normal file
View file

@ -0,0 +1,8 @@
namespace Elsa.Workflows.Models;
public class ScheduledActivity
{
public string? ActivityNodeId { get; set; }
public string? OwnerActivityInstanceId { get; set; }
public ScheduledActivityOptions? Options { get; set; }
}

13
src/modules/Elsa.Workflows.Core/Models/ScheduledActivityOptions.cs Normal file
View file

@ -0,0 +1,13 @@
using Elsa.Workflows.Memory;
namespace Elsa.Workflows.Models;
public class ScheduledActivityOptions
{
public string? CompletionCallback { get; set; }
public object? Tag { get; set; }
public ICollection<Variable>? Variables { get; set; }
public string? ExistingActivityInstanceId { get; set; }
public bool PreventDuplicateScheduling { get; set; }
public IDictionary<string,object>? Input { get; set; }
}

6
src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs
View file

@ -77,7 +77,9 @@ public class WorkflowStateExtractor : IWorkflowStateExtractor
private void ApplyProperties(WorkflowState state, WorkflowExecutionContext workflowExecutionContext)
{
workflowExecutionContext.Properties = state.Properties;
// Merge properties.
foreach (var property in state.Properties)
workflowExecutionContext.Properties[property.Key] = property.Value;
}
private static void ApplyActivityExecutionContexts(WorkflowState state, WorkflowExecutionContext workflowExecutionContext)
@ -248,7 +250,7 @@ public class WorkflowStateExtractor : IWorkflowStateExtractor
// // If there are any faulted contexts, keep everything so that the user can fix the issue and potentially reschedule existing instances.
// if (contexts.Any(x => x.Status == ActivityStatus.Faulted))
return contexts;
return contexts;
// return contexts
// .Where(x => !x.IsCompleted)

5
src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowInbox.cs
View file

@ -32,13 +32,14 @@ public interface IWorkflowInbox
/// <param name="message">The message to deliver.</param>
/// <param name="cancellationToken">An optional cancellation token.</param>
ValueTask<DeliverWorkflowInboxMessageResult> DeliverAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default);
/// <summary>
/// Broadcasts the specified message, which may trigger new workflows and resume existing ones.
/// </summary>
/// <param name="message">The message to broadcast.</param>
/// <param name="options">An optional set of delivery options.</param>
/// <param name="cancellationToken">An optional cancellation token.</param>
ValueTask<DeliverWorkflowInboxMessageResult> BroadcastAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default);
ValueTask<DeliverWorkflowInboxMessageResult> BroadcastAsync(WorkflowInboxMessage message, BroadcastWorkflowInboxMessageOptions? options, CancellationToken cancellationToken = default);
/// <summary>
/// Finds all messages matching the specified filter.

4
src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs
View file

@ -11,7 +11,7 @@ namespace Elsa.Extensions;
public static class ActivityExecutionPipelineBuilderExtensions
{
/// <summary>
/// Installs the <see cref="BackgroundActivityCollectorMiddleware"/>.
/// Installs the <see cref="BackgroundActivityInvokerMiddleware"/>.
/// </summary>
public static IActivityExecutionPipelineBuilder UseBackgroundActivityInvoker(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware<BackgroundActivityCollectorMiddleware>();
public static IActivityExecutionPipelineBuilder UseBackgroundActivityInvoker(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware<BackgroundActivityInvokerMiddleware>();
}

2
src/modules/Elsa.Workflows.Runtime/Handlers/CancelBackgroundActivities.cs
View file

@ -31,7 +31,7 @@ public class CancelBackgroundActivities : INotificationHandler<WorkflowBookmarks
/// <inheritdoc />
public async Task HandleAsync(WorkflowBookmarksIndexed notification, CancellationToken cancellationToken)
{
var removedBookmarks = notification.IndexedWorkflowBookmarks.RemovedBookmarks.Where(x => x.Name == BackgroundActivityCollectorMiddleware.BackgroundActivityBookmarkName);
var removedBookmarks = notification.IndexedWorkflowBookmarks.RemovedBookmarks.Where(x => x.Name == BackgroundActivityInvokerMiddleware.BackgroundActivityBookmarkName);
foreach (var removedBookmark in removedBookmarks)
{

11
src/modules/Elsa.Workflows.Runtime/Handlers/ReadWorkflowInboxMessage.cs
View file

@ -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<WorkflowInboxMessag
{
_workflowInbox = workflowInbox;
}
/// <inheritdoc />
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);
}
}

101
src/modules/Elsa.Workflows.Runtime/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs
View file

@ -1,6 +1,8 @@
using System.Text.Json;
using Elsa.Extensions;
using Elsa.Workflows.Middleware.Activities;
using Elsa.Workflows.Models;
using Elsa.Workflows.Options;
using Elsa.Workflows.Pipelines.ActivityExecution;
using Elsa.Workflows.Runtime.Bookmarks;
using Elsa.Workflows.Runtime.Middleware.Workflows;
@ -12,19 +14,17 @@ namespace Elsa.Workflows.Runtime.Middleware.Activities;
/// Collects the current activity for scheduling for execution from a background job if the activity is of kind <see cref="ActivityKind.Job"/> or <see cref="Task"/>.
/// The actual scheduling of the activity happens in <see cref="ScheduleBackgroundActivitiesMiddleware"/>.
/// </summary>
public class BackgroundActivityCollectorMiddleware : DefaultActivityInvokerMiddleware
public class BackgroundActivityInvokerMiddleware : DefaultActivityInvokerMiddleware
{
/// <summary>
/// A key into the activity execution context's transient properties that indicates whether the current activity is being executed in the background.
/// </summary>
public static readonly object IsBackgroundExecution = new();
internal static string GetBackgroundActivityOutputKey(string activityId) => $"__BackgroundActivityOutput:{activityId}";
internal static string GetBackgroundActivityOutputKey(string activityNodeId) => $"__BackgroundActivityOutput:{activityNodeId}";
internal static string GetBackgroundActivityOutcomesKey(string activityNodeId) => $"__BackgroundActivityOutcomes:{activityNodeId}";
internal static string GetBackgroundActivityJournalDataKey(string activityNodeId) => $"__BackgroundActivityJournalData:{activityNodeId}";
internal static string GetBackgroundActivityScheduledActivitiesKey(string activityNodeId) => $"__BackgroundActivityScheduledActivities:{activityNodeId}";
internal static readonly object BackgroundActivitySchedulesKey = new();
internal const string BackgroundActivityBookmarkName = "BackgroundActivity";
/// <inheritdoc />
public BackgroundActivityCollectorMiddleware(ActivityMiddlewareDelegate next) : base(next)
public BackgroundActivityInvokerMiddleware(ActivityMiddlewareDelegate next) : base(next)
{
}
@ -37,8 +37,17 @@ public class BackgroundActivityCollectorMiddleware : DefaultActivityInvokerMiddl
ScheduleBackgroundActivity(context);
else
{
CaptureOutputIfAny(context);
await base.ExecuteActivityAsync(context);
// This part is either executed from the background, or in the foreground when the activity is resumed.
var isResuming = !GetIsBackgroundExecution(context) && context.ActivityDescriptor.Kind is ActivityKind.Task or ActivityKind.Job;
if (isResuming)
{
CaptureOutputIfAny(context);
CaptureJournalData(context);
await CompleteBackgroundActivityOutcomesAsync(context);
await CompleteBackgroundActivityScheduledActivitiesAsync(context);
}
}
}
@ -51,11 +60,13 @@ public class BackgroundActivityCollectorMiddleware : DefaultActivityInvokerMiddl
var activityDescriptor = context.ActivityDescriptor;
var kind = activityDescriptor.Kind;
return !context.TransientProperties.ContainsKey(IsBackgroundExecution)
return !GetIsBackgroundExecution(context)
&& context.WorkflowExecutionContext.ExecuteDelegate == null
&& (kind is ActivityKind.Job || (kind == ActivityKind.Task && activity.GetRunAsynchronously()));
}
private static bool GetIsBackgroundExecution(ActivityExecutionContext context) => context.TransientProperties.ContainsKey(BackgroundActivityExecutionContextExtensions.IsBackgroundExecution);
/// <summary>
/// Schedules the current activity for execution in the background.
/// </summary>
@ -77,21 +88,79 @@ public class BackgroundActivityCollectorMiddleware : DefaultActivityInvokerMiddl
private static void CaptureOutputIfAny(ActivityExecutionContext context)
{
var activity = context.Activity;
var inputKey = GetBackgroundActivityOutputKey(activity.Id);
var inputKey = GetBackgroundActivityOutputKey(activity.NodeId);
var capturedOutput = context.WorkflowExecutionContext.GetProperty<IDictionary<string, object>>(inputKey);
if (!context.WorkflowInput.TryGetValue(inputKey, out var capturedOutput))
if (capturedOutput == null)
return;
var input = (IDictionary<string, object>)capturedOutput;
foreach (var inputEntry in input)
foreach (var outputEntry in capturedOutput)
{
var outputDescriptor = context.ActivityDescriptor.Outputs.FirstOrDefault(x => x.Name == inputEntry.Key);
var outputDescriptor = context.ActivityDescriptor.Outputs.FirstOrDefault(x => x.Name == outputEntry.Key);
if (outputDescriptor == null)
continue;
var output = (Output?)outputDescriptor.ValueGetter(activity);
context.Set(output, inputEntry.Value);
context.Set(output, outputEntry.Value);
}
}
private void CaptureJournalData(ActivityExecutionContext context)
{
var activity = context.Activity;
var journalDataKey = GetBackgroundActivityJournalDataKey(activity.NodeId);
var journalData = context.WorkflowExecutionContext.GetProperty<IDictionary<string, object>>(journalDataKey);
if (journalData == null)
return;
foreach (var journalEntry in journalData)
context.JournalData[journalEntry.Key] = journalEntry.Value;
}
private async Task CompleteBackgroundActivityOutcomesAsync(ActivityExecutionContext context)
{
var outcomesKey = GetBackgroundActivityOutcomesKey(context.NodeId);
var outcomes = context.WorkflowExecutionContext.GetProperty<ICollection<string>>(outcomesKey);
if (outcomes != null)
{
await context.CompleteActivityWithOutcomesAsync(outcomes.ToArray());
// Remove the outcomes from the workflow execution context.
context.WorkflowExecutionContext.Properties.Remove(outcomesKey);
}
}
private async Task CompleteBackgroundActivityScheduledActivitiesAsync(ActivityExecutionContext context)
{
var scheduledActivitiesKey = GetBackgroundActivityScheduledActivitiesKey(context.NodeId);
var scheduledActivitiesJson = context.WorkflowExecutionContext.GetProperty<string>(scheduledActivitiesKey);
var scheduledActivities = scheduledActivitiesJson != null ? JsonSerializer.Deserialize<ICollection<ScheduledActivity>>(scheduledActivitiesJson) : null;
if (scheduledActivities != null)
{
foreach (var scheduledActivity in scheduledActivities)
{
var activityNode = scheduledActivity.ActivityNodeId != null ? context.WorkflowExecutionContext.FindActivityByNodeId(scheduledActivity.ActivityNodeId) : null;
var owner = scheduledActivity.OwnerActivityInstanceId != null ? context.WorkflowExecutionContext.ActivityExecutionContexts.FirstOrDefault(x => x.Id == scheduledActivity.OwnerActivityInstanceId) : null;
var options = scheduledActivity.Options != null
? new ScheduleWorkOptions
{
ExistingActivityExecutionContext = scheduledActivity.Options.ExistingActivityInstanceId != null ? context.WorkflowExecutionContext.ActivityExecutionContexts.FirstOrDefault(x => x.Id == scheduledActivity.Options.ExistingActivityInstanceId) : null,
Variables = scheduledActivity.Options?.Variables,
CompletionCallback = !string.IsNullOrEmpty(scheduledActivity.Options?.CompletionCallback) && owner != null ? owner.Activity.GetActivityCompletionCallback(scheduledActivity.Options.CompletionCallback) : default,
PreventDuplicateScheduling = scheduledActivity.Options?.PreventDuplicateScheduling ?? false,
Input = scheduledActivity.Options?.Input,
Tag = scheduledActivity.Options?.Tag
}
: default;
await context.ScheduleActivityAsync(activityNode, owner, options);
}
// Remove the scheduled activities from the workflow execution context.
context.WorkflowExecutionContext.Properties.Remove(scheduledActivitiesKey);
}
}
}

2
src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/ScheduleBackgroundActivitiesMiddleware.cs
View file

@ -43,7 +43,7 @@ public class ScheduleBackgroundActivitiesMiddleware : WorkflowExecutionMiddlewar
var scheduledBackgroundActivities = workflowExecutionContext
.TransientProperties
.GetOrAdd(BackgroundActivityCollectorMiddleware.BackgroundActivitySchedulesKey, () => new List<ScheduledBackgroundActivity>());
.GetOrAdd(BackgroundActivityInvokerMiddleware.BackgroundActivitySchedulesKey, () => new List<ScheduledBackgroundActivity>());
if (scheduledBackgroundActivities.Any())
{

15
src/modules/Elsa.Workflows.Runtime/Models/BroadcastWorkflowInboxMessageOptions.cs Normal file
View file

@ -0,0 +1,15 @@
namespace Elsa.Workflows.Runtime.Models;
/// <summary>
/// Represents the options for broadcasting a workflow inbox message.
/// </summary>
public class BroadcastWorkflowInboxMessageOptions
{
/// <summary>
/// Gets or sets a value indicating whether the dispatch should be executed asynchronously.
/// </summary>
/// <value>
/// <c>true</c> if the dispatch should be executed asynchronously; otherwise, <c>false</c>.
/// </value>
public bool DispatchAsynchronously { get; set; } = true;
}

7
src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowInboxMessageReceived.cs
View file

@ -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.
/// </summary>
/// <param name="InboxMessage">The inbox message that was received.</param>
public record WorkflowInboxMessageReceived(WorkflowInboxMessage InboxMessage) : INotification;
public record WorkflowInboxMessageReceived(
WorkflowInboxMessage InboxMessage,
WorkflowInboxMessageDeliveryOptions Options,
ICollection<WorkflowExecutionResult> WorkflowExecutionResults) : INotification;

8
src/modules/Elsa.Workflows.Runtime/Options/WorkflowInboxMessageDeliveryOptions.cs
View file

@ -1,7 +1,3 @@
using Elsa.Mediator;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Runtime.Notifications;
namespace Elsa.Workflows.Runtime.Options;
/// <summary>
@ -10,7 +6,7 @@ namespace Elsa.Workflows.Runtime.Options;
public class WorkflowInboxMessageDeliveryOptions
{
/// <summary>
/// The strategy to use when publishing the <see cref="WorkflowInboxMessageReceived"/> notification.
/// Whether to dispatch the message to the workflow dispatcher or send immediately.
/// </summary>
public IEventPublishingStrategy EventPublishingStrategy { get; set; } = NotificationStrategy.Background;
public bool DispatchAsynchronously { get; set; } = true;
}

2
src/modules/Elsa.Workflows.Runtime/Results/DeliverWorkflowInboxMessageResult.cs
View file

@ -3,4 +3,4 @@ namespace Elsa.Workflows.Runtime.Results;
/// <summary>
/// Result of delivering a workflow inbox message.
/// </summary>
public record DeliverWorkflowInboxMessageResult;
public record DeliverWorkflowInboxMessageResult(ICollection<WorkflowExecutionResult> WorkflowExecutionResults);

36
src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs
View file

@ -1,3 +1,4 @@
using System.Text.Json;
using Elsa.Common.Models;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Helpers;
@ -57,6 +58,7 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker
public async Task ExecuteAsync(ScheduledBackgroundActivity scheduledBackgroundActivity, CancellationToken cancellationToken = default)
{
var workflowInstanceId = scheduledBackgroundActivity.WorkflowInstanceId;
var workflowState = await _workflowRuntime.ExportWorkflowStateAsync(workflowInstanceId, cancellationToken);
if (workflowState == null)
@ -69,7 +71,6 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker
var workflow = await _workflowDefinitionService.MaterializeWorkflowAsync(workflowDefinition, cancellationToken);
var workflowExecutionContext = await WorkflowExecutionContext.CreateAsync(_serviceProvider, workflow, workflowState, cancellationTokens: cancellationToken);
var originalBookmarks = workflowExecutionContext.Bookmarks.ToList();
var activityNodeId = scheduledBackgroundActivity.ActivityNodeId;
var activityExecutionContext = workflowExecutionContext.ActivityExecutionContexts.First(x => x.NodeId == activityNodeId);
@ -77,7 +78,7 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker
await _variablePersistenceManager.LoadVariablesAsync(workflowExecutionContext);
// Mark the activity as being invoked from a background worker.
activityExecutionContext.TransientProperties[BackgroundActivityCollectorMiddleware.IsBackgroundExecution] = true;
activityExecutionContext.SetIsBackgroundExecution();
// Invoke the activity.
await _activityInvoker.InvokeAsync(activityExecutionContext);
@ -111,32 +112,25 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker
outputValues[outputDescriptor.Name] = outputValue;
}
// TODO: Instead of importing the entire workflow state, we should only import the following:
// - Variables
// - Activity state
// - Activity output
// - Bookmarks
workflowState = _workflowStateExtractor.Extract(workflowExecutionContext);
await _variablePersistenceManager.SaveVariablesAsync(workflowExecutionContext);
await _workflowRuntime.ImportWorkflowStateAsync(workflowState, cancellationToken);
// Process bookmarks.
var newBookmarks = workflowExecutionContext.Bookmarks.ToList();
var diff = Diff.For(originalBookmarks, newBookmarks);
await _bookmarksPersister.PersistBookmarksAsync(workflowExecutionContext, diff);
// Resume the workflow, passing along the activity output.
// TODO: This approach will fail if the output is non-serializable. We need to find a way to pass the output to the workflow without serializing it.
// Resume the workflow, passing along activity output, outcomes and scheduled activities.
var bookmarkId = scheduledBackgroundActivity.BookmarkId;
var inputKey = BackgroundActivityCollectorMiddleware.GetBackgroundActivityOutputKey(activityNodeId);
var inputKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityOutputKey(activityNodeId);
var outcomesKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityOutcomesKey(activityNodeId);
var journalDataKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityJournalDataKey(activityNodeId);
var scheduledActivitiesKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityScheduledActivitiesKey(activityNodeId);
var outcomes = activityExecutionContext.GetBackgroundOutcomes().ToList();
var scheduledActivities = activityExecutionContext.GetBackgroundScheduledActivities().ToList();
var dispatchRequest = new DispatchWorkflowInstanceRequest
{
InstanceId = workflowInstanceId,
BookmarkId = bookmarkId,
Input = new Dictionary<string, object>
Properties = new Dictionary<string, object>
{
[inputKey] = outputValues
[outcomesKey] = outcomes,
[scheduledActivitiesKey] = JsonSerializer.Serialize(scheduledActivities),
[inputKey] = outputValues,
[journalDataKey] = activityExecutionContext.JournalData
}
};

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

@ -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
/// </summary>
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<WorkflowExecutionResult>();
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
/// <inheritdoc />
public async ValueTask<DeliverWorkflowInboxMessageResult> DeliverAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default)
{
await ResumeWorkflowsAsync(message, cancellationToken);
return new DeliverWorkflowInboxMessageResult();
await ResumeWorkflowsAsynchronouslyAsync(message, cancellationToken);
return new DeliverWorkflowInboxMessageResult(new List<WorkflowExecutionResult>());
}
/// <inheritdoc />
public async ValueTask<DeliverWorkflowInboxMessageResult> 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<DeliverWorkflowInboxMessageResult> 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<WorkflowExecutionResult>());
}
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<WorkflowExecutionResult>());
}
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<IEnumerable<WorkflowExecutionResult>> 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
});
}
/// <inheritdoc />
public async ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default)
{

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

@ -423,11 +423,5 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
private async Task<IDistributedSynchronizationHandle> AcquireLockAsync(string resource, CancellationToken cancellationToken)
{
return await _distributedLockProvider.AcquireLockAsync(resource, TimeSpan.FromMinutes(2), cancellationToken);
// if (AcquiredLock.Value?.Key == resource)
// return AcquiredLock.Value.Lock;
//
// var distributedLock = await _distributedLockProvider.AcquireLockAsync(resource, TimeSpan.FromMinutes(2), cancellationToken);
// AcquiredLock.Value = new AcquiredLock { Lock = distributedLock, Key = resource };
// return distributedLock;
}
}

8
src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs
View file

@ -31,7 +31,7 @@ public class EventPublisher : IEventPublisher
IDictionary<string, object>? 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);
}
/// <inheritdoc />
@ -43,12 +43,12 @@ public class EventPublisher : IEventPublisher
IDictionary<string, object>? 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<ICollection<WorkflowExecutionResult>> 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<Event>(eventBookmark, workflowInstanceId, correlationId, activityInstanceId, input);
var options = new WorkflowInboxMessageDeliveryOptions
{
EventPublishingStrategy = publishingStrategy,
DispatchAsynchronously = dispatchAsynchronously
};
var result = await _workflowInbox.SubmitAsync(message, options, cancellationToken);