elsa-workflows / elsa-workflows/elsa-core

MassTransit & CorrelationId in concurrent processes

Open
#6,672 0 comments 0 reactions 0 assignees View on GitHub
triaged
Dominant language
C#
Stars
7.9k
Forks
1.5k
Avg merge
15h 22m
Merged PRs (30d)
114

Description

## Enhancement Request

### Enhancement Overview
This Issue is a continuation of [Issue#6336](https://github.com/elsa-workflows/elsa-core/issues/6336)

### Proposed Enhancement
According to [MassTransit documentation](https://masstransit.io/documentation/concepts/messages#message-correlation) there are some correlation conventions.
So in MassTransit library, if there is a property in the message with name "CorrelationId" of type "Guid" it will be used for CorrelationId.

Why not use a similar convention in Elsa MassTransit Feature when processing messages in WorkflowMessageConsumer? But instead of type "Guid" I suggest using type "string".

I've tested this code and it works fine.
`
public async Task Consume(ConsumeContext context)
{
var cancellationToken = context.CancellationToken;
var messageType = typeof(T);
var message = context.Message;
var activityTypeName = ActivityTypeNameHelper.GenerateTypeName(messageType);
var stimulus = new MessageReceivedBookmarkPayload(messageType);

// =====================================================================
// Looking for CorrelationId
// =====================================================================
// 1-st priority: from context
string? correlationId = context.CorrelationId?.ToString();

// 2-nd priority: from message, if one has property with name CorrelationId with type string
if (string.IsNullOrEmpty(correlationId))
{
PropertyInfo? property = messageType.GetProperty(CorrelationIdPropName, typeof(string));
correlationId = property?.GetValue(message)?.ToString();
}
// =====================================================================

var input = new Dictionary
{
[MessageReceived.InputKey] = message
};
var stimulusMetadata = new StimulusMetadata
{
CorrelationId = correlationId,
Input = input
};
await stimulusSender.SendAsync(activityTypeName, stimulus, stimulusMetadata, cancellationToken);
}
`

Please have a look at this, and consider including it in next release of Elsa.

It's just a couple of lines, but it takes off serious limitation of using MassTransit feature in concurrent processes.

### Alternative Solutions

### Use Cases

### Impact of Enhancement

### Visuals and Mockups

### Additional Context

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.