I have just stared learning spring-integration I want to receive a message on a queue and perform 2 steps in parallel: Step 1 -> Process it using a bean Step 2 -> Transform and send it to another queue. Something like :
return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "queue1")
.configureContainer(simpleMessageListenerContainerSpec -> {
simpleMessageListenerContainerSpec.concurrentConsumers(3);
}))
.log(message -> "SERVICE EVENT QUEUE : Received Message : " + message.getPayload())
.handle(serviceBean, "process")
.<String,String>transform(String::toLowerCase)
.log(message -> "SERVICE EVENT QUEUE : Transformed Message : " + message.getPayload())
.handle(
Amqp.outboundAdapter(rabbitTemplate)
.exchangeName("exchange")
.routingKey("queue2.routing"))
.get();
What am I missing? The action after the first handle is not getting executed. I think I am not understanding this part correctly. Also how can I do these 2 steps in parallel?
You should start from the theory to understand many concepts and components in Spring Integration.
"2 steps in parallel" - is exactly a publish-subscribe pattern: https://www.enterpriseintegrationpatterns.com/patterns/messaging/PublishSubscribeChannel.html and Spring Integration provides an implementation for it: https://docs.spring.io/spring-integration/docs/current/reference/html/core.html#channel-implementations-publishsubscribechannel. As you see according the docs to achieve a parallelism you need to configure such a channel with a TaskExecutor
.
With Java DSL we provide a high-level API for publish-subscribe configuration:
https://docs.spring.io/spring-integration/docs/current/reference/html/dsl.html#java-dsl-subflows
So, to make your .handle(serviceBean, "process")
and .<String,String>transform(String::toLowerCase)
as parallel you need to have something like this:
return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "queue1")
.configureContainer(simpleMessageListenerContainerSpec -> {
simpleMessageListenerContainerSpec.concurrentConsumers(3);
}))
.log(message -> "SERVICE EVENT QUEUE : Received Message : " + message.getPayload())
.publishSubscribeChannel(Executors.newCachedThreadPool(), s -> s
.subscribe(f -> f
.handle(serviceBean, "process")))
.<String,String>transform(String::toLowerCase)
.log(message -> "SERVICE EVENT QUEUE : Transformed Message : " + message.getPayload())
.handle(
Amqp.outboundAdapter(rabbitTemplate)
.exchangeName("exchange")
.routingKey("queue2.routing"))
.get();