For people in hurry, refer the steps and the demo.

Introduction

Control Bus Pattern is a Enterprise Integration Pattern is used to control distributed systems in Spring Integration. In this blog, I will show you how a Control Bus can control your application or a component to start or stop listening to JMS message. In this example, we are using JMS queue to start and stop the jms inbound-channel-adapter, we can also do this with jdbc inbound-channel-adapter and control this thru an external application. The other way to do the same is by using MBean as in this example.

In this use case, there is a Spring Integration flow. This Spring Integration flow can be controlled by sending start / stop message to inbound-channel-adapter from a ActiveMQ JMS queue.

Details Control Bus with Spring Integration

Control Bus Spring Integration JMS

To start implementing this use case, we write the JUnit test 1st. If you notice once the inboundAdapter is started the message is received from the adapterOutchannel. Once the inboundAdapter is stopped no message is received. This is demonstrated as below,

@Test
public void demoControlBus() {
assertNull(adapterOutputChanel.receive(1000));
controlChannel.send(new GenericMessage<String>("@inboundAdapter.start()"));
assertNotNull(adapterOutputChanel.receive(1000));
controlChannel.send(new GenericMessage<String>("@inboundAdapter.stop()"));
assertNull(adapterOutputChanel.receive(1000));
}

The test configuration looks as below,

<int:inbound-channel-adapter id="inboundAdapter"
channel="controlbus-managed-p2p-pollable-channel" expression="'Hello'"
auto-startup="false">
<int:poller fixed-rate="6000" />
</int:inbound-channel-adapter>

If you run the “mvn test” the tests work. In the main configuration, we will be configuring actual queues and jms inbound-channel-adapter as below,

<int-jms:inbound-channel-adapter id="inboundAdapter"
channel="controlbus-managed-p2p-pollable-channel"
jms-template="jmsTemplate">
<int:poller fixed-rate="6000" />
</int-jms:inbound-channel-adapter>

<int-jms:inbound-channel-adapter id="controlBusAdapter"
channel="control-channel"
jms-template="controlBusJmsTemplate">
<int:poller fixed-rate="6000" />
</int-jms:inbound-channel-adapter>

Now when you start the component as “Run on Server” in STS IDE and post a message on
MyQueue, you can see the subscribers received the messages on the console. You can issue “@inboundAdapter.stop()” on the ControlBusQueue, it will stop the inbound-channel-adapter, it will also throw java.lang.InterruptedException, it looks like a false alarm. To test if the inbound-channel-adapter is stopped, post a message on to MyQueue, the component will not process the message. Now issue “@inboundAdapter.start()” on the ControlBusQueue, it will process the earlier message and start listening for new messages.

Conclusion

If you notice in this blog, we can control the component to listen to message using Control Bus.  The other way to do the same is by using MBean as in this example.