Spring Integration's MessageHandler interface is implemented by many of the components within the framework. In other words, this is not part of the public API, and a developer would not typically implement MessageHandler directly. Nevertheless, it is used by a Message Consumer for actually handling the consumed Messages, and so being aware of this strategy interface does help in terms of understanding the overall role of a consumer. The interface is defined as follows:

public interface MessageHandler {

    void handleMessage(Message<?> message);

}

Despite its simplicity, this provides the foundation for most of the components that will be covered in the following chapters (Routers, Transformers, Splitters, Aggregators, Service Activators, etc). Those components each perform very different functionality with the Messages they handle, but the requirements for actually receiving a Message are the same, and the choice between polling and event-driven behavior is also the same. Spring Integration provides two endpoint implementations that "host" these callback-based handlers and allow them to be connected to Message Channels.

Because it is the simpler of the two, we will cover the Event Driven Consumer endpoint first. You may recall that the SubscribableChannel interface provides a subscribe() method and that the method accepts a MessageHandler parameter (as shown in the section called “SubscribableChannel”):

subscribableChannel.subscribe(messageHandler);

Since a handler that is subscribed to a channel does not have to actively poll that channel, this is an Event Driven Consumer, and the implementation provided by Spring Integration accepts a a SubscribableChannel and a MessageHandler:

SubscribableChannel channel = (SubscribableChannel) context.getBean("subscribableChannel");

EventDrivenConsumer consumer = new EventDrivenConsumer(channel, exampleHandler);

Spring Integration also provides a PollingConsumer, and it can be instantiated in the same way except that the channel must implement PollableChannel:

PollableChannel channel = (PollableChannel) context.getBean("pollableChannel");

PollingConsumer consumer = new PollingConsumer(channel, exampleHandler);

There are many other configuration options for the Polling Consumer. For example, the trigger is a required property:

PollingConsumer consumer = new PollingConsumer(channel, handler);

consumer.setTrigger(new IntervalTrigger(30, TimeUnit.SECONDS));

Spring Integration currently provides two implementations of the Trigger interface: IntervalTrigger and CronTrigger. The IntervalTrigger is typically defined with a simple interval (in milliseconds), but also supports an 'initialDelay' property and a boolean 'fixedRate' property (the default is false, i.e. fixed delay):

IntervalTrigger trigger = new IntervalTrigger(1000);
trigger.setInitialDelay(5000);
trigger.setFixedRate(true);

The CronTrigger simply requires a valid cron expression (see the Javadoc for details):

CronTrigger trigger = new CronTrigger("*/10 * * * * MON-FRI");

In addition to the trigger, several other polling-related configuration properties may be specified:

PollingConsumer consumer = new PollingConsumer(channel, handler);

consumer.setMaxMessagesPerPoll(10);

consumer.setReceiveTimeout(5000);

The 'maxMessagesPerPoll' property specifies the maximum number of messages to receive within a given poll operation. This means that the poller will continue calling receive() without waiting until either null is returned or that max is reached. For example, if a poller has a 10 second interval trigger and a 'maxMessagesPerPoll' setting of 25, and it is polling a channel that has 100 messages in its queue, all 100 messages can be retrieved within 40 seconds. It grabs 25, waits 10 seconds, grabs the next 25, and so on.

The 'receiveTimeout' property specifies the amount of time the poller should wait if no messages are available when it invokes the receive operation. For example, consider two options that seem similar on the surface but are actually quite different: the first has an interval trigger of 5 seconds and a receive timeout of 50 milliseconds while the second has an interval trigger of 50 milliseconds and a receive timeout of 5 seconds. The first one may receive a message up to 4950 milliseconds later than it arrived on the channel (if that message arrived immediately after one of its poll calls returned). On the other hand, the second configuration will never miss a message by more than 50 milliseconds. The difference is that the second option requires a thread to wait, but as a result it is able to respond much more quickly to arriving messages. This technique, known as "long polling", can be used to emulate event-driven behavior on a polled source.

A Polling Consumer may also delegate to a Spring TaskExecutor, and it can be configured to participate in Spring-managed transactions. The following example shows the configuration of both:

PollingConsumer consumer = new PollingConsumer(channel, handler);

