Spring Integration: интеграция REST в JMS дает ошибку времени ожидания

Question

Привет я пытаюсь реализовать следующий поток с помощью Spring Integration. Предоставьте сервис REST, управляйте полезной нагрузкой, запишите измененную полезную нагрузку в JMS и ответьте на сервис REST.

@Bean
public IntegrationFlow httpInputFlow() {
    return IntegrationFlows
            .from(Http.inboundGateway(“/company”)
                     .requestMapping(r -> r
                             .methods(HttpMethod.POST))
                    .requestPayloadType(com.poc.model.Company.class))
            .transform(jsonToObjectTransformer())
            .channel(requestChannel())
            .get();
}

@Bean
@Transformer(inputChannel=“requestChannel”, outputChannel=“responseChannel”)
ObjectToJsonTransformer jsonToObjectTransformer() {
    return new ObjectToJsonTransformer();
}
@Bean
public DirectChannel requestChannel() {
    return MessageChannels.direct().get();
}
@Bean
public DirectChannel responseChannel() {
    return MessageChannels.direct().get();
}
@Bean
IntegrationFlow outboundFlow() throws Exception {
    return IntegrationFlows.from(responseChannel()) .handle(Jms.outboundAdapter(connectionFactory()).destination(“samplequeue”))
            .get();
}

Полезная нагрузка запроса достигает очереди. Однако тайм-аут службы REST без ответа. Мне нужно, чтобы преобразованная полезная нагрузка была помещена в очередь, а также отправлена ​​в ответ для остальной службы. Я не смог найти доступных примеров.

Любая помощь приветствуется.

Answer 1

responseChannel должно быть publishSubscribe и у тебя должен быть еще один bridge subscrbiber, позволяющий отправить ответ обратно на входящий HTTP-шлюз. Jms.outboundAdapter() в одностороннем компоненте и не дает никаких ответов. Вот как ваш поток останавливается и не отвечает обратно к точке входа.

Ваш вариант использования может быть написан так:

@Bean
public IntegrationFlow httpInputFlow() {
    return IntegrationFlows
            .from(Http.inboundGateway("/company")
                    .requestMapping(r -> r
                            .methods(HttpMethod.POST))
                    .reqestPayloadType(com.poc.model.Company.class))
            .transform(jsonToObjectTransformer())
            .publishSubscribeChannel(subscribers ->
                    subscribers.subscribe(f -> f
                        .handle(Jms.outboundAdapter(connectionFactory()).destination("samplequeue"))))
            .bridge(null)
            .get();
}

publishSubscribeChannel() обеспечивает необходимый канал.subscribers.subscribe() добавить что Jms.outboundAdapter() в качестве первого подписчика. .bridge() добавляется как последний (второй в нашем случае) подписчик publishSubscribeChannel(),

Идея позади BridgeHandler отправить свой ответ replyChannel заголовок, который заполняется Http.inboundGateway(),