diff --git a/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs b/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs index d0a32890a..905bc9d53 100644 --- a/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs +++ b/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs @@ -32,20 +32,32 @@ public class MessageReceived : Trigger 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(InputKey, out var message)) + if (!context.TryGetWorkflowInput(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.