TaskExecutor taskExecutor = (TaskExecutor) context.getBean("exampleExecutor");
consumer.setTaskExecutor(taskExecutor);

PlatformTransactionManager txManager = (PlatformTransationManager) context.getBean("exampleTxManager");
consumer.setTransactionManager(txManager);

The examples above show dependency lookups, but keep in mind that these consumers will most often be configured as Spring bean definitions. In fact, Spring Integration also provides a FactoryBean that creates the appropriate consumer type based on the type of channel, and there is full XML namespace support to even further hide those details. The namespace-based configuration will be featured as each component type is introduced.

Throughout the reference manual, you will see specific configuration examples for endpoint elements, such as router, transformer, service-activator, and so on. Most of these will support an "input-channel" attribute and many will support an "output-channel" attribute. After being parsed, these endpoint elements produce an instance of either the PollingConsumer or the EventDrivenConsumer depending on the type of the "input-channel" that is referenced: PollableChannel or SubscribableChannel respectively. When the channel is pollable, then the polling behavior is determined based on the endpoint element's "poller" sub-element and its attributes. For example, a simple interval-based poller with a 1-second interval would be configured like this:

 <transformer input-channel="pollable"
              ref="transformer"
              output-channel="output">
     <poller fixed-rate="1000"/>
</transformer>

As an alternative to 'fixed-rate' you cna also use 'fixed-delay' attribute.

For a poller based on a Cron expression, use the "cron" attribute instead:

 <transformer input-channel="pollable"
              ref="transformer"
              output-channel="output">
     <poller cron="*/10 * * * * MON-FRI"/>
 </transformer>

If the input channel is a PollableChannel, then the poller configuration is required. Specifically, as mentioned above, the 'trigger' is a required property of the PollingConsumer class. Therefore, if you omit the "poller" sub-element for a Polling Consumer endpoint's configuration, an Exception may be thrown. The exception will also be thrown if you attempt to configure a poller on the element that is connected to a non-pollable channel.

It is also possible to create top-level pollers in which case only a "ref" is required:

 <poller id="weekdayPoller" cron="*/10 * * * * MON-FRI"/>
   
 <transformer input-channel="pollable"
              ref="transformer"
              output-channel="output">
    <poller ref="weekdayPoller"/>
 </transformer>

In fact, to simplify the configuration, you can define a global default poller. A single top-level poller within an ApplicationContext may have the default attribute with a value of "true". In that case, any endpoint with a PollableChannel for its input-channel that is defined within the same ApplicationContext and has no explicitly configured 'poller' sub-element will use that default.

 <poller id="defaultPoller" default="true" max-messages-per-poll="5" fixed-rate="3000"/>
    
 <!-- No <poller/> sub-element is necessary since there is a default -->
 <transformer input-channel="pollable"
              ref="transformer"
              output-channel="output"/>

Spring Integration also provides transaction support for the pollers so that each receive-and-forward operation can be performed as an atomic unit-of-work. To configure transactions for a poller, simply add the <transactional/> sub-element. The attributes for this element should be familiar to anyone who has experience with Spring's Transaction management:

<poller fixed-delay="1000">
    <transactional transaction-manager="txManager"
                   propagation="REQUIRED"
                   isolation="REPEATABLE_READ"
                   timeout="10000"
                   read-only="false"/>
</poller>

AOP Advice chains

Since Spring transaction support depends on the Proxy mechanism  with TransactionInterceptor (AOP Advice) handling transactional behavior of the message flow initiated by the poler, some times there is a need to provide extra Advice(s) to handle other cross cutting behavior associated with the poller. For that poller defines an 'advice-chain' element allowing you to add more advices - class that  implements MethodInterceptor interface.. 

<service-activator id="advicedSa" input-channel="goodInputWithAdvice" ref="testBean"
		method="good" output-channel="output">
	<poller max-messages-per-poll="1" fixed-rate="10000">
		<transactional transaction-manager="txManager" />
		 <advice-chain>
			<ref bean="adviceA" />
			<beans:bean class="org.bar.SampleAdvice"/>
		</advice-chain>
	</poller>
</service-activator>

