I would like to create an application that would perform the following steps:
- Receive request thru RestController
- Send the received message to a queue (AMQP - MessageChannel) (correlationId?)
- Wait for the reply in another queue (AMQP - MessageChannel) (correlationId?)
- Return the response on the same thread as the request in step 1.
I thought about using IntegrationFlow for this, but I'm not able to adapt the steps.
Besides, would you know what is the best way to implement this flow?
I tried to implement with the following code:
package poc.integration.http;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpMethod;
import org.springframework.integration.amqp.channel.PollableAmqpChannel;
import org.springframework.integration.amqp.dsl.Amqp;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.http.dsl.Http;
import org.springframework.integration.http.inbound.HttpRequestHandlingMessagingGateway;
import org.springframework.integration.http.inbound.RequestMapping;
import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.messaging.MessageChannel;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@Configuration
@EnableIntegration
public class IntegrationFlowConfig {
final Logger logger = LoggerFactory.getLogger(IntegrationFlowConfig.class);
@Bean
public HttpRequestHandlingMessagingGateway inbound() {
HttpRequestHandlingMessagingGateway gateway = new HttpRequestHandlingMessagingGateway(true);
gateway.setRequestMapping(mapping());
gateway.setRequestPayloadTypeClass(String.class);
gateway.setRequestChannelName("httpRequest");
return gateway;
}
@Bean
public RequestMapping mapping() {
RequestMapping requestMapping = new RequestMapping();
requestMapping.setPathPatterns("/foo");
requestMapping.setMethods(HttpMethod.GET);
return requestMapping;
}
@ServiceActivator(inputChannel = "httpResponse")
@Bean
public HttpRequestExecutingMessageHandler outbound() {
HttpRequestExecutingMessageHandler handler =
new HttpRequestExecutingMessageHandler("http://10.141.201.206:80/foo");
handler.setHttpMethod(HttpMethod.GET);
handler.setExpectedResponseType(String.class);
return handler;
}
@ServiceActivator(inputChannel="httpRequest", outputChannel="httpResponse")
public Object processarMensagem(Object mensagem) {
return mensagem + " - done";
}
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata pollerAmqp(ThreadPoolTaskScheduler taskScheduler) {
final PollerMetadata poller = new PollerMetadata();
poller.setTaskExecutor(taskScheduler);
poller.setReceiveTimeout(-1);
return poller;
}
@Bean
public MessageChannel httpRequest(AmqpTemplate amqpTemplate) {
PollableAmqpChannel channel = new PollableAmqpChannel("httpRequest", amqpTemplate,
DefaultAmqpHeaderMapper.outboundMapper(), DefaultAmqpHeaderMapper.inboundMapper());
channel.setExtractPayload(true);
return channel;
}
@Bean
public MessageChannel httpResponse(AmqpTemplate amqpTemplate) {
PollableAmqpChannel channel = new PollableAmqpChannel("httpResponse", amqpTemplate,
DefaultAmqpHeaderMapper.outboundMapper(), DefaultAmqpHeaderMapper.inboundMapper());
channel.setExtractPayload(true);
return channel;
}
}
but i'm receiving the message: No reply received within timeout
