MassTransit with RabbitMQ and Spring Boot: How to Align Message Metadata Handling for Request-Reply Pattern?

[oussamahdidou]

I'm working on integrating MassTransit with RabbitMQ in a .NET application, and I need to implement a request-reply pattern with a Spring Boot listener. However, there seems to be a difference in how MassTransit and Spring Boot handle the message metadata.

MassTransit Message Structure:

Properties
 - message_id: 10340000-afb3-7824-10a6-08dce792c869 
 - expiration: 30000
 - delivery_mode: 2
 - headers:
 - ConversationId: 10340000-afb3-7824-6a92-08dce792c869
 - MT-Host-Info: { "MachineName":"DESKTOP-KS55GT8",
 "ProcessName":"UsersService",
 "ProcessId":13328,
 "Assembly":"UsersService",
 "AssemblyVersion":"1.0.0.0",
 "FrameworkVersion":"8.0.4",
 "MassTransitVersion":"8.2.5.0",
 "OperatingSystemVersion":"Microsoft Windows NT 10.0.19045.0"
 }
 - MT-MessageType: urn:message:UsersService.Messages:RequestMessage
 - MT-Response-Address: rabbitmq://localhost/DESKTOPKS55GT8_UsersService_bus_ny4yyyfxsphnjwh1bdqqxrmedz?temporary=true
 - MT-Source-Address: rabbitmq://localhost/DESKTOPKS55GT8_UsersService_bus_ny4yyyfxsphnjwh1bdqqxrmedz?temporary=true
 - MessageId: 10340000-afb3-7824-10a6-08dce792c869
 - RequestId: 10340000-afb3-7824-9af9-08dce792c868
 - publishId: 2
 - content_type: application/json
 - Payload: - 31 bytes
 - Encoding: string
 - { "payload": "Test Value" }
 - Redelivered: false
The MT-Response-Address header is set by MassTransit, which contains the response address, but my Spring Boot listener is expecting this information to be in the reply_to header:

Spring Boot Expected Message Structure:

Properties
 - reply_to: amq.rabbitmq.reply-to.g1h2AA9yZXBseUAxMjM3MzY2MzIAAFNIAAAAAGb/7rs=.SzKpGjLslrBY1mMHuXMWhQ==
 - correlation_id: 1
 - priority: 0
 - delivery_mode: 2
 - headers:
     - __TypeId__: com.api.stockservice.domain.event.RequestMessage
 - content_encoding: UTF-8
 - content_type: application/json
 - Payload: - 49 bytes
 - Encoding: string
 - { "Payload": "etutrtjfjtf", "payload": "etutrtjfjtf" }

this is the springboot listener

package com.api.stockservice.infrastructure.messaging;

import com.api.stockservice.domain.event.RequestMessage;
import com.api.stockservice.domain.event.ResponseMessage;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

@Component
public class RequestListener {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RabbitListener(queues = "my-request-queue")
    @SendTo("request.messageProperties.header['MT-Response-Address']")
    public ResponseMessage handleRequest(@Payload RequestMessage request) {
        // Process the request
        String result = "Processed: " + request.getPayload();
        System.out.println("Received result" + result);
        // Create a response
        return new ResponseMessage(result);
    }
}

this is the masstransit requester

using MassTransit;
using UsersService.Messages;

namespace UsersService.Producers
{
    public class MyRequester
    {
        private readonly IRequestClient<RequestMessage> _requestClient;

        public MyRequester(IRequestClient<RequestMessage> requestClient)
        {
            _requestClient = requestClient;
        }

        public async Task SendRequest(string value)
        {
            var request = new RequestMessage { Payload = value };

            // Send the request and wait for the response
            var response = await _requestClient.GetResponse<ResponseMessage>(request);

            Console.WriteLine($"Response: {response.Message.Result}");
        }
    }

}

this is the masstransit config program.cs

builder.Services.AddMassTransit(x =>
{
    //x.AddConsumer<RequestConsumer>();
    x.AddConsumer<FirstEventConsumer>();

    x.AddConsumer<SecondEventConsumer>();
    x.AddRequestClient<RequestMessage>(new Uri("queue:my-request-queue"));
    x.UsingRabbitMq((context, cfg) =>
    {
     
        cfg.Host("rabbitmq://localhost", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.UseNewtonsoftRawJsonSerializer();
        cfg.UseNewtonsoftJsonDeserializer();
        
        cfg.ConfigureEndpoints(context);
        //cfg.ReceiveEndpoint("my-request-queue", e =>
        //{
        //    e.ConfigureConsumer<RequestConsumer>(context);
        //});
        cfg.ReceiveEndpoint("first-publish-queue", e =>
        {
            e.Bind("publish_exchange");
            e.ConfigureConsumer<FirstEventConsumer>(context);
        });
        cfg.ReceiveEndpoint("second-publish-queue", e =>
        {
            e.Bind("publish_exchange");
            e.ConfigureConsumer<SecondEventConsumer>(context);
        });
    });
});

builder.Services.AddMassTransitHostedService();

Questions

Is there any way to configure MassTransit to use properties instead of headers when sending messages to RabbitMQ, so that the message format aligns with how Spring Boot processes it? Essentially, I need MassTransit to follow a more standard message mechanism in terms of metadata handling.
so MassTransit can seamlessly send and reply with other systems that use the same approach for metadata handling as springboot