Mismatch between PublishMessage and Stimulus message

[muharemkapo]

Hello,

white trying out PublishMessage activity using MassTransit over RabbitMq, I have noticed that hashes created for the registered trigger and for the stimulus (which should match the trigger's hash) are different. Here is the workflow that publishes the message:

Here is the workload that receives it:

Here is the definition of the message:

I have also added this in the MassTransit configuration:

When stopping at the breakpoint here, one hash is shown:

When triggering message, stimulus has a different hash:

This is caused by the names in the hash function being different, i.e. for the trigger "MessageReceived" is the used name, whereas for the stimulus, "OcrDispatchWorkflowInput" is being used.

These are the packages I am using:

I wish I had time to fix this myself, but unfortunately that is not the case. Let me know if you need more details.

[muharemkapo]

Hello, as a workaround I have created my own consumer:

public class EvoWorkflowMessageConsumer<T>(IStimulusSender stimulusSender) : IConsumer<T>
    where T : BaseMassTransitMessage
{
    /// <inheritdoc />
    public async Task Consume(ConsumeContext<T> context)
    {
        var cancellationToken = context.CancellationToken;
        var messageType = typeof(T);
        var message = context.Message;
        var activityTypeName = ActivityTypeNameHelper.GenerateTypeName(message.GetTriggeredType());
        var stimulus = new MyRabbitTriggerBookmarkPayload(messageType);
        
        var correlationId = context.CorrelationId?.ToString();
        var input = new Dictionary<string, object>
        {
            [OcrRabbitTrigger.InputKey] = message
        };
        var stimulusMetadata = new StimulusMetadata
        {
            CorrelationId = correlationId,
            Input = input
        };
        await stimulusSender.SendAsync(activityTypeName, stimulus, stimulusMetadata, cancellationToken);
    }
}

Registered it like this:

.UseMassTransit(massTransit =>
                {
                    massTransit.AddConsumer<EvoWorkflowMessageConsumer<MyFinishedSignalMessage>>("my-finished-signal-message", false);
                    massTransit.UseRabbitMq(feature =>
                    {

Created my own trigger and message like this:

[Browsable(false)]
public class MyRabbitTrigger : Trigger<object>
{
    public const string InputKey = "Message";

    /// <inheritdoc />
    public MyRabbitTrigger([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)
    {
    }

    /// <summary>
    /// The message type to receive.
    /// </summary>
    public Type MessageType { get; set; } = null!;

    /// <inheritdoc />
    protected override object GetTriggerPayload(TriggerIndexingContext context) => GetBookmarkPayload(context.ExpressionExecutionContext);

    /// <inheritdoc />
    protected override 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 (!TryGetMessage(context, out var message))
        {
            // Create bookmarks for when we receive the expected HTTP request.
            context.CreateBookmark(GetBookmarkPayload(context.ExpressionExecutionContext), ResumeAsync, includeActivityInstanceId: false);
            return default;
        }

        return ExecuteInternalAsync(context, message);
    }

    private ValueTask ResumeAsync(ActivityExecutionContext context)
    {
        if(!TryGetMessage(context, out var message))
            throw new InvalidOperationException("Message was not received.");
        return ExecuteInternalAsync(context, message);
    }

    private ValueTask ExecuteInternalAsync(ActivityExecutionContext context, object message)
    {
        // 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.
        return context.CompleteActivityAsync();
    }

    private bool TryGetMessage(ActivityExecutionContext context, out object message)
    {
        if (!context.TryGetWorkflowInput(InputKey, out message))
            return false;

        if (message.GetType() == MessageType)
            return true;

        message = null!;
        return false;
    }

    private object GetBookmarkPayload(ExpressionExecutionContext context)
    {
        // Generate bookmark data for message type.
        return new MyRabbitTriggerBookmarkPayload(MessageType);
    }
}

public record MyRabbitTriggerBookmarkPayload(Type MessageType)
{
    public static Type HandledType => typeof(MyRabbitTrigger);
}

public class MyFinishedSignalMessage(string message) : BaseMassTransitMessage
{
    public string Message { get; init; } = message;

    public override Type GetTriggeredType()
    {
        return typeof(MyRabbitTrigger);
    }
}

So instead of using MessageReceived, I used this MyRabbitTrigger in my workflow. It might not be the best way since the message is tied to trigger and cannot be used for multiple triggers, but this was the easiest way to go around the issue and still use RabbitMq as triggering mechanism, which would allow triggering workflow from the external sources (not sure if this is possible with dispatched workflow).

Also, I have decorated MassTransitActivityProvider (IActivityProvider) with my own instance of IActivityProvider:

public class EvoMassTransitActivityProvider(
    IActivityProvider activityProvider,
    IActivityFactory activityFactory,
    IActivityDescriber activityDescriber) : IActivityProvider
{
    /// <inheritdoc />
    public async ValueTask<IEnumerable<ActivityDescriptor>> GetDescriptorsAsync(
        CancellationToken cancellationToken = default)
    {
        var descriptors = (await activityProvider.GetDescriptorsAsync(cancellationToken)).ToList();
        descriptors.Add(await CreateOcrRabbitTriggerDescriptor(cancellationToken));
        return descriptors;
    }

    private async Task<ActivityDescriptor> CreateOcrRabbitTriggerDescriptor(
        CancellationToken cancellationToken = default)
    {
        var messageType = typeof(OcrFinishedSignalMessage);
        
        var activityAttr = messageType.GetCustomAttribute<ActivityAttribute>();
        var typeName = activityAttr?.Type ?? messageType.Name;
        var fullTypeName = ActivityTypeNameHelper.GenerateTypeName(messageType);
        var displayNameAttr = messageType.GetCustomAttribute<DisplayNameAttribute>();
        var displayName = displayNameAttr?.DisplayName ??
                          activityAttr?.DisplayName ?? typeName.Humanize(LetterCasing.Title);
        var categoryAttr = messageType.GetCustomAttribute<CategoryAttribute>();
        var category = categoryAttr?.Category ?? activityAttr?.Category ?? "MassTransit";
        var descriptionAttr = messageType.GetCustomAttribute<DescriptionAttribute>();
        var description = descriptionAttr?.Description ?? activityAttr?.Description;

        var outputDescriptor =
            await activityDescriber.DescribeOutputProperty<OcrRabbitTrigger, object>(x => x.Result!, cancellationToken);
        var openOutputType = typeof(Output<>);
        var outputType = openOutputType.MakeGenericType(messageType);
        outputDescriptor.Type = outputType;

        return new ActivityDescriptor
        {
            Name = typeName,
            TypeName = fullTypeName,
            Version = 1,
            DisplayName = displayName,
            Description = description,
            Category = category,
            Kind = ActivityKind.Trigger,
            IsBrowsable = true,
            Outputs =
            {
                outputDescriptor
            },
            Constructor = context =>
            {
                var activity = (OcrRabbitTrigger)activityFactory.Create(typeof(OcrRabbitTrigger), context);
                activity.Type = fullTypeName;
                activity.MessageType = messageType;
                return activity;
            }
        };
    }
}

Decorated it like this:

webApplicationBuilder.Services.Decorate<IActivityProvider, EvoMassTransitActivityProvider>();

Not sure if this was necessary.