Как я могу передать объект `Сообщение` на маршрут?

Question

Я создаю поток, который принимает сообщения от RabbitMQ и после этого распределяет по соответствующим службам по типу, используя маршрутизатор. Методы в службах принимают аргумент Message<?>, потому что мне нужно использовать там заголовки. Но в этом методе я получаю только полезные сообщения с типом java.lang.String вместо org.springframework.messaging.Message и Я получаю ошибку java.lang.ClassCastException: java.lang.String cannot be cast to org.springframework.messaging.Message.

Полезная нагрузка мне не подходит, потому что мне нужно получать заголовки из сообщения.

@Bean
public IntegrationFlow testFlow(String queueName,
                                ConnectionFactory connectionFactory,
                                Service1 service1,
                                Service2 service2) {
    SimpleMessageListenerContainer consumerListener = new SimpleMessageListenerContainer(connectionFactory);
    consumerListener.addQueueNames(queueName);
    return IntegrationFlows.from(Amqp.inboundAdapter(consumerListener))
            .transform(s -> s, ConsumerEndpointSpec::transactional)
            .<Message<?>, String>route(HeadersUtil::getType, m -> m
                    .subFlowMapping(Type.SERVICE_1, sf -> sf.handle(service1::handleProcedure))
                    .subFlowMapping(Type.SERVICE_2, sf -> sf.handle(service2::handleProcedure)))
            .get();
}

Подпись метода handleProcedure выглядит следующим образом:

void handleProcedure(Message<?> message)

Я ожидаю получить заголовки Message в методе handleProcedure, но сейчас получаю исключение.

Answer 1

Я думаю, вы не правильно поняли трассировку стека.

Ваш void handleProcedure(Message<?> message) и его ссылка на метод service1::handleProcedure полностью соответствуют сигнатуре public B handle(MessageHandler messageHandler) { метода в IntegrationFlowDefinition.

Ваша проблема здесь:

.<Message<?>, String>route(HeadersUtil::getType, 

Ваш HeadersUtil::getType ожидает сообщения, но тип для лямбда-вызова - payload, который в вашем случае равен String.

Это должно работать:

.<Message<?>, String>route(Message.class, HeadersUtil::getType, 

Answer 2

Используемая вами handle(GenericHandler<P>) версия .handle получает только полезную нагрузку.

Если вы хотите получить полное сообщение, вам нужно использовать другую перегруженную .handle, напримеркак handle("service1", "handleProcedure") или .handle(service1, "handleProcedure").