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

Fix consumption of event payload (#4909)

Browse source 
The 'input' parameter in 'PublishAsync', 'DispatchAsync' and related methods across multiple classes has been replaced with 'payload'. The renaming of 'input' to 'payload' provides a more intuitive understanding of the parameter as the actual content or data of the event being published or dispatched. It also includes changes in 'PublishInternalAsync' where the workflow input dictionary is now being populated with the payload.
This commit is contained in:
Sipke Schoorstra 2024-02-09 08:20:54 +01:00 committed by GitHub
parent cfb6fa532b
commit d4e1cf0985
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 21 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -71,7 +71,6 @@ services
.AddWorkflowsFrom<Program>()
.UseFluentStorageProvider()
.UseFileStorage()
// .UseFileStorage(sp => StorageFactory.Blobs.AzureBlobStorageWithSas(configuration.GetConnectionString("AzureStorageSasUrl")))
.UseIdentity(identity =>
{
if (useMongoDb)

4
src/modules/Elsa.Workflows.Runtime/Activities/Event.cs
View file

@ -16,6 +16,8 @@ namespace Elsa.Workflows.Runtime.Activities;
[UsedImplicitly]
public class Event : Trigger<object?>
{
internal const string EventPayloadWorkflowInputKey = "__EventPayloadWorkflowInput";
/// <inheritdoc />
internal Event([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
{
@ -79,6 +81,8 @@ public class Event : Trigger<object?>
return;
}
var input = context.GetWorkflowInput<object?>(EventPayloadWorkflowInputKey);
context.SetResult(input);
await context.CompleteActivityAsync();
}
}

10
src/modules/Elsa.Workflows.Runtime/Activities/PublishEvent.cs
View file

@ -40,8 +40,8 @@ public class PublishEvent : Activity
/// <summary>
/// The input to send as the event body.
/// </summary>
[Input(Description = "The input to send as the event body.")]
public Input<IDictionary<string, object>?> Input { get; set; } = default!;
[Input(Description = "The payload to send as the event body.")]
public Input<object> Payload { get; set; } = default!;
/// <inheritdoc />
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
@ -50,10 +50,10 @@ public class PublishEvent : Activity
var correlationId = CorrelationId.GetOrDefault(context);
var isLocalEvent = IsLocalEvent.GetOrDefault(context);
var workflowInstanceId = isLocalEvent ? context.WorkflowExecutionContext.Id : default;
var input = Input.GetOrDefault(context);
var payload = Payload.GetOrDefault(context);
var publisher = context.GetRequiredService<IEventPublisher>();
await publisher.DispatchAsync(eventName, correlationId, workflowInstanceId, null, input, context.CancellationToken);
await publisher.DispatchAsync(eventName, correlationId, workflowInstanceId, null, payload, context.CancellationToken);
await context.CompleteActivityAsync();
}
}

4
src/modules/Elsa.Workflows.Runtime/Contracts/IEventPublisher.cs
View file

@ -11,10 +11,10 @@ public interface IEventPublisher
/// <summary>
/// Synchronously publishes the specified event using the workflow runtime, effectively triggering all <see cref="Event"/> activities matching the provided event name.
/// </summary>
Task<ICollection<WorkflowExecutionResult>> PublishAsync(string eventName, string? correlationId = default, string? workflowInstanceId = default, string? activityInstanceId = default, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default);
Task<ICollection<WorkflowExecutionResult>> PublishAsync(string eventName, string? correlationId = default, string? workflowInstanceId = default, string? activityInstanceId = default, object? payload = default, CancellationToken cancellationToken = default);
/// <summary>
/// Asynchronously publishes the specified event using the workflow dispatcher.
/// </summary>
Task DispatchAsync(string eventName, string? correlationId = default, string? workflowInstanceId = default, string? activityInstanceId = default, IDictionary<string, object>? input = default, CancellationToken cancellationToken = default);
Task DispatchAsync(string eventName, string? correlationId = default, string? workflowInstanceId = default, string? activityInstanceId = default, object? payload = default, CancellationToken cancellationToken = default);
}

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

@ -28,10 +28,10 @@ public class EventPublisher : IEventPublisher
string? correlationId = default,
string? workflowInstanceId = default,
string? activityInstanceId = default,
IDictionary<string, object>? input = default,
object? payload = default,
CancellationToken cancellationToken = default)
{
return await PublishInternalAsync(eventName, false, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken);
return await PublishInternalAsync(eventName, false, correlationId, workflowInstanceId, activityInstanceId, payload, cancellationToken);
}
/// <inheritdoc />
@ -40,10 +40,10 @@ public class EventPublisher : IEventPublisher
string? correlationId = default,
string? workflowInstanceId = default,
string? activityInstanceId = default,
IDictionary<string, object>? input = default,
object? payload = default,
CancellationToken cancellationToken = default)
{
await PublishInternalAsync(eventName, true, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken);
await PublishInternalAsync(eventName, true, correlationId, workflowInstanceId, activityInstanceId, payload, cancellationToken);
}
private async Task<ICollection<WorkflowExecutionResult>> PublishInternalAsync(
@ -52,11 +52,15 @@ public class EventPublisher : IEventPublisher
string? correlationId = default,
string? workflowInstanceId = default,
string? activityInstanceId = default,
IDictionary<string, object>? input = default,
object? payload = default,
CancellationToken cancellationToken = default)
{
var eventBookmark = new EventBookmarkPayload(eventName);
var message = NewWorkflowInboxMessage.For<Event>(eventBookmark, workflowInstanceId, correlationId, activityInstanceId, input);
var workflowInput = new Dictionary<string, object>
{
[Event.EventPayloadWorkflowInputKey] = payload ?? new Dictionary<string, object>()
};
var message = NewWorkflowInboxMessage.For<Event>(eventBookmark, workflowInstanceId, correlationId, activityInstanceId, workflowInput);
var options = new WorkflowInboxMessageDeliveryOptions
{
DispatchAsynchronously = dispatchAsynchronously