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

Merge pull request #6302 from elsa-workflows/bug/6294

Browse source 
Refine message handling in MessageReceived activity
This commit is contained in:
Sipke Schoorstra 2025-01-15 15:04:10 +01:00 committed by GitHub
parent eba7eb289d 1ee8fbdb5a
commit 3966824d29
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 48 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
src/modules/Elsa.MassTransit/Activities/MessageReceived.cs
View file

@ -32,20 +32,32 @@ public class MessageReceived : Trigger<object>
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
// If we did not receive external input, it means we are just now encountering this activity and we need to block execution by creating a bookmark.
if (!context.TryGetWorkflowInput<object>(InputKey, out var message))
if (!context.TryGetWorkflowInput<object>(InputKey, out var message) || message.GetType() != MessageType)
{
// Create bookmarks for when we receive the expected HTTP request.
context.CreateBookmark(GetBookmarkPayload(context.ExpressionExecutionContext));
context.CreateBookmark(GetBookmarkPayload(context.ExpressionExecutionContext), ResumeAsync, includeActivityInstanceId: false);
return;
}
// Provide the received message as output.
context.Set(Result, message);
// Remove the input to prevent it from being passed to the next activity.
context.WorkflowInput.Remove(InputKey);
// Complete.
await context.CompleteActivityAsync();
}
private ValueTask ResumeAsync(ActivityExecutionContext context)
{
// Remove the input to prevent it from being passed to the next activity.
context.WorkflowInput.Remove(InputKey);
// Complete.
return context.CompleteActivityAsync();
}
private object GetBookmarkPayload(ExpressionExecutionContext context)
{
// Generate bookmark data for message type.

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

@ -198,7 +198,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="completionCallback">An optional callback to invoke when the activity completes.</param>
/// <param name="tag">An optional tag to associate with the activity execution.</param>
/// <param name="variables">An optional list of variables to declare with the activity execution.</param>
public ValueTask ScheduleActivityAsync(IActivity? activity, ActivityCompletionCallback? completionCallback, object? tag = default, IEnumerable<Variable>? variables = default)
public ValueTask ScheduleActivityAsync(IActivity? activity, ActivityCompletionCallback? completionCallback, object? tag = null, IEnumerable<Variable>? variables = null)
{
var options = new ScheduleWorkOptions
{
@ -214,7 +214,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// </summary>
/// <param name="activity">The activity to schedule.</param>
/// <param name="options">The options used to schedule the activity.</param>
public async ValueTask ScheduleActivityAsync(IActivity? activity, ScheduleWorkOptions? options = default)
public async ValueTask ScheduleActivityAsync(IActivity? activity, ScheduleWorkOptions? options = null)
{
await ScheduleActivityAsync(activity, this, options);
}
@ -225,7 +225,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="activity">The activity to schedule.</param>
/// <param name="owner">The activity execution context that owns the scheduled activity.</param>
/// <param name="options">The options used to schedule the activity.</param>
public async ValueTask ScheduleActivityAsync(IActivity? activity, ActivityExecutionContext? owner, ScheduleWorkOptions? options = default)
public async ValueTask ScheduleActivityAsync(IActivity? activity, ActivityExecutionContext? owner, ScheduleWorkOptions? options = null)
{
var activityNode = activity != null
? WorkflowExecutionContext.FindNodeByActivity(activity) ?? throw new InvalidOperationException("The specified activity is not part of the workflow.")
@ -239,7 +239,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="activityNode">The activity node to schedule.</param>
/// <param name="owner">The activity execution context that owns the scheduled activity.</param>
/// <param name="options">The options used to schedule the activity.</param>
public async ValueTask ScheduleActivityAsync(ActivityNode? activityNode, ActivityExecutionContext? owner = default, ScheduleWorkOptions? options = default)
public async ValueTask ScheduleActivityAsync(ActivityNode? activityNode, ActivityExecutionContext? owner = null, ScheduleWorkOptions? options = null)
{
if (this.GetIsBackgroundExecution())
{
@ -261,7 +261,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
Variables = options?.Variables?.ToList(),
Input = options?.Input
}
: default
: null
};
var scheduledActivities = this.GetBackgroundScheduledActivities().ToList();
@ -303,7 +303,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="completionCallback">The callback to invoke when the activities complete.</param>
/// <param name="tag">An optional tag to associate with the activity execution.</param>
/// <param name="variables">An optional list of variables to declare with the activity execution.</param>
public ValueTask ScheduleActivities(IEnumerable<IActivity?> activities, ActivityCompletionCallback? completionCallback, object? tag = default, IEnumerable<Variable>? variables = default)
public ValueTask ScheduleActivities(IEnumerable<IActivity?> activities, ActivityCompletionCallback? completionCallback, object? tag = null, IEnumerable<Variable>? variables = null)
{
var options = new ScheduleWorkOptions
{
@ -319,7 +319,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// </summary>
/// <param name="activities">The activities to schedule.</param>
/// <param name="options">The options used to schedule the activities.</param>
public async ValueTask ScheduleActivities(IEnumerable<IActivity?> activities, ScheduleWorkOptions? options = default)
public async ValueTask ScheduleActivities(IEnumerable<IActivity?> activities, ScheduleWorkOptions? options = null)
{
foreach (var activity in activities)
await ScheduleActivityAsync(activity, options);
@ -331,7 +331,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="payloads">The payloads to create bookmarks for.</param>
/// <param name="callback">An optional callback that is invoked when the bookmark is resumed.</param>
/// <param name="includeActivityInstanceId">Whether or not the activity instance ID should be included in the bookmark payload.</param>
public void CreateBookmarks(IEnumerable<object> payloads, ExecuteActivityDelegate? callback = default, bool includeActivityInstanceId = true)
public void CreateBookmarks(IEnumerable<object> payloads, ExecuteActivityDelegate? callback = null, bool includeActivityInstanceId = true)
{
foreach (var payload in payloads)
CreateBookmark(new CreateBookmarkArgs
@ -360,7 +360,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="callback">An optional callback that is invoked when the bookmark is resumed.</param>
/// <param name="metadata">Custom properties to associate with the bookmark.</param>
/// <returns>The created bookmark.</returns>
public Bookmark CreateBookmark(ExecuteActivityDelegate callback, IDictionary<string, string>? metadata = default)
public Bookmark CreateBookmark(ExecuteActivityDelegate callback, IDictionary<string, string>? metadata = null)
{
return CreateBookmark(new CreateBookmarkArgs
{
@ -377,7 +377,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="includeActivityInstanceId">Whether or not the activity instance ID should be included in the bookmark payload.</param>
/// <param name="customProperties">Custom properties to associate with the bookmark.</param>
/// <returns>The created bookmark.</returns>
public Bookmark CreateBookmark(object stimulus, ExecuteActivityDelegate callback, bool includeActivityInstanceId = true, IDictionary<string, string>? customProperties = default)
public Bookmark CreateBookmark(object stimulus, ExecuteActivityDelegate callback, bool includeActivityInstanceId = true, IDictionary<string, string>? customProperties = null)
{
return CreateBookmark(new CreateBookmarkArgs
{
@ -395,7 +395,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="includeActivityInstanceId">Specifies whether to include the activity instance ID in the bookmark information. Defaults to true.</param>
/// <param name="customProperties">Additional custom properties to associate with the bookmark. Defaults to null.</param>
/// <returns>The created bookmark.</returns>
public Bookmark CreateBookmark(object stimulus, bool includeActivityInstanceId, IDictionary<string, string>? customProperties = default)
public Bookmark CreateBookmark(object stimulus, bool includeActivityInstanceId, IDictionary<string, string>? customProperties = null)
{
return CreateBookmark(new CreateBookmarkArgs
{
@ -411,9 +411,9 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="stimulus">The payload to associate with the bookmark.</param>
/// <param name="metadata">Custom properties to associate with the bookmark.</param>
/// <returns>The created bookmark.</returns>
public Bookmark CreateBookmark(object stimulus, IDictionary<string, string>? metadata = default)
public Bookmark CreateBookmark(object stimulus, IDictionary<string, string>? metadata = null)
{
return CreateBookmark(new CreateBookmarkArgs
return CreateBookmark(new()
{
Stimulus = stimulus,
Metadata = metadata
@ -424,7 +424,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// Creates a bookmark so that this activity can be resumed at a later time.
/// Creating a bookmark will automatically suspend the workflow after all pending activities have executed.
/// </summary>
public Bookmark CreateBookmark(CreateBookmarkArgs? options = default)
public Bookmark CreateBookmark(CreateBookmarkArgs? options = null)
{
var payload = options?.Stimulus;
var callback = options?.Callback;
@ -569,7 +569,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// </summary>
/// <param name="output">The output.</param>
/// <returns>The output value.</returns>
public object? Get(Output? output) => output == null ? default : Get(output.MemoryBlockReference());
public object? Get(Output? output) => output == null ? null : Get(output.MemoryBlockReference());
/// <summary>
/// Gets the value of the specified memory block.
@ -593,7 +593,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
public T? Get<T>(MemoryBlockReference blockReference)
{
var value = Get(blockReference);
return value != default ? value.ConvertTo<T>() : default;
return value != null ? value.ConvertTo<T>() : default;
}
/// <summary>
@ -628,7 +628,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="blockReference">The memory block reference.</param>
/// <param name="value">The value to set.</param>
/// <param name="configure">An optional callback that can be used to configure the memory block.</param>
public void Set(MemoryBlockReference blockReference, object? value, Action<MemoryBlock>? configure = default) => ExpressionExecutionContext.Set(blockReference, value, configure);
public void Set(MemoryBlockReference blockReference, object? value, Action<MemoryBlock>? configure = null) => ExpressionExecutionContext.Set(blockReference, value, configure);
/// <summary>
/// Sets a value at the specified output.
@ -637,7 +637,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="value">The value to set.</param>
/// <param name="outputName">The name of the output.</param>
/// <typeparam name="T">The type of the output.</typeparam>
public void Set<T>(Output<T>? output, T? value, [CallerArgumentExpression("output")] string? outputName = default) => Set((Output?)output, value, outputName);
public void Set<T>(Output<T>? output, T? value, [CallerArgumentExpression("output")] string? outputName = null) => Set((Output?)output, value, outputName);
/// <summary>
/// Sets a value at the specified output.
@ -645,7 +645,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
/// <param name="output">The output.</param>
/// <param name="value">The value to set.</param>
/// <param name="outputName">The name of the output.</param>
public void Set(Output? output, object? value, [CallerArgumentExpression("output")] string? outputName = default)
public void Set(Output? output, object? value, [CallerArgumentExpression("output")] string? outputName = null)
{
// Store the value in the expression execution memory block.
ExpressionExecutionContext.Set(output, value);
@ -667,7 +667,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
private MemoryBlock? GetMemoryBlock(MemoryBlockReference locationBlockReference)
{
return ExpressionExecutionContext.TryGetBlock(locationBlockReference, out var memoryBlock) ? memoryBlock : default;
return ExpressionExecutionContext.TryGetBlock(locationBlockReference, out var memoryBlock) ? memoryBlock : null;
}
void IDisposable.Dispose()

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

@ -25,13 +25,13 @@ public static partial class ActivityExecutionContextExtensions
/// <summary>
/// Attempts to get a value from the input provided via <see cref="WorkflowExecutionContext"/>. If a value was found, an attempt is made to convert it into the specified type <code>T</code>.
/// </summary>
public static bool TryGetWorkflowInput<T>(this ActivityExecutionContext context, string key, out T value, JsonSerializerOptions? serializerOptions = default)
public static bool TryGetWorkflowInput<T>(this ActivityExecutionContext context, string key, out T value, JsonSerializerOptions? serializerOptions = null)
{
var wellKnownTypeRegistry = context.GetRequiredService<IWellKnownTypeRegistry>();
if (context.WorkflowInput.TryGetValue(key, out var v))
{
value = v.ConvertTo<T>(new ObjectConverterOptions(serializerOptions, wellKnownTypeRegistry))!;
value = v.ConvertTo<T>(new(serializerOptions, wellKnownTypeRegistry))!;
return true;
}
@ -42,15 +42,15 @@ public static partial class ActivityExecutionContextExtensions
/// <summary>
/// Gets a value from the input provided via <see cref="WorkflowExecutionContext"/>. If a value was found, an attempt is made to convert it into the specified type <code>T</code>.
/// </summary>
public static T GetWorkflowInput<T>(this ActivityExecutionContext context, JsonSerializerOptions? serializerOptions = default) => context.GetWorkflowInput<T>(typeof(T).Name, serializerOptions);
public static T GetWorkflowInput<T>(this ActivityExecutionContext context, JsonSerializerOptions? serializerOptions = null) => context.GetWorkflowInput<T>(typeof(T).Name, serializerOptions);
/// <summary>
/// Gets a value from the input provided via <see cref="WorkflowExecutionContext"/>. If a value was found, an attempt is made to convert it into the specified type <code>T</code>.
/// </summary>
public static T GetWorkflowInput<T>(this ActivityExecutionContext context, string key, JsonSerializerOptions? serializerOptions = default)
public static T GetWorkflowInput<T>(this ActivityExecutionContext context, string key, JsonSerializerOptions? serializerOptions = null)
{
var wellKnownTypeRegistry = context.GetRequiredService<IWellKnownTypeRegistry>();
return context.WorkflowInput[key].ConvertTo<T>(new ObjectConverterOptions(serializerOptions, wellKnownTypeRegistry))!;
return context.WorkflowInput[key].ConvertTo<T>(new(serializerOptions, wellKnownTypeRegistry))!;
}
/// <summary>
@ -61,7 +61,7 @@ public static partial class ActivityExecutionContextExtensions
/// <exception cref="Exception">Thrown when the specified activity does not implement <see cref="IActivityWithResult"/>.</exception>
public static void SetResult(this ActivityExecutionContext context, object? value)
{
var activity = context.Activity as IActivityWithResult ?? throw new Exception($"Cannot set result on activity {context.Activity.Id} because it does not implement {nameof(IActivityWithResult)}.");
var activity = context.Activity as IActivityWithResult ?? throw new($"Cannot set result on activity {context.Activity.Id} because it does not implement {nameof(IActivityWithResult)}.");
context.Set(activity.Result, value, "Result");
}
@ -79,7 +79,7 @@ public static partial class ActivityExecutionContextExtensions
/// <param name="storageDriverType">The type of storage driver to use for the variable.</param>
/// <param name="configure">A callback to configure the memory block.</param>
/// <returns>The created <see cref="Variable"/>.</returns>
public static Variable CreateVariable(this ActivityExecutionContext context, string name, object? value, Type? storageDriverType = default, Action<MemoryBlock>? configure = default) =>
public static Variable CreateVariable(this ActivityExecutionContext context, string name, object? value, Type? storageDriverType = null, Action<MemoryBlock>? configure = null) =>
context.ExpressionExecutionContext.CreateVariable(name, value, storageDriverType, configure);
/// <summary>
@ -90,7 +90,7 @@ public static partial class ActivityExecutionContextExtensions
/// <param name="value">The value of the variable.</param>
/// <param name="configure">A callback to configure the memory block.</param>
/// <returns>The created <see cref="Variable"/>.</returns>
public static Variable SetVariable(this ActivityExecutionContext context, string name, object? value, Action<MemoryBlock>? configure = default) =>
public static Variable SetVariable(this ActivityExecutionContext context, string name, object? value, Action<MemoryBlock>? configure = null) =>
context.ExpressionExecutionContext.SetVariable(name, value, configure);
/// <summary>

2
src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs
View file

@ -88,7 +88,7 @@ public static class WorkflowExecutionContextExtensions
: WorkflowExecutionContext.Noop;
// Store the bookmark to resume in the context.
workflowExecutionContext.ResumedBookmarkContext = new ResumedBookmarkContext(bookmark);
workflowExecutionContext.ResumedBookmarkContext = new(bookmark);
logger.LogDebug("Scheduled activity {ActivityId} to resume from bookmark {BookmarkId}", bookmarkedActivity.Id, bookmark.Id);
return workItem;

6
src/modules/Elsa.Workflows.Core/Models/Bookmark.cs
View file

@ -26,13 +26,13 @@ public record Bookmark(
string? ActivityInstanceId,
DateTimeOffset CreatedAt,
bool AutoBurn = true,
string? CallbackMethodName = default,
string? CallbackMethodName = null,
bool AutoComplete = true,
IDictionary<string, string>? Metadata = default)
IDictionary<string, string>? Metadata = null)
{
/// <inheritdoc />
[JsonConstructor]
public Bookmark() : this("", "", "", null, "", "", "", default, default)
public Bookmark() : this("", "", "", null, "", "", "", default, false)
{
}
}

4
src/modules/Elsa.Workflows.Runtime/Services/StimulusSender.cs
View file

@ -38,7 +38,7 @@ public class StimulusSender(
var resumed = await ResumeExistingWorkflowsAsync(stimulusHash, metadata, cancellationToken);
responses.AddRange(resumed);
return new SendStimulusResult(responses);
return new(responses);
}
private async Task<ICollection<RunWorkflowInstanceResponse>> TriggerNewWorkflowsAsync(string stimulusHash, StimulusMetadata? metadata = null, CancellationToken cancellationToken = default)
@ -129,7 +129,7 @@ public class StimulusSender(
WorkflowInstanceId = workflowInstanceId,
BookmarkId = metadata?.BookmarkId,
StimulusHash = stimulusHash,
Options = new ResumeBookmarkOptions
Options = new()
{
Input = input,
Properties = properties