springspring-integrationspring-integration-dslspring-integration-amqp

Spring Integration AMQP


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?


Solution

  • 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();