Spring Integration Managing Lifecycle using ControlBus

In this article we will see an example of Control Bus. But what is a control bus?

We use Control Bus to administration and control of an enterprise system. The Control Bus uses the same messaging mechanism used by the application data, but uses separate channels to transmit data that is relevant to the management of components involved in the message flow.

In this example, we will invoke a management operation on an endpoint by sending control messages to a message channel dedicated to play the role of an operation channel. The message sent to a Control Bus will be treated as a command to be executed.

Manage lifecycle of an adapter using ControlBus

The control bus is a Spring Integration component that accepts messages on its input channel just like any other endpoint like service activator, jms, or other type of message endpoint.
The message received by a Control Bus will be treated as a command to be executed. In the below example, we will see how we can start a inbound-channel-adapter through a Control Bus message.

Control Bus Manages Adapter Lifecycle

Few points to note about:

By default, we have set auto-startup to false.
control-bus is the tag that starts the control bus. Its input channel is set using attribute input-channel.
Once the inbound-channel-adapter starts it will post a message to the adapterOutputChannel

applicationContext.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:beans="http://www.springframework.org/schema/beans" xmlns:groovy="http://www.springframework.org/schema/integration/groovy"
	xsi:schemaLocation="http://www.springframework.org/schema/integration
		http://www.springframework.org/schema/integration/spring-integration.xsd
		http://www.springframework.org/schema/beans
		http://www.springframework.org/schema/beans/spring-beans.xsd">


	<control-bus input-channel="controlBusChannel" auto-startup="true"/>

	<inbound-channel-adapter id="someAdapter" channel="adapterOutputChannel" auto-startup="false" ref="someService" method="process">
		<poller fixed-rate="1000"/>
	</inbound-channel-adapter>
	
	<beans:bean id="someService" class="com.javarticles.spring.integration.controlbus.SomeService"/>	
	
	<channel id="adapterOutputChannel">
		<queue/>
	</channel>

</beans:beans>

Once the inbound-channel-adapter is started, it will invoke SomeService.process which in turn will return value to be posted to the outbound channel adapterOutputChannel.

SomeService:

package com.javarticles.spring.integration.controlbus;

public class SomeService {
    public String process() {
        return "Processed";
    }
}

We will post a message @someAdapter.start() to controlBusChannel. The message will indicate the Control Bus to start the inbound-channel-adapter service identified by id someAdapter and action start().

SpringIntegrationControlBusManageAdpterLifecycleExample:

package com.javarticles.spring.integration.controlbus;

import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;

public class SpringIntegrationControlBusManageAdpterLifecycleExample {

    public static void main(String[] args) throws InterruptedException {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
                "applicationContext.xml");
        try {
            PollableChannel adapterOutputChanel = context.getBean("adapterOutputChannel", PollableChannel.class);
            System.out.println("Received before adapter started: " + adapterOutputChanel.receive(1000));
            MessageChannel controlBusChannel = (MessageChannel) context
                    .getBean("controlBusChannel");
            controlBusChannel.send(MessageBuilder.withPayload(
                    "@someAdapter.start()").build());            
            System.out.println("Received after adapter started: " + adapterOutputChanel.receive(1000));
        } finally {
            context.close();
        }
    }
}

After the adapter is started, we are able to receive the message posted by the adapter.

Output:

INFO: Adding {service-activator} as a subscriber to the 'controlBusChannel' channel
May 02, 2015 9:57:07 AM org.springframework.integration.channel.DirectChannel adjustCounterIfNecessary
INFO: Channel 'org[email protected]246b179d.controlBusChannel' has 1 subscriber(s).
May 02, 2015 9:57:07 AM org.springframework.integration.endpoint.EventDrivenConsumer start
INFO: started org.springframework.integration.config.ConsumerEndpointFactoryBean#0
May 02, 2015 9:57:07 AM org.springframework.integration.endpoint.EventDrivenConsumer logComponentSubscriptionEvent
INFO: Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
May 02, 2015 9:57:07 AM org.springframework.integration.channel.PublishSubscribeChannel adjustCounterIfNecessary
INFO: Channel 'org.springframework.context.support.ClassPathXmlAp[email protected]' has 1 subscriber(s).
May 02, 2015 9:57:07 AM org.springframework.integration.endpoint.EventDrivenConsumer start
INFO: started _org.springframework.integration.errorLogger
Received before adapter started: null
May 02, 2015 9:57:08 AM org.springframework.integration.endpoint.SourcePollingChannelAdapter start
INFO: started someAdapter
Received after adapter started: GenericMessage [payload=Processed, headers={id=9daf08de-744f-47e6-e829-79e2adf0ddca, timestamp=1430540828276}]

Download the source code

This was an example about managing lifecycle using ControlBus. You can download the source code here: springintegrationControlBus.zip