For more information on how to implement MethodInterceptor please refer to AOP sections of Spring reference manual (section 7 and 8). Advice chain can also be applied on the poller that does not have any transaction configuration essentially allowing you to enhance the behavior of the message flow initiated by the poller.

The polling threads may be executed by any instance of Spring's TaskExecutor abstraction. This enables concurrency for an endpoint or group of endpoints. As of Spring 3.0, there is a "task" namespace in the core Spring Framework, and its <executor/> element supports the creation of a simple thread pool executor. That element accepts attributes for common concurrency settings such as pool-size and queue-capacity. Configuring a thread-pooling executor can make a substantial difference in how the endpoint performs under load. These settings are available per-endpoint since the performance of an endpoint is one of the major factors to consider (the other major factor being the expected volume on the channel to which the endpoint subscribes). To enable concurrency for a polling endpoint that is configured with the XML namespace support, provide the 'task-executor' reference on its <poller/> element and then provide one or more of the properties shown below:

 <poller task-executor="pool" fixed-rate="1000"/>
    
 <task:executor id="pool"
                pool-size="5-25"
                queue-capacity="20"
                keep-alive="120"/>

If no 'task-executor' is provided, the consumer's handler will be invoked in the caller's thread. Note that the "caller" is usually the default TaskScheduler (see Section B.3, “Configuring the Task Scheduler”). Also, keep in mind that the 'task-executor' attribute can provide a reference to any implementation of Spring's TaskExecutor interface by specifying the bean name. The "executor" element above is simply provided for convenience.

As mentioned in the background section for Polling Consumers above, you can also configure a Polling Consumer in such a way as to emulate event-driven behavior. With a long receive-timeout and a short interval-trigger, you can ensure a very timely reaction to arriving messages even on a polled message source. Note that this will only apply to sources that have a blocking wait call with a timeout. For example, the File poller does not block, each receive() call returns immediately and either contains new files or not. Therefore, even if a poller contains a long receive-timeout, that value would never be usable in such a scenario. On the other hand when using Spring Integration's own queue-based channels, the timeout value does have a chance to participate. The following example demonstrates how a Polling Consumer will receive Messages nearly instantaneously.

 <service-activator input-channel="someQueueChannel"
        output-channel="output">  
     <poller receive-timeout="30000" fixed-rate="10"/>
        
 </service-activator>

Using this approach does not carry much overhead since internally it is nothing more then a timed-wait thread which does not require nearly as much CPU resource usage as a thrashing, infinite while loop for example.

Throughout the reference manual, you will also see specific configuration and implementation examples of various endpoints which can accept a Message or any arbitrary Object as an input parameter. In the case of an Object, such parameter will be mapped to a Message payload or part of the payload or header (when using Spring Expression Language). However there are times when the type of input parameter of the endpoint method does not match the type of the payload or its part. In this scenario we need to perform type conversion. Spring Integration provides a convenient way for registering type converters (using Spring 3.x ConversionService) within its own instance of the conversion service bean named integrationConversionService which is automatically created as soon as the first converter is defined. To register such converter all you need is to implement  org.springframework.core.convert.converter.Converter and register via cionvinient namespace support:

 <int:converter ref="sampleConverter"/>

	<bean id="sampleConverter" class="foo.bar.TestConverter"/>

or

 <int:converter>
		<bean class="org.springframework.integration.config.xml.ConverterParserTests$TestConverter3"/>
	</int:converter>

If you want the polling to be asynchronous, Poller can optionaly specify 'task-executor' attribute pointing to an existing instance of TaskExecutor bean (Spring 3.0 provides a convinient namespaces configuration via the task namespace). However, there are certain things you must understand when configuring Poller with TaskExecutor. 

The problem is that there are two configurations in place. The Poller and the TaskExecutor and they both have to be in tune with each other otherwise you might end up creating an artificial memory leak. 

