From 96dd00838ac7babf2fe744aca3cc515d5ff67768 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Wed, 15 Jan 2025 14:27:37 +0100 Subject: [PATCH] Refine message handling in MessageReceived activity. Ensure type validation for incoming messages and clean up workflow input to prevent unintended data propagation. Added a dedicated ResumeAsync method to handle bookmark resumption logic consistently. Fixes #6294 --- .../Activities/MessageReceived.cs | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) 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.