MassTransit с RabbitMQ и Spring Boot: Как согласовать обработку метаданных сообщений для паттерна запрос-ответ?

Question

Я работаю над интеграцией MassTransit с RabbitMQ в приложении .NET и мне нужно реализовать паттерн запрос-ответ с слушателем Spring Boot. Однако, похоже, что существует разница в том, как MassTransit и Spring Boot обрабатывают метаданные сообщений.

Структура сообщения MassTransit:

Свойства
 - 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

Заголовок MT-Response-Address устанавливается MassTransit и содержит адрес ответа, но мой слушатель Spring Boot ожидает, что эта информация будет в заголовке reply_to:

Ожидаемая структура сообщения Spring Boot:

Свойства
 - 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" }

это слушатель springboot

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) {
        // Обработка запроса
        String result = "Processed: " + request.getPayload();
        System.out.println("Получен результат: " + result);
        // Создание ответа
        return new ResponseMessage(result);
    }
}

это запросчик masstransit

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 };

            // Отправка запроса и ожидание ответа
            var response = await _requestClient.GetResponse<ResponseMessage>(request);

            Console.WriteLine($"Ответ: {response.Message.Result}");
        }
    }
}

это конфигурация masstransit в 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();

Вопросы

Есть ли способ настроить MassTransit так, чтобы использовать свойства вместо заголовков при отправке сообщений в RabbitMQ, чтобы формат сообщения соответствовал тому, как его обрабатывает Spring Boot? По сути, мне нужно, чтобы MassTransit следовал более стандартному механизму передачи сообщений в терминах обработки метаданных.
Таким образом, MassTransit может бесшовно отправлять и отвечать другим системам, которые используют тот же подход к обработке метаданных, что и springboot.

MassTransit с RabbitMQ и Spring Boot: Как согласовать обработку метаданных сообщений для паттерна запрос-ответ?

Вопрос или проблема

Вопросы

Ответ или решение

1. Конфигурация MassTransit

Пример настройки MassTransit

2. Настройка Spring Boot

Пример реализации слушателя Spring Boot

3. Устранение разногласий

Заключение