Let's look at the following configuration provided by one of the users on the Spring's forums (http://forum.springsource.org/showthread.php?t=94519):

<int:service-activator input-channel="publishChannel" ref="myService">
		<int:poller receive-timeout="5000" task-executor="taskExecutor" fixed-rate="50"/>
</si:service-activator>
	
<task:executor id="taskExecutor" pool-size="20" queue-capacity="20"/>

The above configuration demonstrates one of those out of tune configurations.

The poller keeps scheduling new tasks even though all the threads are blocked waiting for either a new message to arrive, or the timeout to expire. Given that there are 20 threads executing tasks with a 5 second timeout, they will be executed at a rate of 4 per second (5000/20 = 250ms). But, new tasks are being scheduled at a rate of 20 per second, so the internal queue in the task executor will grow at a rate of 16 per second (while the process is idle), so we essentially have a memory leak.

One of the ways to handle this is to set queue-capacity attribute of Task Executor to 0. You can also manage it by specifying what to do with messages that can not be queued up by setting rejection-policy attribute of Task Executor (e.g., DISCARD). In other words there are certain details you must understand with regard to configuring the TaskExecutor. Please refer to - Section 25 - Task Execution and Scheduling of Spring reference manual.

Working with Objects instead of Messages is an improvement. However, it would be even better to have no dependency on the Spring Integration API at all - including the gateway class. For that reason, Spring Integration also provides a GatewayProxyFactoryBean that generates a proxy for any interface and internally invokes the gateway methods shown above. Namespace support is also provided as demonstrated by the following example.

<gateway id="fooService"
         service-interface="org.example.FooService"
         default-request-channel="requestChannel"
         default-reply-channel="replyChannel"/>

Then, the "fooService" can be injected into other beans, and the code that invokes the methods on that proxied instance of the FooService interface has no awareness of the Spring Integration API. The general approach is similar to that of Spring Remoting (RMI, HttpInvoker, etc.). See the "Samples" Appendix for an example that uses this "gateway" element (in the Cafe demo).

The reason that the attributes on the 'gateway' element are named 'default-request-channel' and 'default-reply-channel' is that you may also provide per-method channel references by using the @Gateway annotation.

 public interface Cafe {

     @Gateway(requestChannel="orders")
     void placeOrder(Order order);

 }

... as well as method sub element if yuo prefer XML configuration (see next paragraph)

It is also possible to pass values to be interpreted as Message headers on the Message that is created and sent to the request channel by using the @Header annotation:

 public interface FileWriter {

     @Gateway(requestChannel="filesOut")
     void write(byte[] content, @Header(FileHeaders.FILENAME) String filename);

 }

If you prefer XML way of configuring Gateway methods, you can provide method sub-elements to the gateway configuration (see below)

<si:gateway id="myGateway" service-interface="org.foo.bar.TestGateway"
      default-request-channel="inputC">
  <si:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/>
      <si:method name="echoUpperCase" request-channel="inputB"/>
      <si:method name="echoViaDefault"/>
</si:gateway>

You can also provide individual headers per method invocation via XML. This could be very useful if the headers you want to set are static in nature and you don't want to embed them in the gateway's method signature via @Header annotations. For example, in the Loan Broker example we want to influence how aggregation of the Loan quotes will be done based on what type of request was initiated (single quote or all quotes). Determining the type of the request by evaluating what gateway method was invoked, although possible would violate the separation of concerns paradigm (method is a java artifact),  but expressing your intention (meta information) via Message headers is natural in a Messaging architecture.

<int:gateway id="loanBrokerGateway"
         service-interface="org.springframework.integration.loanbroker.LoanBrokerGateway">
  <int:method name="getLoanQuote" request-channel="loanBrokerPreProcessingChannel">
    <int:header name="RESPONSE_TYPE" value="BEST"/>
  </int:method>
  <int:method name="getAllLoanQuotes" request-channel="loanBrokerPreProcessingChannel">
    <int:header name="RESPONSE_TYPE" value="ALL"/>
  </int:method>
</int:gateway>

In the above case you can clearly see how a different header value will be set for the 'RESPONSE_TYPE' header based on the gateway's method.

As with anything else, Gateway invocation might result in errors. By default any error that has occurred downstream will be re-thrown as a MessagingExeption (RuntimeException) upon the Gateway's method invocation. However there are times when you may want to treat an Exception as a valid reply, by mapping it to a Message. To accomplish this our Gateway provides support for Exception mappers via the exception-mapper attribute.

<si:gateway id="sampleGateway"
    default-request-channel="gatewayChannel"
    service-interface="foo.bar.SimpleGateway"
    exception-mapper="exceptionMapper"/>

<bean id="exceptionMapper" class="foo.bar.SampleExceptionMapper"/>

        

foo.bar.SampleExceptionMapper is the implementation of org.springframework.integration.message.InboundMessageMapper which only defines one method: toMessage(Object object).

public static class SampleExceptionMapper implements InboundMessageMapper<Throwable>{
  public Message<?> toMessage(Throwable object) throws Exception {
    MessageHandlingException ex = (MessageHandlingException) object;
    return MessageBuilder.withPayload("Error happened in message: " +
                ex.getFailedMessage().getPayload()).build();
  }

}
        

As a pattern the Messaging Gateway is a very nice way to hide messaging-specific code while still exposing the full capabilities of the messaging system. And GatewayProxyFactoryBean provides a convenient way to expose a Proxy over a service-interface thus giving you a POJO-based access to a messaging system (based on objects in your own domain, or primitives/Strings, etc).  But when a gateway is exposed via simple POJO methods which return values it does imply that for each Request message (generated when the method is invoked) there must be a Reply message (generated when the method has returned). Since Messaging systems naturally are asynchronous you may not always be able to guarantee the contract where "for each request there will always be be a reply".  With Spring Integration 2.0 we are introducing support for an Asynchronous Gateway which is a convenient way to initiate flows where you may not know if a reply is expected or how long will it take for it to arrive.

A natural way to handle these types of scenarios in Java would be relying upon java.util.concurrent.Future instances, and that is exactly what Spring Integration uses to support an Asynchronous Gateway.

From the XML configuration, there is nothing different and you still define Asynchronous Gateway the same way as a regular Gateway.

<int:gateway id="mathService" 
     service-interface="org.springframework.integration.sample.gateway.futures.MathServiceGateway"
     default-request-channel="requestChannel"/>

However the Gateway Interface (service-interface) is a bit different.

public interface MathServiceGateway {
  Future<Integer> multiplyByTwo(int i);
}

As you can see from the example above the return type for the gateway method is Future. When GatewayProxyFactoryBean sees that the return type of the gateway method is Future, it immediately switches to the async mode by utilizing an AsyncTaskExecutor. That is all. The call to a method always returns immediately with Future encapsulating  the interaction with the framework. Now you can interact with the Future at your own pace to get the result, timeout, get the exception etc...

MathServiceGateway mathService = ac.getBean("mathService", MathServiceGateway.class);
Future<Integer> result = mathService.multiplyByTwo(number);
// do something else here since the reply might take a moment
int finalResult =  result.get(1000, TimeUnit.SECONDS);

For a more detailed example, please refer to the async-gateway sample distributed within the Spring Integration samples.

As it was explained earlier, Gateway provides a convenient way of interacting with Messaging system via POJO method invocations, but realizing that a typical method invocation, which is generally expected to always return (even with Exception), might not always map one-to-one to message exchanges (e.g., reply message might not be coming which is equivalent to method not returning), it is important to go over several scenarios especially in the Sync Gateway case and understand what the default behavior of the Gateway and how to deal with these scenarios to make Sync Gateway behavior more predictable regardless of the outcome of the message flow that was initialed from such Gateway.

There are certain attributes that could be configured to make Sync Gateway behavior more predictable, but some of them might not always work as you might have expected. One of them is reply-timeout. So, lets look at the reply-timeout attribute and see how it can/can't influence the behavior of the Sync Gateway in various scenarios. We will look at single-theraded scenario (all components downstream are connected via Direct Channel) and multi-theraded scenarios (e.g., somewhere downstream you may have Pollable or Executor Channel which breaks single-thread boundary)

Long running process downstream

Sync Gateway - single-threaded. If a component downstream is still running (e.g., infinite loop or a very slow service), then setting reply-timeout has no effect and Gateway method call will not return until such downstream service exits (e.g., return or exception). Sync Gateway - multi-threaded. If a component downstream is still running (e.g., infinite loop or a very slow service), in a multi-threaded message flow setting reply-timeout will have an effect by allowing gateway method invocation to return once the timeout has been reached, since GatewayProxyFactoryBean  will simply poll on the reply channel waiting for a message untill the timeout expires. However it could result in the 'null' return from the Gateway method if the timeout has been reached before the actual reply was produced. It is also important to understand that the reply message (if produced) will be sent to a reply channel after Gateway method invocation might have returned, so you must be aware of that and design your flow with this in mind.

Downstream component returns 'null'

Sync Gateway - single-threaded. If a component downstream returns 'null' and no reply-timeout has been configured, the Gateway method call will hang indefinitely unless: a) reply-timeout has been configured or b) requires-reply attribute has been set on the downstream component (e.g., service-activator) that might return 'null'. In this case, the exception will be thrown and propagated to the Gateway. Sync Gateway - multi-threaded. Behavior is the same as above.

