Introduction

Combining vFabric RabbitMQ and Spring Integration we can create loosely coupled Enterprise class message-based workflows using Spring AMQP. Refer to the Spring Integration, AMQP Backed Message Channels documentation.

In this demo, we will design a Spring Integration workflow, wherein a message is published on to a p2p-pollable-channel and a publish-subscribe-channel is listening to it, it will pickup that message and pass it to 2 different service activators to further process it. I will also demonstrate that, both these message channels are on their own, listening to an external applications and when an external application publishes a message to these queues they will process from there on in the workflow. A typical use case of this is in an enterprise IT environment where we can create robust workflows, wherein if a message is not processed in a particular step, we can message the data and process it from there on. Since Spring AMQP is a wire protocol the external application can be used by any network protocol to publish the message to the vFabric RabbitMQ queue and the Spring Integration flow is triggered.

AMQP Backed Message Channel in Spring Integration flow

As always in my blog, as per TDD, I will be writing the test first as seen below,

public class PublisherSubscriberTest {

@Test
public void testIntegration() {
try {
String request = streamToString(getClass().getResourceAsStream(
"/data/payload.xml"));
Message<String> message = MessageBuilder.withPayload(request)
.build();
channel.send(message);
//assert various messages
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

The Spring Integration Configuration flow:

<int:bridge input-channel="p2p-pollable-channel"
output-channel="pub-sub-channel" />

<int:service-activator input-channel="pub-sub-channel"
id="serviceActivator1" ref="serviceActivator1Bean" method="logXml" />

<int:service-activator input-channel="pub-sub-channel"
id="serviceActivator2" ref="serviceActivator2Bean" method="logXml" />

<bean id="serviceActivator1Bean"
class="com.goSmarter.amqp.service.Subscriber1ServiceActivator">
</bean>

<bean id="serviceActivator2Bean"
class="com.goSmarter.amqp.service.Subscriber2ServiceActivator">
</bean>

Test channel definition:

<int:poller default="true" fixed-rate="1000" />
<int:channel id="p2p-pollable-channel" />
<int:publish-subscribe-channel id="pub-sub-channel" />

Actual channel definition used as a part of web.xml is seen here

<!-- A reference to the org.springframework.amqp.rabbit.connection.ConnectionFactory -->
<rabbit:connection-factory id="connectionFactory" />
<!-- Creates a org.springframework.amqp.rabbit.core.RabbitAdmin to manage
exchanges, queues and bindings -->
<rabbit:admin connection-factory="connectionFactory" />

<int-amqp:channel id="p2p-pollable-channel" connection-factory="connectionFactory" />
<int-amqp:publish-subscribe-channel id="pub-sub-channel" connection-factory="connectionFactory" />

When you run this application in STS as “Run on Server” for the 1st time, you will notice in the vFabric RabbitMQ admin console will have one Queue, p2p-pollable-channel and Exchange called si.fanout.pub-sub-channel. You'll notice the Spring Integration framework appended “si.fanout” in front of pub-sub-channel. If you want to test if the end to end is working, you can post a message on either the Queue or Exchange, it will continue the workflow from that point onwards.

If you are using AMQP backed Channels, you can reply a payload from any channels to continue the workflow, this will be useful in error handling and Payload retry.

Conclusion

In this sample, we have demonstrated that with a single component backed by AMQP, we can build a loosely coupled workflow. Based on the context, there can be 2 different ways of building publish subscriber workflow, one is as described above and another one is one publisher component will have several subscriber components.

I hope this blog helped you.