Downstream component return signature is 'void' while Gateway method signature is non-void

Sync Gateway - single-threaded. If a component downstream returns 'void' and no reply-timeout has been configured, the Gateway method call will hang indefinitely unless reply-timeout has been configured  Sync Gateway - multi-threaded Behavior is the same as above.

Downstream component results in Runtime Exception (regardless of the method signature)

Sync Gateway - single-threaded. If a component downstream throws a Runtime Exception, such exception will be propagated via Error Message back to the gateway and re-thrown. Sync Gateway - multi-threaded Behavior is the same as above.

The Service Activator is the endpoint type for connecting any Spring-managed Object to an input channel so that it may play the role of a service. If the service produces output, it may also be connected to an output channel. Alternatively, an output producing service may be located at the end of a processing pipeline or message flow in which case, the inbound Message's "replyChannel" header can be used. This is the default behavior if no output channel is defined, and as with most of the configuration options you'll see here, the same behavior actually applies for most of the other components we have seen.

To create a Service Activator, use the 'service-activator' element with the 'input-channel' and 'ref' attributes:

<service-activator input-channel="exampleChannel" ref="exampleHandler"/>

The configuration above assumes that "exampleHandler" either contains a single method annotated with the @ServiceActivator annotation or that it contains only one public method at all. To delegate to an explicitly defined method of any object, simply add the "method" attribute.

<service-activator input-channel="exampleChannel" ref="somePojo" method="someMethod"/>

In either case, when the service method returns a non-null value, the endpoint will attempt to send the reply message to an appropriate reply channel. To determine the reply channel, it will first check if an "output-channel" was provided in the endpoint configuration:

<service-activator input-channel="exampleChannel" output-channel="replyChannel"
                   ref="somePojo" method="someMethod"/>

If no "output-channel" is available, it will then check the Message's REPLY_CHANNEL header value. If that value is available, it will then check its type. If it is a MessageChannel, the reply message will be sent to that channel. If it is a String, then the endpoint will attempt to resolve the channel name to a channel instance. If the channel cannot be resolved, then a ChannelResolutionException will be thrown.

The argument in the service method could be either a Message or an arbitrary type. If the latter, then it will be assumed that it is a Message payload, which will be extracted from the message and injected into such service method. This is generally the recommended approach as it follows and promotes a POJO model when working with Spring Integration. Arguments may also have @Header, @Headers annotations as described in Section B.5, “Annotation Support”

Using a "ref" attribute is generally recommended if the custom Service Activator handler implementation can be reused in other <service-activator> definitions. However if the custom Service Activator handler implementation should be scoped to a single definition of the <service-activator>, you can use an inner bean definition:

<service-activator id="exampleServiceActivator" input-channel="inChannel"
            output-channel = "outChannel" method="foo">
    <beans:bean class="org.foo.ExampleServiceActivator"/>
</service-activator>

Service Activators and Spring Expression Language (SpEL)

Since Spring Integration 2.0 Service Activators can also benefit from SpEL ()http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/expressions.html in several ways.

First, you may now invoke any bean method without pointing to this bean via ref attribute or including it as an inner definition. For example:

<int:service-activator input-channel="in" output-channel="out" 
	expression="@accountService.processAccount(payload)"/>
	
	<bean id="accountService" class="foo.bar.Account"/>

In the above configuration instead of injecting 'accountService' using ref or as inner bean we are simply using @beanId notation and invoking a method which takes the type compatible with Message payload. For simple scenarios your Service Activators do not even have to reference a bean if all logic can be encapsulated by such expression.

<int:service-activator input-channel="in" output-channel="out" expression="payload * 2"/>

In the above configuration our service logic is to simply multiply the payload value by 2 and SpEL lets us handle it relatively easy.

A Delayer is a simple endpoint that allows a Message flow to be delayed by a certain interval. When a Message is delayed, the original sender will not block. Instead, the delayed Messages will be scheduled with an instance of java.util.concurrent.ScheduledExecutorService to be sent to the output channel after the delay has passed. This approach is scalable even for rather long delays, since it does not result in a large number of blocked sender Threads. On the contrary, in the typical case a thread pool will be used for the actual execution of releasing the Messages. Below you will find several examples of configuring a Delayer.

The <delayer> element is used to delay the Message flow between two Message Channels. As with the other endpoints, you can provide the "input-channel" and "output-channel" attributes, but the delayer also requires at least the 'default-delay' attribute with the number of milliseconds that each Message should be delayed.

 <delayer input-channel="input" default-delay="3000" output-channel="output"/>

If you need per-Message determination of the delay, then you can also provide the name of a header within the 'delay-header-name' attribute:

 <delayer input-channel="input" output-channel="output"
          default-delay="3000" delay-header-name="delay"/>

In the example above the 3 second delay would only apply in the case that the header value is not present for a given inbound Message. If you only want to apply a delay to Messages that have an explicit header value, then you can set the 'default-delay' to 0. For any Message that has a delay of 0 (or less), the Message will be sent directly. In fact, if there is not a positive delay value for a Message, it will be sent to the output channel on the calling Thread.

The delayer delegates to an instance of Spring's TaskScheduler abstraction. The default scheduler is a ThreadPoolTaskScheduler instance with a pool size of 1. If you want to delegate to a different scheduler, you can provide a reference through the delayer element's 'scheduler' attribute:

 <delayer input-channel="input" output-channel="output"
          default-delay="0" delay-header-name="delay"
          scheduler="exampleTaskScheduler"/>

 <task:scheduler id="exampleTaskScheduler" pool-size="3"/>

Depending on the complexity of your integration requirements Groovy scripts could be provided inline as CDATA in XML configuration or as a reference to a file containing Groovy script. To enable Groovy support Spring Integration defines GroovyScriptExecutingMessageProcessor which will create a groovy Binding object identifying Message Payload as payload variable and Message Headers as headers variable. All that is left for you to do is write script that uses these variables. Below are couple of sample configurations:

Filter

<filter input-channel="referencedScriptInput">
   <groovy:script location="some/path/to/groovy/file/GroovyFilterTests.groovy"/>
</filter>

<filter input-channel="inlineScriptInput">
     <groovy:script><![CDATA[
     return payload == 'good'
   ]]></groovy:script>
</filter>

You see that script could be included inline or via location attribute using the groovy namespace sport. 

Other supported elements are router, service-activator, transformer, splitter

Another interesting aspect of using Groovy support is framework's ability to update (reload) scripts  without restarting the Application Context. To accomplish this all you need is specify refresh-check-delay attribute on script element. The reason for this attribute is to make reloading of the script more efficient. 

<groovy:script location="..." refresh-check-delay="5000"/>

In the above example for the next 5 seconds after you update the script you'll still be using the old script and after 5 seconds the context will be updated with the new script. This is a good example where  'near real time' is acceptable.

<groovy:script location="..." refresh-check-delay="0"/>

In the above example the context will be updated with the new script every time the script is modified. Basically this is the example of the 'real-time' and might not be the most efficient way.

<groovy:script location="..." refresh-check-delay="-1"/>

Any negative number value means the script will never be refreshed after initial initialization of application context. DEFAULT BEHAVIOR

As described in (EIP), the idea behind the Control Bus is that the same messaging system can be used for monitoring and managing the components within the framework as is used for "application-level" messaging. In Spring Integration we build upon the adapters described above so that it's possible to send Messages as a means of invoking exposed operations.

 <groovy:control-bus input-channel="operationChannel"/>

The Control Bus has an input channel that can be accessed for invoking operations on the beans in the application context.

The groovy control bus executes messages on the input channel as Groovy scripts. It takes a message, compiles the body to a Script, customizes it with a GroovyObjectCustomizer, and then executes it. The default customizer just exposes all the beans in the application context as script context objects.