Spring Integration Reference Manual

One of the key themes of the Spring Framework is inversion of control. In its broadest sense, this means that the framework handles responsibilities on behalf of the components that are managed within its context. The components themselves are simplified since they are relieved of those responsibilities. For example, dependency injection relieves the components of the responsibility of locating or creating their dependencies. Likewise, aspect-oriented programming relieves business components of generic cross-cutting concerns by modularizing them into reusable aspects. In each case, the end result is a system that is easier to test, understand, maintain, and extend.

Furthermore, the Spring framework and portfolio provide a comprehensive programming model for building enterprise applications. Developers benefit from the consistency of this model and especially the fact that it is based upon well-established best practices such as programming to interfaces and favoring composition over inheritance. Spring's simplified abstractions and powerful support libraries boost developer productivity while simultaneously increasing the level of testability and portability.

Spring Integration is a new member of the Spring portfolio motivated by these same goals and principles. It extends the Spring programming model into the messaging domain and builds upon Spring's existing enterprise integration support to provide an even higher level of abstraction. It supports message-driven architectures where inversion of control applies to runtime concerns, such as when certain business logic should execute and where the response should be sent. It supports routing and transformation of messages so that different transports and different data formats can be integrated without impacting testability. In other words, the messaging and integration concerns are handled by the framework, so business components are further isolated from the infrastructure and developers are relieved of complex integration responsibilities.

As an extension of the Spring programming model, Spring Integration provides a wide variety of configuration options including annotations, XML with namespace support, XML with generic "bean" elements, and of course direct usage of the underlying API. That API is based upon well-defined strategy interfaces and non-invasive, delegating adapters. Spring Integration's design is inspired by the recognition of a strong affinity between common patterns within Spring and the well-known Enterprise Integration Patterns as described in the book of the same name by Gregor Hohpe and Bobby Woolf (Addison Wesley, 2004). Developers who have read that book should be immediately comfortable with the Spring Integration concepts and terminology.

Spring Integration is motivated by the following goals:

Spring Integration is guided by the following principles:

From the vertical perspective, a layered architecture facilitates separation of concerns, and interface-based contracts between layers promote loose coupling. Spring-based applications are typically designed this way, and the Spring framework and portfolio provide a strong foundation for following this best practice for the full-stack of an enterprise application. Message-driven architectures add a horizontal perspective, yet these same goals are still relevant. Just as "layered architecture" is an extremely generic and abstract paradigm, messaging systems typically follow the similarly abstract "pipes-and-filters" model. The "filters" represent any component that is capable of producing and/or consuming messages, and the "pipes" transport the messages between filters so that the components themselves remain loosely-coupled. It is important to note that these two high-level paradigms are not mutually exclusive. The underlying messaging infrastructure that supports the "pipes" should still be encapsulated in a layer whose contracts are defined as interfaces. Likewise, the "filters" themselves would typically be managed within a layer that is logically above the application's service layer, interacting with those services through interfaces much in the same way that a web-tier would.

1.3.1. Message

In Spring Integration, a Message is a generic wrapper for any Java object combined with metadata used by the framework while handling that object. It consists of a payload and headers. The payload can be of any type and the headers hold commonly required information such as id, timestamp, expiration, and return address. Headers are also used for passing values to and from connected transports. For example, when creating a Message from a received File, the file name may be stored in a header to be accessed by downstream components. Likewise, if a Message's content is ultimately going to be sent by an outbound Mail adapter, the various properties (to, from, cc, subject, etc.) may be configured as Message header values by an upstream component. Developers can also store any arbitrary key-value pairs in the headers.

1.3.2. Message Channel

A Message Channel represents the "pipe" of a pipes-and-filters architecture. Producers send Messages to a channel, and consumers receive Messages from a channel. The Message Channel therefore decouples the messaging components, and also provides a convenient point for interception and monitoring of Messages.

A Message Channel may follow either Point-to-Point or Publish/Subscribe semantics. With a Point-to-Point channel, at most one consumer can receive each Message sent to the channel. Publish/Subscribe channels, on the other hand, will attempt to broadcast each Message to all of its subscribers. Spring Integration supports both of these.

Whereas "Point-to-Point" and "Publish/Subscribe" define the two options for how many consumers will ultimately receive each Message, there is another important consideration: should the channel buffer messages? In Spring Integration, Pollable Channels are capable of buffering Messages within a queue. The advantage of buffering is that it allows for throttling the inbound Messages and thereby prevents overloading a consumer. However, as the name suggests, this also adds some complexity, since a consumer can only receive the Messages from such a channel if a poller is configured. On the other hand, a consumer connected to a Subscribable Channel is simply Message-driven. The variety of channel implementations available in Spring Integration will be discussed in detail in Section 3.2, “Message Channel Implementations”.

A Message Endpoint represents the "filter" of a pipes-and-filters architecture. As mentioned above, the endpoint's primary role is to connect application code to the messaging framework and to do so in a non-invasive manner. In other words, the application code should ideally have no awareness of the Message objects or the Message Channels. This is similar to the role of a Controller in the MVC paradigm. Just as a Controller handles HTTP requests, the Message Endpoint handles Messages. Just as Controllers are mapped to URL patterns, Message Endpoints are mapped to Message Channels. The goal is the same in both cases: isolate application code from the infrastructure. These concepts are discussed at length along with all of the patterns that follow in the Enterprise Integration Patterns book. Here, we provide only a high-level description of the main endpoint types supported by Spring Integration and their roles. The chapters that follow will elaborate and provide sample code as well as configuration examples.

1.4.3. Router

A Message Router is responsible for deciding what channel or channels should receive the Message next (if any). Typically the decision is based upon the Message's content and/or metadata available in the Message Headers. A Message Router is often used as a dynamic alternative to a statically configured output channel on a Service Activator or other endpoint capable of sending reply Messages. Likewise, a Message Router provides a proactive alternative to the reactive Message Filters used by multiple subscribers as described above.

1.4.6. Service Activator

A Service Activator is a generic endpoint for connecting a service instance to the messaging system. The input Message Channel must be configured, and if the service method to be invoked is capable of returning a value, an output Message Channel may also be provided.

Note

The output channel is optional, since each Message may also provide its own 'Return Address' header. This same rule applies for all consumer endpoints.

The Service Activator invokes an operation on some service object to process the request Message, extracting the request Message's payload and converting if necessary (if the method does not expect a Message-typed parameter). Whenever the service object's method returns a value, that return value will likewise be converted to a reply Message if necessary (if it's not already a Message). That reply Message is sent to the output channel. If no output channel has been configured, then the reply will be sent to the channel specified in the Message's "return address" if available.

A request-reply "Service Activator" endpoint connects a target object's method to input and output Message Channels.

1.4.7. Channel Adapter

A Channel Adapter is an endpoint that connects a Message Channel to some other system or transport. Channel Adapters may be either inbound or outbound. Typically, the Channel Adapter will do some mapping between the Message and whatever object or resource is received-from or sent-to the other system (File, HTTP Request, JMS Message, etc). Depending on the transport, the Channel Adapter may also populate or extract Message header values. Spring Integration provides a number of Channel Adapters, and they will be described in upcoming chapters.

An inbound "Channel Adapter" endpoint connects a source system to a MessageChannel.

An outbound "Channel Adapter" endpoint connects a MessageChannel to a target system.

Here is the definition of the Message interface:

public interface Message<T> {

    T getPayload();

    MessageHeaders getHeaders();

}

The Message is obviously a very important part of the API. By encapsulating the data in a generic wrapper, the messaging system can pass it around without any knowledge of the data's type. As an application evolves to support new types, or when the types themselves are modified and/or extended, the messaging system will not be affected by such changes. On the other hand, when some component in the messaging system does require access to information about the Message, such metadata can typically be stored to and retrieved from the metadata in the Message Headers.

Just as Spring Integration allows any Object to be used as the payload of a Message, it also supports any Object types as header values. In fact, the MessageHeaders class implements the java.util.Map interface:

public final class MessageHeaders implements Map<String, Object>, Serializable {
	...
}

As an implementation of Map, the headers can obviously be retrieved by calling get(..) with the name of the header. Alternatively, you can provide the expected Class as an additional parameter. Even better, when retrieving one of the pre-defined values, convenient getters are available. Here is an example of each of these three options:

 Object someValue = message.getHeaders().get("someKey");

 CustomerId customerId = message.getHeaders().get("customerId", CustomerId.class);

 Long timestamp = message.getHeaders().getTimestamp();
 

The following Message headers are pre-defined:

Many inbound and outbound adapter implementations will also provide and/or expect certain headers, and additional user-defined headers can also be configured.

The base implementation of the Message interface is GenericMessage<T>, and it provides two constructors:

new GenericMessage<T>(T payload);

new GenericMessage<T>(T payload, Map<String, Object> headers)

When a Message is created, a random unique id will be generated. The constructor that accepts a Map of headers will copy the provided headers to the newly created Message.

There are also two convenient subclasses available: StringMessage and ErrorMessage. The former accepts a String as its payload:

StringMessage message = new StringMessage("hello world");

String s = message.getPayload();

And, the latter accepts any Throwable object as its payload:

ErrorMessage message = new ErrorMessage(someThrowable);

Throwable t = message.getPayload();

Notice that these implementations take advantage of the fact that the GenericMessage base class is parameterized. Therefore, as shown in both examples, no casting is necessary when retrieving the Message payload Object.

You may notice that the Message interface defines retrieval methods for its payload and headers but no setters. The reason for this is that a Message cannot be modified after its initial creation. Therefore, when a Message instance is sent to multiple consumers (e.g. through a Publish Subscribe Channel), if one of those consumers needs to send a reply with a different payload type, it will need to create a new Message. As a result, the other consumers are not affected by those changes. Keep in mind, that multiple consumers may access the same payload instance or header value, and whether such an instance is itself immutable is a decision left to the developer. In other words, the contract for Messages is similar to that of an unmodifiable Collection, and the MessageHeaders' map further exemplifies that; even though the MessageHeaders class implements java.util.Map, any attempt to invoke a put operation (or 'remove' or 'clear') on the MessageHeaders will result in an UnsupportedOperationException.

Rather than requiring the creation and population of a Map to pass into the GenericMessage constructor, Spring Integration does provide a far more convenient way to construct Messages: MessageBuilder. The MessageBuilder provides two factory methods for creating Messages from either an existing Message or with a payload Object. When building from an existing Message, the headers and payload of that Message will be copied to the new Message:

Message<String> message1 = MessageBuilder.withPayload("test")
        .setHeader("foo", "bar")
        .build();

Message<String> message2 = MessageBuilder.fromMessage(message1).build();

assertEquals("test", message2.getPayload());
assertEquals("bar", message2.getHeaders().get("foo"));

If you need to create a Message with a new payload but still want to copy the headers from an existing Message, you can use one of the 'copy' methods.

Message<String> message3 = MessageBuilder.withPayload("test3")
        .copyHeaders(message1.getHeaders())
        .build();

Message<String> message4 = MessageBuilder.withPayload("test4")
        .setHeader("foo", 123)
        .copyHeadersIfAbsent(message1.getHeaders())
        .build();

assertEquals("bar", message3.getHeaders().get("foo"));
assertEquals(123, message4.getHeaders().get("foo"));

Notice that the copyHeadersIfAbsent does not overwrite existing values. Also, in the second example above, you can see how to set any user-defined header with setHeader. Finally, there are set methods available for the predefined headers as well as a non-destructive method for setting any header (MessageHeaders also defines constants for the pre-defined header names).

Message<Integer> importantMessage = MessageBuilder.withPayload(99)
        .setPriority(MessagePriority.HIGHEST)
        .build();

assertEquals(MessagePriority.HIGHEST, importantMessage.getHeaders().getPriority());

Message<Integer> anotherMessage = MessageBuilder.fromMessage(importantMessage)
        .setHeaderIfAbsent(MessageHeaders.PRIORITY, MessagePriority.LOW)
        .build();

assertEquals(MessagePriority.HIGHEST, anotherMessage.getHeaders().getPriority());

The MessagePriority is only considered when using a PriorityChannel (as described in the next chapter). It is defined as an enum with five possible values:

public enum MessagePriority {
    HIGHEST,
    HIGH,
    NORMAL,
    LOW,
    LOWEST
}

Spring Integration's top-level MessageChannel interface is defined as follows.

public interface MessageChannel {

    String getName();

    boolean send(Message message);

    boolean send(Message message, long timeout);
}

When sending a message, the return value will be true if the message is sent successfully. If the send call times out or is interrupted, then it will return false.

public interface PollableChannel extends MessageChannel {

    Message<?> receive();

    Message<?> receive(long timeout);

    List<Message<?>> clear();

    List<Message<?>> purge(MessageSelector selector);

}

public interface SubscribableChannel extends MessageChannel {

    boolean subscribe(MessageHandler handler);

    boolean unsubscribe(MessageHandler handler);

}

Spring Integration provides several different Message Channel implementations. Each is briefly described in the sections below.

public QueueChannel(int capacity)

One of the advantages of a messaging architecture is the ability to provide common behavior and capture meaningful information about the messages passing through the system in a non-invasive way. Since the Messages are being sent to and received from MessageChannels, those channels provide an opportunity for intercepting the send and receive operations. The ChannelInterceptor strategy interface provides methods for each of those operations:

public interface ChannelInterceptor {

    Message<?> preSend(Message<?> message, MessageChannel channel);

    void postSend(Message<?> message, MessageChannel channel, boolean sent);

    boolean preReceive(MessageChannel channel);

    Message<?> postReceive(Message<?> message, MessageChannel channel);
}

After implementing the interface, registering the interceptor with a channel is just a matter of calling:

channel.addInterceptor(someChannelInterceptor);

The methods that return a Message instance can be used for transforming the Message or can return 'null' to prevent further processing (of course, any of the methods can throw a RuntimeException). Also, the preReceive method can return 'false' to prevent the receive operation from proceeding.

Spring Integration also provides an implementation of the Wire Tap pattern. It is a simple interceptor that sends the Message to another channel without otherwise altering the existing flow. It can be very useful for debugging and monitoring. An example is shown in Section 3.5.10, “Wire Tap”.

Because it is rarely necessary to implement all of the interceptor methods, a ChannelInterceptorAdapter class is also available for sub-classing. It provides no-op methods (the void method is empty, the Message returning methods return the Message as-is, and the boolean method returns true). Therefore, it is often easiest to extend that class and just implement the method(s) that you need as in the following example.

public class CountingChannelInterceptor extends ChannelInterceptorAdapter {

    private final AtomicInteger sendCount = new AtomicInteger();

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        sendCount.incrementAndGet();
        return message;
    }
}

As you will see when the endpoints and their various configuration options are introduced, Spring Integration provides a foundation for messaging components that enables non-invasive invocation of your application code from the messaging system. However, sometimes it is necessary to invoke the messaging system from your application code. For convenience when implementing such use-cases, Spring Integration provides a MessageChannelTemplate that supports a variety of operations across the Message Channels, including request/reply scenarios. For example, it is possible to send a request and wait for a reply.

MessageChannelTemplate template = new MessageChannelTemplate();

Message reply = template.sendAndReceive(new StringMessage("test"), someChannel);

In that example, a temporary anonymous channel would be created internally by the template. The 'sendTimeout' and 'receiveTimeout' properties may also be set on the template, and other exchange types are also supported.

public boolean send(final Message<?> message, final MessageChannel channel) { ... }

public Message<?> sendAndReceive(final Message<?> request, final MessageChannel channel) { .. }

public Message<?> receive(final PollableChannel<?> channel) { ... }

To create a Message Channel instance, you can use the 'channel' element:

<channel id="exampleChannel"/>

The default channel type is Point to Point. To create a Publish Subscribe channel, use the "publish-subscribe-channel" element:

<publish-subscribe-channel id="exampleChannel"/>

To create a Datatype Channel that only accepts messages containing a certain payload type, provide the fully-qualified class name in the channel element's datatype attribute:

<channel id="numberChannel" datatype="java.lang.Number"/>

Note that the type check passes for any type that is assignable to the channel's datatype. In other words, the "numberChannel" above would accept messages whose payload is java.lang.Integer or java.lang.Double. Multiple types can be provided as a comma-delimited list:

<channel id="stringOrNumberChannel" datatype="java.lang.String,java.lang.Number"/>

When using the "channel" element without any sub-elements, it will create a DirectChannel instance (a SubscribableChannel).

However, you can alternatively provide a variety of "queue" sub-elements to create any of the pollable channel types (as described in Section 3.2, “Message Channel Implementations”). Examples of each are shown below.

<channel id="directChannel"/>

<channel id="failFastChannel">
    <dispatcher failover="false"/>
</channel>

<channel id="channelWithFixedOrderSequenceFailover">
    <dispatcher load-balancer="none"/>
</channel>

<channel id="queueChannel">
    <queue capacity="25"/>
</channel>

<publish-subscribe-channel id="pubsubChannel" task-executor="someExecutor"/>

<publish-subscribe-channel id="pubsubChannel" apply-sequence="true"/>

<channel id="executorChannel">
    <dispatcher task-executor="someExecutor"/>
</channel>

<channel id="executorChannelWithoutFailover">
    <dispatcher task-executor="someExecutor" failover="false"/>
</channel>

<channel id="priorityChannel">
    <priority-queue capacity="20"/>
</channel>

<channel id="priorityChannel" datatype="example.Widget">
    <priority-queue comparator="widgetComparator"
                    capacity="10"/>
</channel>

<channel id="rendezvousChannel"/>
    <rendezvous-queue/>
</channel>

<thread-local-channel id="threadLocalChannel"/>

<channel id="exampleChannel">
    <interceptors>
        <ref bean="trafficMonitoringInterceptor"/>
    </interceptors>
</channel>

<int:channel-interceptor pattern="input*, bar*, foo" order="3">
    <bean class="foo.barSampleInterceptor"/>
</int:channel-interceptor>

<int:channel-interceptor ref="myInterceptor" pattern="input*, bar*, foo" order="3"/> 
		
<bean id="myInterceptor" class="foo.barSampleInterceptor"/>

<int:channel id="inputChannel"> 
	<int:interceptors>
		<int:wire-tap channel="logger"/> 
	</int:interceptors>
</int:channel>

 <channel id="in">
     <interceptors>
         <wire-tap channel="logger"/>
     </interceptors>
 </channel>

 <logging-channel-adapter id="logger" level="DEBUG"/>

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 Section 3.1.2, “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. 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>
         <interval-trigger interval="1000"/>
     </poller>
</transformer>

For a poller based on a Cron expression, use the "cron-trigger" child element instead:

 <transformer input-channel="pollable"
              ref="transformer"
              output-channel="output">
     <poller>
         <cron-trigger expression="*/10 * * * * MON-FRI"/>
     </poller>
 </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. However, it is also possible to create top-level pollers in which case only a "ref" is required:

 <poller id="weekdayPoller">
     <cron-trigger expression="*/10 * * * * MON-FRI"/>
 </poller>

 <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">
     <interval-trigger interval="3" time-unit="SECONDS"/>
 </poller>

 <!-- 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>
    <interval-trigger interval="1000"/>
    <transactional transaction-manager="txManager"
                   propagation="REQUIRED"
                   isolation="REPEATABLE_READ"
                   timeout="10000"
                   read-only="false"/>
</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"/>
     <interval-trigger interval="5" time-unit="SECONDS"/>
 </poller>

 <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">
         <interval-trigger interval="10"/>
     </poller>
 </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>

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 or @MessageMapping 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>

An "inbound-channel-adapter" element can invoke any method on a Spring-managed Object and send a non-null return value to a MessageChannel after converting it to a Message. When the adapter's subscription is activated, a poller will attempt to receive messages from the source. The poller will be scheduled with the TaskScheduler according to the provided configuration. To configure the polling interval or cron expression for an individual channel-adapter, provide a 'poller' element with either an 'interval-trigger' (in milliseconds) or 'cron-trigger' sub-element.

<inbound-channel-adapter ref="source1" method="method1" channel="channel1">
    <poller>
        <interval-trigger interval="5000"/>
    </poller>
</inbound-channel-adapter>

<inbound-channel-adapter ref="source2" method="method2" channel="channel2">
    <poller>
        <cron-trigger expression="30 * 9-17 * * MON-FRI"/>
    </poller>
</channel-adapter>

An "outbound-channel-adapter" element can also connect a MessageChannel to any POJO consumer method that should be invoked with the payload of Messages sent to that channel.

<outbound-channel-adapter channel="channel1" ref="target1" method="method1"/>

If the channel being adapted is a PollableChannel, provide a poller sub-element:

<outbound-channel-adapter channel="channel2" ref="target2" method="method2">
    <poller>
        <interval-trigger interval="3000"/>
    </poller>
</outbound-channel-adapter>
<beans:bean id="target1" class="org.bar.Foo"/>

Using a "ref" attribute is generally recommended if the POJO consumer implementation can be reused in other <outbound-channel-adapter> definitions. However if the consumer implementation should be scoped to a single definition of the <outbound-channel-adapter>, you can define it as inner bean:

<outbound-channel-adapter channel="channel2" method="method2">
        <beans:bean class="org.bar.Foo"/>
    
</outbound-channel-adapter>

Any Channel Adapter can be created without a "channel" reference in which case it will implicitly create an instance of DirectChannel. The created channel's name will match the "id" attribute of the <inbound-channel-adapter/> or <outbound-channel-adapter element. Therefore, if the "channel" is not provided, the "id" is required.

Since content-based routing often requires some domain-specific logic, most use-cases will require Spring Integration's options for delegating to POJOs using the XML namespace support and/or Annotations. Both of these are discussed below, but first we present a couple implementations that are available out-of-the-box since they fulfill generic, but common, requirements.

<bean id="payloadTypeRouter" class="org.springframework.integration.router.PayloadTypeRouter">
    <property name="payloadTypeChannelMap">
        <map>
            <entry key="java.lang.String" value-ref="stringChannel"/>
            <entry key="java.lang.Integer" value-ref="integerChannel"/>
        </map>
    </property>
</bean>

<payload-type-router input-channel="routingChannel">
	<mapping type="java.lang.String" channel="stringChannel" />
	<mapping type="java.lang.Integer" channel="integerChannel" />
</payload-type-router>

 <bean id="myHeaderValueRouter"
		class="org.springframework.integration.router.HeaderValueRouter">
  <constructor-arg value="someHeaderName" /> 
  <property name="channelResolver">
    <bean class="org.springframework.integration.channel.MapBasedChannelResolver">
      <property name="channelMap">
        <map>
          <entry key="someHeaderValue" value-ref="channelA" />
          <entry key="someOtherHeaderValue" value-ref="channelB" />
        </map>
      </property>
    </bean>
  </property>
</bean>

 
<bean id="myHeaderValueRouter"
	class="org.springframework.integration.router.HeaderValueRouter">
	<constructor-arg value="someHeaderName" /> 
</bean>

<header-value-router input-channel="routingChannel" header-name="testHeader">
	<mapping value="someHeaderValue" channel="channelA" />
	<mapping value="someOtherHeaderValue" channel="channelB" />
</header-value-router>

<header-value-router input-channel="routingChannel" header-name="testHeader"/>

<bean id="recipientListRouter" class="org.springframework.integration.router.RecipientListRouter">
    <property name="channels">
        <list>
            <ref bean="channel1"/>
            <ref bean="channel2"/>
            <ref bean="channel3"/>
        </list>
    </property>
</bean>

Configuration for RecipientListRouter is also supported via namespace support provided by Spring Integration (see Section B.2, “Namespace Support”). The example below demonstrates namespace-based configuration of RecipientListRouter and all the supported attributes using Spring Integration namespace support:

<recipient-list-router id="customRouter" input-channel="routingChannel"
				timeout="1234"
				ignore-send-failures="true"
				apply-sequence="true">
	<recipient channel="channel1"/>
	<recipient channel="channel2"/>
</recipient-list-router>

The "router" element provides a simple way to connect a router to an input channel, and also accepts the optional default output channel. The "ref" may provide the bean name of a custom Router implementation (extending AbstractMessageRouter):

<router ref="payloadTypeRouter" input-channel="input1" default-output-channel="defaultOutput1"/>

<router ref="recipientListRouter" input-channel="input2" default-output-channel="defaultOutput2"/>

<router ref="customRouter" input-channel="input3" default-output-channel="defaultOutput3"/>

<beans:bean id="customRouterBean class="org.foo.MyCustomRouter"/>

Alternatively, the "ref" may point to a simple Object that contains the @Router annotation (see below), or the "ref" may be combined with an explicit "method" name. When specifying a "method", the same behavior applies as described in the @Router annotation section below.

<router input-channel="input" ref="somePojo" method="someMethod"/>

Using a "ref" attribute is generally recommended if the custom router implementation can be reused in other <router> definitions. However if the custom router implementation should be scoped to a concrete definition of the <router>, you can provide an inner bean definition:

<router method="someMethod" input-channel="input3" default-output-channel="defaultOutput3">
	<beans:bean class="org.foo.MyCustomRouter"/>
</router>

When using the @Router annotation, the annotated method can return either the MessageChannel or String type. In the case of the latter, the endpoint will resolve the channel name as it does for the default output. Additionally, the method can return either a single value or a collection. When a collection is returned, the reply message will be sent to multiple channels. To summarize, the following method signatures are all valid.

@Router
public MessageChannel route(Message message) {...}

@Router
public List<MessageChannel> route(Message message) {...}

@Router
public String route(Foo payload) {...}

@Router
public List<String> route(Foo payload) {...}

In addition to payload-based routing, a common requirement is to route based on metadata available within the message header as either a property or attribute. Rather than requiring use of the Message type as the method parameter, the @Router annotation may also use the @Header parameter annotation that is documented in Section B.5, “Annotation Support”.

@Router
public List<String> route(@Header("orderStatus") OrderStatus status)

Message Filters are used to decide whether a Message should be passed along or dropped based on some criteria such as a Message Header value or even content within the Message itself. Therefore, a Message Filter is similar to a router, except that for each Message received from the filter's input channel, that same Message may or may not be sent to the filter's output channel. Unlike the router, it makes no decision regarding which Message Channel to send to but only decides whether to send.

In Spring Integration, a Message Filter may be configured as a Message Endpoint that delegates to some implementation of the MessageSelector interface. That interface is itself quite simple:

 public interface MessageSelector {

     boolean accept(Message<?> message);

 }

The MessageFilter constructor accepts a selector instance:

 MessageFilter filter = new MessageFilter(someSelector);

The <filter> element is used to create a Message-selecting endpoint. In addition to "input-channel" and "output-channel" attributes, it requires a "ref". The "ref" may point to a MessageSelector implementation:

 <filter input-channel="input" ref="selector" output-channel="output"/>

 <bean id="selector" class="example.MessageSelectorImpl"/>

Alternatively, the "method" attribute can be added at which point the "ref" may refer to any object. The referenced method may expect either the Message type or the payload type of inbound Messages. The return value of the method must be a boolean value. Any time the method returns 'true', the Message will be passed along to the output-channel.

 <filter input-channel="input" output-channel="output"
         ref="exampleObject" method="someBooleanReturningMethod"/>

 <bean id="exampleObject" class="example.SomeObject"/>

If the selector or adapted POJO method returns false, there are a few settings that control the fate of the rejected Message. By default (if configured like the example above), the rejected Messages will be silently dropped. If rejection should instead indicate an error condition, then set the 'throw-exception-on-rejection' flag to true:

 <filter input-channel="input" ref="selector"
      output-channel="output" throw-exception-on-rejection="true"/> 

If you want the rejected messages to go to a specific channel, provide that reference as the 'discard-channel':

 <filter input-channel="input" ref="selector"
      output-channel="output" discard-channel="rejectedMessages"/> 

Using a "ref" attribute is generally recommended if the custom filter implementation can be reused in other <filter> definitions. However if the custom filter implementation should be scoped to a single <filter> element, provide an inner bean definition:

<filter method="someMethod" input-channel="inChannel" output-channel="outChannel">
	<beans:bean class="org.foo.MyCustomFilter"/>
</filter>

With the introduction of SpEL Spring Integration has added the expression attribute to the filter element. It can be used to avoid Java entirely for simple filters.

          <filter input-channel="input" expression="payload.equals(nonsense)"/>
      

The string passed as the expression attribute will be evaluated as a SpEL expression in the context of the message. If it is needed to include the result of an expression in the scope of the application context you can use the #{} notation as defined in the SpEL reference documentation SpEL reference documentation .

          <filter input-channel="input" expression="payload.matches(#{filterPatterns.nonsensePattern})"/>
      

Message Transformers play a very important role in enabling the loose-coupling of Message Producers and Message Consumers. Rather than requiring every Message-producing component to know what type is expected by the next consumer, Transformers can be added between those components. Generic transformers, such as one that converts a String to an XML Document, are also highly reusable.

For some systems, it may be best to provide a Canonical Data Model, but Spring Integration's general philosophy is not to require any particular format. Rather, for maximum flexibility, Spring Integration aims to provide the simplest possible model for extension. As with the other endpoint types, the use of declarative configuration in XML and/or Annotations enables simple POJOs to be adapted for the role of Message Transformers. These configuration options will be described below.

The <transformer> element is used to create a Message-transforming endpoint. In addition to "input-channel" and "output-channel" attributes, it requires a "ref". The "ref" may either point to an Object that contains the @Transformer annotation on a single method (see below) or it may be combined with an explicit method name value provided via the "method" attribute.

<transformer id="testTransformer" ref="testTransformerBean" input-channel="inChannel" 
					   method="transform" output-channel="outChannel"/>
<beans:bean id="testTransformerBean" class="org.foo.TestTransformer" />

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

<transformer id="testTransformer" input-channel="inChannel" method="transform"
								output-channel="outChannel">
	<beans:bean class="org.foo.TestTransformer"/>
</transformer>

The method that is used for transformation may expect either the Message type or the payload type of inbound Messages. It may also accept Message header values either individually or as a full map by using the @Header and @Headers parameter annotations respectively. The return value of the method can be any type. If the return value is itself a Message, that will be passed along to the transformer's output channel. If the return type is a Map, and the original Message payload was not a Map, the entries in that Map will be added to the Message headers of the original Message (the keys must be Strings). If the return value is null, then no reply Message will be sent (effectively the same behavior as a Message Filter returning false). Otherwise, the return value will be sent as the payload of an outbound reply Message.

There are a also a few Transformer implementations available out of the box. Because, it is fairly common to use the toString() representation of an Object, Spring Integration provides an ObjectToStringTransformer whose output is a Message with a String payload. That String is the result of invoking the toString operation on the inbound Message's payload.

 <object-to-string-transformer input-channel="in" output-channel="out"/>

A potential example for this would be sending some arbitrary object to the 'outbound-channel-adapter' in the file namespace. Whereas that Channel Adapter only supports String, byte-array, or java.io.File payloads by default, adding this transformer immediately before the adapter will handle the necessary conversion. Of course, that works fine as long as the result of the toString() call is what you want to be written to the File. Otherwise, you can just provide a custom POJO-based Transformer via the generic 'transformer' element shown previously.

If you need to serialize an Object to a byte array or deserialize a byte array back into an Object, Spring Integration provides symmetrical serialization transformers.

 <payload-serializing-transformer input-channel="objectsIn" output-channel="bytesOut"/>

 <payload-deserializing-transformer input-channel="bytesIn" output-channel="objectsOut"/>

If you only need to add headers to a Message, and they are not dynamically determined by Message content, then referencing a custom implementation may be overkill. For that reason, Spring Integration provides the 'header-enricher' element.

 <header-enricher input-channel="in" output-channel="out">
     <header name="foo" value="123"/>
     <header name="bar" ref="someBean"/>
 </header-enricher>

As added convenience, Spring Integration also provides Object-to-Map and Map-to-Object transformers which utilize Spring Expression Language (SpEL) to serialize and de-serialize the object graphs. Object hierarchy is introspected to the most primitive types (e.g., String, int etc.). The path to this type is described via SpEL, which becomes the keykey in the transformed Map with primitive type being the value.

For example:

public class Parent{
    private Child child;
    private String name; 
    // setters and getters are omitted
}

public class Child{
   private String name; 
   private List<String> nickNames;
   // setters and getters are omitted
}

... will be transformed to a Map which looks like this: {person.name=George, person.child.name=Jenna, person.child.nickNames[0]=Bimbo . . . etc}

SpEL-based Map allows you to describe the object structure without sharing the actual types allowing you to restore/rebuild the object graph into a differently typed Object graph as long as you maintain the structure.

For example: The above structure could be easily restored back to the following Object graph via Map-to-Object transformer:

public class Father{
    private Kid child;
    private String name; 
    // setters and getters are omitted
}

public class Kid{
   private String name; 
   private List<String> nickNames;
   // setters and getters are omitted
}

To configure these transformers, Spring Integration provides namespace support Object-to-Map:

<object-to-map-transformer input-channel="directInput" output-channel="output"/>

Map-to-Object

<int:map-to-object-transformer input-channel="input" 
	                       output-channel="output" 
	                        type="org.foo.Person"/>

or

<int:map-to-object-transformer input-channel="inputA" 
	                              output-channel="outputA" 
	                              ref="person"/>
<bean id="person" class="org.foo.Person" scope="prototype"/>	

The @Transformer annotation can also be added to methods that expect either the Message type or the message payload type. The return value will be handled in the exact same way as described above in the section describing the <transformer> element.

@Transformer
Order generateOrder(String productId) {
    return new Order(productId);
}

Transformer methods may also accept the @Header and @Headers annotations that is documented in Section B.5, “Annotation Support”

@Transformer
Order generateOrder(String productId, @Header("customerName") String customer) {
    return new Order(productId, customer);
}

The Splitter is a component whose role is to partition a message in several parts, and send the resulting messages to be processed independently. Very often, they are upstream producers in a pipeline that includes an Aggregator.

The API for performing splitting consists from one base class, AbstractMessageSplitter, which is a MessageHandler implementation, encapsulating features which are common to splitters, such as filling in the appropriate message headers CORRELATION_ID, SEQUENCE_SIZE, and SEQUENCE_NUMBER on the messages that are produced. This allows to track down the messages and the results of their processing (in a typical scenario, these headers would be copied over to the messages that are produced by the various transforming endpoints), and use them, for example, in a Composed Message Processor scenario.

An excerpt from AbstractMessageSplitter can be seen below:

public abstract class AbstractMessageSplitter
                extends AbstractReplyProducingMessageConsumer {
  ...
  protected abstract Object splitMessage(Message<?> message);

}

For implementing a specific Splitter in an application, a developer can extend AbstractMessageSplitter and implement the splitMessage method, thus defining the actual logic for splitting the messages. The return value can be one of the following:

In Spring Integration, any POJO can implement the splitting algorithm, provided that it defines a method that accepts a single argument and has a return value. In this case, the return value of the method will be interpreted as described above. The input argument might either be a Message or a simple POJO. In the latter case, the splitter will receive the payload of the incoming message. Since this decouples the code from the Spring Integration API and will typically be easier to test, it is the recommended approach.

A splitter can be configured through XML as follows:

<channel id="inputChannel"/>

<splitter id="splitter" (1)
  ref="splitterBean" (2)
  method="split" (3)
  input-channel="inputChannel" (4)
  output-channel="outputChannel" (5)/>

<channel id="outputChannel"/>

<beans:bean id="splitterBean" class="sample.PojoSplitter"/>

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

<splitter id="testSplitter" input-channel="inChannel" method="split"
								output-channel="outChannel">
	<beans:bean class="org.foo.TestSplitter"/>
</spliter>

The @Splitter annotation is applicable to methods that expect either the Message type or the message payload type, and the return values of the method should be a collection of any type. If the returned values are not actual Message objects, then each of them will be sent as the payload of a message. Those messages will be sent to the output channel as designated for the endpoint on which the @Splitter is defined.

@Splitter
List<LineItem> extractItems(Order order) {
    return order.getItems()
}

Basically a mirror-image of the Splitter, the Aggregator is a type of Message Handler that receives multiple Messages and combines them into a single Message. In fact, Aggregators are often downstream consumers in a pipeline that includes a Splitter.

Technically, the Aggregator is more complex than a Splitter, because it is required to maintain state (the Messages to be aggregated), to decide when the complete group of Messages is available. In order to do this it requires a MessageStore

The Aggregator combines a group of related messages, by correlating and storing them, until the group is deemed complete. At that point, the Aggregator will create a single message by processing the whole group, and will send that aggregated message as output.

An main aspect of implementing an Aggregator is providing the logic that has to be executed when the aggregation (creation of a single message out of many) takes place. The other two aspects are correlation and release

In Spring Integration, the grouping of the messages for aggregation (correlation) is done by default based on their CORRELATION_ID message header (i.e. the messages with the same CORRELATION_ID will be grouped together). However, this can be customized, and the users can opt for other ways of specifying how the messages should be grouped together, by using a CorrelationStrategy (see below).

To determine whether or not a group of messages may be processed, a ReleaseStrategy is consulted. The default release strategy for aggregator will release groups that have all messages from the sequence, but this can be entirely customized

The Aggregation API consists of a number of classes:

public abstract class AbstractAggregatingMessageGroupProcessor
              implements MessageGroupProcessor {

    protected Map<String, Object> aggregateHeaders(MessageGroup group) {
  ....
    }

  	protected abstract Object aggregatePayloads(MessageGroup group);

}

private volatile CorrelationStrategy correlationStrategy =
          new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID);

public interface ReleaseStrategy {

  boolean canRelease(MessageGroup messages);

}

public interface CorrelationStrategy {

  Object getCorrelationKey(Message<?> message);

}

Spring Integration supports the configuration of an aggregator via XML through the <aggregator/> element. Below you can see an example of an aggregator with all optional parameters defined.

<channel id="inputChannel"/>

<aggregator id="completelyDefinedAggregator" (1)
    input-channel="inputChannel" (2)
    output-channel="outputChannel"  (3)
    discard-channel="discardChannel"  (4)
    ref="aggregatorBean" (5)
    method="add" (6)
    release-strategy="releaseStrategyBean"  (7)
    release-strategy-method="canRelease" (8)
    correlation-strategy="correlationStrategyBean" (9)
    correlation-strategy-method="groupNumbersByLastDigit" (10)
    message-store="messageStore" (11)
    send-partial-result-on-expiry="true" (12)
    send-timeout="86420000" (13) /> 

<channel id="outputChannel"/>

<bean id="aggregatorBean" class="sample.PojoAggregator"/>

<bean id="releaseStrategyBean" class="sample.PojoReleaseStrategy"/>

<bean id="correlationStrategyBean" class="sample.PojoCorrelationStrategy"/>

Using a "ref" attribute is generally recommended if a custom aggregator handler implementation can be reused in other <aggregator> definitions. However if a custom aggregator handler implementation should be scoped to a concrete definition of the <aggregator>, you can use an inner bean definition (starting with version 1.0.3) for custom aggregator handlers within the <aggregator> element:

<aggregator input-channel="input" method="sum" output-channel="output">
    <beans:bean class="org.foo.ExampleAggregator"/>
</aggregator>

An example implementation of the aggregator bean looks as follows:

public class PojoAggregator {

  public Long add(List<Long> results) {
    long total = 0l;
    for (long partialResult: results) {
      total += partialResult;
    }
    return total;
  }

}

An implementation of the completion strategy bean for the example above may be as follows:

public class PojoReleaseStrategy {
...
  public boolean canRelease(List<Long> numbers) {
    int sum = 0;
    for (long number: numbers) {
      sum += number;
    }
    return sum >= maxValue;
  }
}

An implementation of the correlation strategy bean for the example above may be as follows:

public class PojoCorrelationStrategy {
...
  public Long groupNumbersByLastDigit(Long number) {
    return number % 10;
  }
}

For example, this aggregator would group numbers by some criterion (in our case the remainder after dividing by 10) and will hold the group until the sum of the numbers which represents the payload exceeds a certain value.

Aggregator (and some other patterns in Spring Integration) is a stateful pattern that requires decisions to be made based on a group of messages that have arrived over a period of time, all with the same correlation key. The design of the interfaces in the stateful patterns (e.g. ReleaseStrategy) is driven by the principle that the components (framework and user) should be to remain stateless. All state is carried by the MessageGroup and its management is delegated to the MessageGroupStore.

The MessageGroupStore accumulates state information in MessageGroups, potentially forever. So to prevent stale state from hanging around, and for volatile stores to provide a hook for cleaning up when the application shots down, the MessageGroupStore allows the user to register callbacks to apply to MessageGroups when they expire. The interface is very straighforward:

public interface MessageGroupCallback {

  void execute(MessageGroupStore messageGroupStore, MessageGroup group);

}

The callback has access directly to the store and the message group so it can manage the persistent state (e.g. by removing the group from the store entirely).

The MessageGroupStore maintains a list of these callbacks which it applies when asked to all messages whose timestamp is earlier than a time supplied as a parameter:

public interface MessageGroupStore {
   void registerMessageGroupExpiryCallback(MessageGroupCallback callback);
   int expireMessageGroups(long timeout);
}

The expireMessageGroups method can be called with a timeout value: any message older than the current time minus this value wiull be expired, and have the callbacks applied. Thus it is the user of the store that defines what is meant by message group "expiry".

As a convenience for users, Spring Integration provides a wrapper for the message expiry in the form of a MessageGroupStoreReaper:

<bean id="reaper" class="org...MessageGroupStoreReaper">
  <property name="messageGroupStore" ref="messageStore"/>
  <property name="timeout" value="10"/>
</bean>

<task:scheduled-tasks scheduler="scheduler">	
  <task:scheduled ref="reaper" method="run" fixed-rate="10000"/>
</task:scheduled-tasks>

The reaper is a Runnable, and all that is happening is that the message group store's expire method is being called in the sample above once every 10 seconds. In addition to the reaper, the expiry callbacks are invoked when the application shuts down via a lifecycle callback in the CorrelatingMessageHandler.

The CorrelatingMessageHandler registers its own expiry callback, and this is the link with the boolean flag send-partial-result-on-expiry in the XML configuration of the aggregator. If the flag is set to true, then when the expiry callback is invoked then any unmarked messages in groups that are not yet released can be sent on to the downstream channel.

An aggregator configured using annotations can look like this.

public class Waiter {
  ... 

  @Aggregator (1) 
  public Delivery aggregatingMethod(List<OrderItem> items) {
    ...
  }

  @ReleaseStrategy (2)
  public boolean releaseChecker(List<Message<?>> messages) {
    ...
  }

  @CorrelationStrategy (3)
  public String correlateBy(OrderItem item) {
    ...
  }

}

All of the configuration options provided by the xml element are also available for the @Aggregator annotation.

The aggregator can be either referenced explicitly from XML or, if the @MessageEndpoint is defined on the class, detected automatically through classpath scanning.

Related to the Aggregator, albeit different from a functional standpoint, is the Resequencer.

The Resequencer works in a similar way to the Aggregator, in the sense that it uses the CORRELATION_ID to store messages in groups, the difference being that the Resequencer does not process the messages in any way. It simply releases them in the order of their SEQUENCE_NUMBER header values.

With respect to that, the user might opt to release all messages at once (after the whole sequence, according to the SEQUENCE_SIZE, has been released), or as soon as a valid sequence is available.

Configuring a resequencer requires only including the appropriate element in XML.

A sample resequencer configuration is shown below.

<channel id="inputChannel"/>

<channel id="outputChannel"/>

<resequencer id="completelyDefinedResequencer" (1)
  input-channel="inputChannel" (2)
  output-channel="outputChannel" (3)
  discard-channel="discardChannel" (4)
  release-partial-sequences="true" (5)
  message-store="messageStore" (6)
  send-partial-result-on-expiry="true" (7)
  send-timeout="86420000" (8) /> 

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"/>

The MessageHandlerChain is an implementation of MessageHandler that can be configured as a single Message Endpoint while actually delegating to a chain of other handlers, such as Filters, Transformers, Splitters, and so on. This can lead to a much simpler configuration when several handlers need to be connected in a fixed, linear progression. For example, it is fairly common to provide a Transformer before other components. Similarly, when providing a Filter before some other component in a chain, you are essentially creating a Selective Consumer. In either case, the chain only requires a single input-channel and a single output-channel as opposed to the configuration of channels for each individual component.

The handler chain simplifies configuration while internally maintaining the same degree of loose coupling between components, and it is trivial to modify the configuration if at some point a non-linear arrangement is required.

Internally, the chain will be expanded into a linear setup of the listed endpoints, separated by direct channels. The reply channel header will not be taken into account within the chain: only after the last handler is invoked will the resulting message be forwarded on to the reply channel or the chain's output channel. Because of this setup all handlers except the last require a setOutputChannel implementation. The last handler only needs an output channel if the outputChannel on the MessageHandlerChain is set.

In most cases there is no need to implement MessageHandlers yourself. The next section will focus on namespace support for the chain element. Most Spring Integration endpoints, like Service Activators and Transformers, are suitable for use within a MessageHandlerChain.

The <chain> element provides an 'input-channel' attribute, and if the last element in the chain is capable of producing reply messages (optional), it also supports an 'output-channel' attribute. The sub-elements are then filters, transformers, splitters, and service-activators. The last element may also be a router.

 <chain input-channel="input" output-channel="output">
     <filter ref="someSelector" throw-exception-on-rejection="true"/>
     <header-enricher error-channel="customErrorChannel">
         <header name="foo" value="bar"/>
     </header-enricher>
     <service-activator ref="someService" method="someMethod"/>
 </chain>

The <header-enricher> element used in the above example will set a message header with name "foo" and value "bar" on the message. A header enricher is a specialization of Transformer that touches only header values. You could obtain the same result by implementing a MessageHandler that did the header modifications and wiring that as a bean.

Some time you need to make a nested call to another chain from within the chain and then come back and continue execution within the original chain. To accomplish this you can utilize Messaging Gateway by including light-configuration via <gateway> element. For example:

 	<si:chain id="main-chain" input-channel="inputA" output-channel="inputB">
		<si:header-enricher>
			<si:header name="name" value="Many" />
		</si:header-enricher>
		<si:service-activator>
			<bean class="org.foo.SampleService" />
		</si:service-activator>
		<si:gateway request-channel="inputC"/>  
	</si:chain>
	<si:chain id="nested-chain-a" input-channel="inputC">
		<si:header-enricher>
			<si:header name="name" value="Moe" />
		</si:header-enricher>
		<si:gateway request-channel="inputD"/> 
		<si:service-activator>
			<bean class="org.foo.SampleService" />
		</si:service-activator>
	</si:chain>	
	<si:chain id="nested-chain-b" input-channel="inputD">
		<si:header-enricher>
			<si:header name="name" value="Jack" />
		</si:header-enricher>
		<si:service-activator>
			<bean class="org.foo.SampleService" />
		</si:service-activator>
	</si:chain>

In the above example the nested-chain-a will be called at the end of main-chain processing by the 'gateway' element configured there. While in nested-chain-a a call to a nested-chain-b will be made after header enrichment and then it will come back to finish execution in nested-chain-b finally getting back to the main-chain. When light version of <gateway> element is defined in the chain SI will construct an instance SimpleMessagingGateway (no need to provide 'service-interface' configuration) which will take the message in its current state and will place it on the channel defined via 'request-channel' attribute. Upon processing Message will be returned to the gateway and continue its journey within the current chain.

A Messaging Bridge is a relatively trivial endpoint that simply connects two Message Channels or Channel Adapters. For example, you may want to connect a PollableChannel to a SubscribableChannel so that the subscribing endpoints do not have to worry about any polling configuration. Instead, the Messaging Bridge provides the polling configuration.

By providing an intermediary poller between two channels, a Messaging Bridge can be used to throttle inbound Messages. The poller's trigger will determine the rate at which messages arrive on the second channel, and the poller's "maxMessagesPerPoll" property will enforce a limit on the throughput.

Another valid use for a Messaging Bridge is to connect two different systems. In such a scenario, Spring Integration's role would be limited to making the connection between these systems and managing a poller if necessary. It is probably more common to have at least a Transformer between the two systems to translate between their formats, and in that case, the channels would be provided as the 'input-channel' and 'output-channel' of a Transformer endpoint. If data format translation is not required, the Messaging Bridge may indeed be sufficient.

The <bridge> element is used to create a Messaging Bridge between two Message Channels or Channel Adapters. Simply provide the "input-channel" and "output-channel" attributes:

 <bridge input-channel="input" output-channel="output"/>

As mentioned above, a common use case for the Messaging Bridge is to connect a PollableChannel to a SubscribableChannel, and when performing this role, the Messaging Bridge may also serve as a throttler:

 <bridge input-channel="pollable" output-channel="subscribable">
     <poller max-messages-per-poll="10">
         <interval-trigger interval="5" time-unit="SECONDS"/>
     </poller>
 </bridge>

Connecting Channel Adapters is just as easy. Here is a simple echo example between the "stdin" and "stdout" adapters from Spring Integration's "stream" namespace.

 <stream:stdin-channel-adapter id="stdin"/>

 <stream:stdout-channel-adapter id="stdout"/>

 <bridge id="echo" input-channel="stdin" output-channel="stdout"/>

Of course, the configuration would be similar for other (potentially more useful) Channel Adapter bridges, such as File to JMS, or Mail to File. The various Channel Adapters will be discussed in upcoming chapters.

Even though the MessageChannelTemplate is fairly straightforward, it does not hide the details of messaging from your application code. To support working with plain Objects instead of messages, Spring Integration provides SimpleMessagingGateway with the following methods:

 public void send(Object object) { ... }

 public Object receive() { ... }

 public Object sendAndReceive(Object object) { ... }

 Message<?> sendAndReceiveMessage(Object object);

It enables configuration of a request and/or reply channel and delegates to instances of the InboundMessageMapper and OutboundMessageMapper strategy interfaces.

 SimpleMessagingGateway gateway = new SimpleMessagingGateway(inboundMapper, outboundMapper);
 gateway.setRequestChannel(requestChannel);
 gateway.setReplyChannel(replyChannel);
 Object result = gateway.sendAndReceive("test");

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

 }

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 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 annotation. 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's 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 to 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 error. By default all error that have occurred downstream will be re-thrown as MessagingExeption (RuntimeException) upon Gateway's method invocation. However there are times when you may want to treat Exception as a valid reply, by mapping it to a Message. To accomplish this Gateway provides support for Exception mappers via 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"/>
				
				

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

Spring Integration provides two approaches: XML and Annotation-driven.

@Publisher(payload="#return", channel="testChannel", headers="bar='123',fname=#args.fname")
public String setName(String fname, String lname){
	return fname + " " + lname;
}

<bean class="org.springframework.integration.aop.PublisherAnnotationBeanPostProcessor"/>

<beans:bean id="testBean" class="org.foo.bar.TestBean" />
<aop:config>
	<aop:advisor advice-ref="interceptor" pointcut="bean(testBean)" />
</aop:config>
	
<publisher id="interceptor" default-channel="defaultChannel">
	<method pattern="echo" payload="'Echoing: ' + #return" headers="foo='bar'" channel="echoChannel"/>
	<method pattern="echoDef*" payload="#return"/>
	<method pattern="foo*"/>
</publisher>
	

<publisher id="anotherInterceptor"/>
	

Spring Integration's File support extends the Spring Integration Core with a dedicated vocabulary to deal with reading, writing, and transforming files. It provides a namespace that enables elements defining Channel Adapters dedicated to files and support for Transformers that can read file contents into strings or byte arrays.

This section will explain the workings of FileReadingMessageSource and FileWritingMessageHandler and how to configure them as beans. Also the support for dealing with files through file specific implementations of Transformer will be discussed. Finally the file specific namespace will be explained.

A FileReadingMessageSource can be used to consume files from the filesystem. This is an implementation of MessageSource that creates messages from a file system directory.

<bean id="pollableFileSource"
      class="org.springframework.integration.file.FileReadingMessageSource"
        p:inputDirectory="file:${input.directory}"/>

To prevent creating messages for certain files, you may supply a FileListFilter. By default, an AcceptOnceFileListFilter is used. This filter ensures files are picked up only once from the directory.

<bean id="pollableFileSource"
      class="org.springframework.integration.file.FileReadingMessageSource"
        p:inputDirectory="file:${input.directory}"
        p:filter-ref="customFilterBean"/>

A common problem with reading files is that a file may be detected before it is ready. The default AcceptOnceFileListFilter does not prevent this. In most cases, this can be prevented if the file-writing process renames each file as soon as it is ready for reading. A pattern-matching filter that accepts only files that are ready (e.g. based on a known suffix), composed with the default AcceptOnceFileListFilter allows for this. The CompositeFileListFilter enables the composition.

<bean id="pollableFileSource"
      class="org.springframework.integration.file.FileReadingMessageSource"
        p:inputDirectory="file:${input.directory}"
        p:filter-ref="compositeFilter"/>
<bean id="compositeFilter" class="org.springframework.integration.file.CompositeFileListFilter">
    <constructor-arg>
        <list>
            <bean class="org.springframework.integration.file.AcceptOnceFileListFilter" />
            <bean class="org.springframework.integration.file.PatternMatchingFileListFilter">
                <constructor-arg value="^test.*$"/>
            </bean>
        </list>
    </constructor-arg>
</bean>

The configuration can be simplified using the file specific namespace. To do this use the following template.

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

Within this namespace you can reduce the FileReadingMessageSource and wrap it in an inbound Channel Adapter like this:

 <file:inbound-channel-adapter id="filesIn"
       directory="file:${input.directory}" prevent-duplicates="true"/>
 					
 <file:inbound-channel-adapter id="filesIn"
        directory="file:${input.directory}" 
        filter="customFilterBean" />							

 <file:inbound-channel-adapter id="filesIn"
        directory="file:${input.directory}" 
        filename-pattern="^test.*$" /> 							

The first channel adapter is relying on the default filter that just prevents duplication, the second is using a custom filter, and the third is using the filename-pattern attribute to add a Pattern based filter to the FileReadingMessageSource. The file-name-pattern and filter attributes are mutually exclusive, but you can use a CompositeFileListFilter to use any combination of filters, including a pattern based filter to fit your particular needs.

When multiple processes are reading from the same directory it can be desirable to lock files to prevent them from being picked up concurrently. To do this you can use a FileLocker. There is a java.nio based implementation available out of the box, but it is also possible to implement your own locking scheme. The nio locker can be injected as follows

 <file:inbound-channel-adapter id="filesIn"
       directory="file:${input.directory}" prevent-duplicates="true">
    <file:nio-locker/>
  </file:inbound-channel-adapter>
            

A custom locker you can configure like this:

 <file:inbound-channel-adapter id="filesIn"
       directory="file:${input.directory}" prevent-duplicates="true">
    <file:locker ref="customLocker"/>
  </file:inbound-channel-adapter>
            

When filtering and locking files is not enough it might be needed to control the way files are listed entirely. To implement this type of requirement you can use an implementation of DirectoryScanner. This scanner allows you to determine entirely what files are listed each poll. This is also the interface that Spring Integration uses internally to wire FileListFilters FileLocker to the FileReadingMessageSource. A custom DirectoryScanner can be injected into the <file:inbound-channel-adapter/> on the scanner attribute.

 <file:inbound-channel-adapter id="filesIn"
       directory="file:${input.directory}" prevent-duplicates="true" scanner="customDirectoryScanner"/>
            

This gives you full freedom to choose the ordering, listing and locking strategies.

To write messages to the file system you can use a FileWritingMessageHandler. This class can deal with File, String, or byte array payloads. In its simplest form the FileWritingMessageHandler only requires a destination directory for writing the files. The name of the file to be written is determined by the handler's FileNameGenerator. The default implementation looks for a Message header whose key matches the constant defined as FileHeaders.FILENAME.

Additionally, you can configure the encoding and the charset that will be used in case of a String payload.

To make things easier you can configure the FileWritingMessageHandler as part of an outbound channel adapter using the namespace.

 <file:outbound-channel-adapter id="filesOut" directory="file:${input.directory.property}"/>

The namespace based configuration also supports a delete-source-files attribute. If set to true, it will trigger deletion of the original source files after writing to a destination. The default value for that flag is false.

 <file:outbound-channel-adapter id="filesOut"
             directory="file:${output.directory}" 
             delete-source-files="true"/>

In cases where you want to continue processing messages based on the written File you can use the outbound-gateway instead. It plays a very similar role as the outbound-channel-adapter. However after writing the File, it will also send it to the reply channel as the payload of a Message.

 <file:outbound-gateway id="mover" request-channel="moveInput"
            reply-channel="output"
            directory="${output.directory}" 
            delete-source-files="true"/>

If you have more elaborate requirements or need to support additional payload types as input to be converted to file content you could extend the FileWritingMessageHandler, but a much better option is to rely on a Transformer.

To transform data read from the file system to objects and the other way around you need to do some work. Contrary to FileReadingMessageSource and to a lesser extent FileWritingMessageHandler, it is very likely that you will need your own mechanism to get the job done. For this you can implement the Transformer interface. Or extend the AbstractFilePayloadTransformer for inbound messages. Some obvious implementations have been provided.

FileToByteArrayTransformer transforms Files into byte[]s using Spring's FileCopyUtils. It is often better to use a sequence of transformers than to put all transformations in a single class. In that case the File to byte[] conversion might be a logical first step.

FileToStringTransformer will convert Files to Strings as the name suggests. If nothing else, this can be useful for debugging (consider using with a Wire Tap).

To configure File specific transformers you can use the appropriate elements from the file namespace.

 <file-to-bytes-transformer input-channel="input" output-channel="output"
             delete-files="true"/>

 <file:file-to-string-transformer input-channel="input" output-channel="output
             delete-files="true" charset="UTF-8"/>

The delete-files option signals to the transformer that it should delete the inbound File after the transformation is complete. This is in no way a replacement for using the AcceptOnceFileListFilter when the FileReadingMessageSource is being used in a multi-threaded environment (e.g. Spring Integration in general).

The main function of an inbound Channel Adapter is to execute a SQL SELECT query and turn the result set into a message. The message payload is the whole result set, expressed as a List, and the types of the items in the list depends on the row-mapping strategy that is used. The default strategy is a generic mapper that just returns a Map for each row i nthe query. Optionally this can be changed by adding a reference to requires a reference to a RowMapper instance (see the Spring JDBC documentation for more detailed information about row mapping).

The inbound adapter also requires a reference to either JdbcTemplate instance or DataSource. The following example defines an inbound Channel Adapter with a DataSource reference.

<jdbc:inbound-channel-adapter query="select * from item where status=2" 
  channel="target" data-source="dataSource" 
  update="update item set status=10 where id in (:idList)" />

As well as the SELECT statement to generate the messages, the adapter above also has an UPDATE statement that is being used to mark the records as processed, so they don't show up in the next poll. The update is parameterised by the list of ids from the original select. This is done through a naming convention by default (a column in the input result set called "id" is translated into a list in the parameter map for the update called "idList"). To change the parameter generation strategy you can inject a SqlParameterSourceFactory into the adapter to override the default behaviour (the adapter has a sql-parameter-source-factory attribute).

<jdbc:inbound-channel-adapter query="..." 
  channel="target" data-source="dataSource" 
  update="...">
  <poller>
    <interval-trigger interval="1000"/>
    <transactional/>
  </poller>
</jdbc:inbound-channel-adapter>

The outbound Channel Adapter is the inverse of the inbound: its role is to handle a message and use it to execute a SQL query. The message payload and headers are available by default as input parameters to the query, for instance:

<jdbc:outbound-channel-adapter 
  query="insert into foos (id, status, name) values (:headers[$id], 0, :payload[foo])"
  channel="input" data-source="dataSource"/>

In the example above, messages arriving on the channel "input" have a payload of a map with key "foo", so the [] operator dereferences that value from the map. The headers are also accessed as a map.

The outbound adapter requires a reference to either a DataSource or a JdbcTemplate. It can also have a SqlParameterSourceFactory injected to control the binding of incoming message to the query.

If the input channel is a direct channel then the outbound adapter runs its query in the same thread, and therefor ethe same transaction (if there is one) as the sender of the message.

The JDBC module provides an implementation of the Spring Integration MessageStore (important in the Claim Check pattern) and MessageGroupStore (important in stateful patterns like Aggregator) backed by a database. Both interfaces are implemented by the JdbcMessageStore and there is also support for configuring store instances in XML. For example:

<jdbc:message-store id="messageStore" data-source="dataSource"/>

A JdbcTemplate can be specified instead of a DataSource.

Other optional attributes are show in the next example:

<jdbc:message-store id="messageStore" data-source="dataSource"
  lob-handler="lobHandler" table-prefix="MY_INT_"/>

Here we have specified a LobHandler for dealing with messages as large objects (e.g. often necessary if using Oracle) and a prefix for the table names in the queries generated by the store. The table name prefix defaults to "INT_".

The inbound Channel Adapter requires a reference to either a single JmsTemplate instance or both ConnectionFactory and Destination (a 'destinationName' can be provided in place of the 'destination' reference). The following example defines an inbound Channel Adapter with a Destination reference.

 <jms:inbound-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel">
     <integration:poller>
         <integration:interval-trigger interval="30" time-unit="SECONDS"/>
     </integration:poller>
 </jms:inbound-channel-adapter>

If 'extract-payload' is set to true (which is the default), the received JMS Message will be passed through the MessageConverter. When relying on the default SimpleMessageConverter, this means that the resulting Spring Integration Message will have the JMS Message's body as its payload. A JMS TextMessage will produce a String-based payload, a JMS BytesMessage will produce a byte array payload, and a JMS ObjectMessage's Serializable instance will become the Spring Integration Message's payload. If instead you prefer to have the raw JMS Message as the Spring Integration Message's payload, then set 'extract-payload' to false.

 <jms:inbound-channel-adapter id="jmsIn"
                              destination="inQueue"
                              channel="exampleChannel"
                              extract-payload="false"/>
     <integration:poller>
         <integration:interval-trigger interval="30" time-unit="SECONDS"/>
     </integration:poller>
 </jms:inbound-channel-adapter>

The "message-driven-channel-adapter" requires a reference to either an instance of a Spring MessageListener container (any subclass of AbstractMessageListenerContainer) or both ConnectionFactory and Destination (a 'destinationName' can be provided in place of the 'destination' reference). The following example defines a message-driven Channel Adapter with a Destination reference.

 <jms:message-driven-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel"/>

The 'extract-payload' property has the same effect as described above, and once again its default value is 'true'. The poller sub-element is not applicable for a message-driven Channel Adapter, as it will be actively invoked. For most usage scenarios, the message-driven approach is better since the Messages will be passed along to the MessageChannel as soon as they are received from the underlying JMS consumer.

The JmsSendingMessageHandler implements the MessageHandler interface and is capable of converting Spring Integration Messages to JMS messages and then sending to a JMS destination. It requires either a 'jmsTemplate' reference or both 'connectionFactory' and 'destination' references (again, the 'destinationName' may be provided in place of the 'destination'). As with the inbound Channel Adapter, the easiest way to configure this adapter is with the namespace support. The following configuration will produce an adapter that receives Spring Integration Messages from the "exampleChannel" and then converts those into JMS Messages and sends them to the JMS Destination reference whose bean name is "outQueue".

<jms:outbound-channel-adapter id="jmsOut" destination="outQueue" channel="exampleChannel"/>

As with the inbound Channel Adapters, there is an 'extract-payload' property. However, the meaning is reversed for the outbound adapter. Rather than applying to the JMS Message, the boolean property applies to the Spring Integration Message payload. In other words, the decision is whether to pass the Spring Integration Message itself as the JMS Message body or whether to pass the Spring Integration Message's payload as the JMS Message body. The default value is once again 'true'. Therefore, if you pass a Spring Integration Message whose payload is a String, a JMS TextMessage will be created. If on the other hand you want to send the actual Spring Integration Message to another system via JMS, then simply set this to 'false'.

Spring Integration's message-driven JMS inbound-gateway delegates to a MessageListener container, supports dynamically adjusting concurrent consumers, and can also handle replies. The inbound gateway requires references to a ConnectionFactory, and a request Destination (or 'requestDestinationName'). The following example defines a JMS "inbound-gateway" that receives from the JMS queue referenced by the bean id "inQueue" and sends to the Spring Integration channel named "exampleChannel".

 <jms:inbound-gateway id="jmsInGateway"
                      request-destination="inQueue"
                      request-channel="exampleChannel"/>

Since the gateways provide request/reply behavior instead of unidirectional send or receive, they also have two distinct properties for the "payload extraction" (as discussed above for the Channel Adapters' 'extract-payload' setting). For an inbound-gateway, the 'extract-request-payload' property determines whether the received JMS Message body will be extracted. If 'false', the JMS Message itself will become the Spring Integration Message payload. The default is 'true'.

Similarly, for an inbound-gateway the 'extract-reply-payload' property applies to the Spring Integration Message that is going to be converted into a reply JMS Message. If you want to pass the whole Spring Integration Message (as the body of a JMS ObjectMessage) then set this to 'false'. By default, it is also 'true' such that the Spring Integration Message payload will be converted into a JMS Message (e.g. String payload becomes a JMS TextMessage).

As with anything else, Gateway invocation might result in error. By default Producer will not be notified of the errors thta might have occurredon ythe consumer side and will time out waiting for the reply. However there might be times when you to communicate error condition back to the consumer, in other words treat the Exception as a valid reply valid reply by mapping it to a Message. To accomplish this JMS Inbound Gateway provides support for Exception mappers via exception-mapper attribute.

<int-jms:inbound-gateway request-destination="requestQueue" 
          request-channel="jmsinputchannel"
          exception-mapper="errorMessageMapper"/>
				
<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();
	}
		
}
				

The outbound Gateway creates JMS Messages from Spring Integration Messages and then sends to a 'request-destination'. It will then handle the JMS reply Message either by using a selector to receive from the 'reply-destination' that you configure, or if no 'reply-destination' is provided, it will create JMS TemporaryQueues. Notice that the "reply-channel" is also provided.

 <jms:outbound-gateway id="jmsOutGateway"
                       request-destination="outQueue"
                       request-channel="outboundJmsRequests"
                       reply-channel="jmsReplies"/>

The 'outbound-gateway' payload extraction properties are inversely related to those of the 'inbound-gateway' (see the discussion above). That means that the 'extract-request-payload' property value applies to the Spring Integration Message that is being converted into a JMS Message to be sent as a request, and the 'extract-reply-payload' property value applies to the JMS Message that is received as a reply and then converted into a Spring Integration Message to be subsequently sent to the 'reply-channel' as shown in the example configuration above.

If you need to convert the message, all JMS adapters and gateways, allow you to provide a MessageConverter via message-converter attribute. Simply provide the bean name of an instance of MessageConverter that is available within the same ApplicationContext. Also, to provide some consistency with Marshaller and Unmarshaller interfaces Spring provides MarshallingMessageConverter which you can configure with your own custom Marshallers and Unmarshallers

 <int-jms:inbound-gateway request-destination="requestQueue" 
                             request-channel="inbound-gateway-channel"
                             message-converter="marshallingMessageConverter"/>
                             
 <bean id="marshallingMessageConverter" 
       class="org.springframework.jms.support.converter.MarshallingMessageConverter">
   <constructor-arg>
      <bean class="org.bar.SampleMarshaller"/>
   </constructor-arg>
   <constructor-arg>
      <bean class="org.bar.SampleUnmarshaller"/>
   </constructor-arg>
 </bean>                                              
 

The Channel Adapters and Gateways featured above are all intended for applications that are integrating with other external systems. The inbound options assume that some other system is sending JMS Messages to the JMS Destination and the outbound options assume that some other system is receiving from the Destination. The other system may or may not be a Spring Integration application. Of course, when sending the Spring Integration Message instance as the body of the JMS Message itself (with the 'extract-payload' value set to false), it is assumed that the other system is based on Spring Integration. However, that is by no means a requirement. That flexibility is one of the benefits of using a Message-based integration option with the abstraction of "channels" or Destinations in the case of JMS.

There are cases where both the producer and consumer for a given JMS Destination are intended to be part of the same application, running within the same process. This could be accomplished by using a pair of inbound and outbound Channel Adapters. The problem with that approach is that two adapters are required even though conceptually the goal is to have a single Message Channel. A better option is supported as of Spring Integration version 2.0. Now it is possible to define a single "channel" when using the JMS namespace.

 <jms:channel id="jmsChannel" queue="exampleQueue"/>

The channel in the above example will behave much like a normal <channel/> element from the main Spring Integration namespace. It can be referenced by both "input-channel" and "output-channel" attributes of any endpoint. The difference is that this channel is backed by a JMS Queue instance named "exampleQueue". This means that asynchronous messaging is possible between the producing and consuming endpoints, but unlike the simpler asynchronous Message Channels created by adding a <queue/> sub-element within a non-JMS <channel/> element, the Messages are not just stored in an in-memory queue. Instead those Messages are passed within a JMS Message body, and the full power of the underlying JMS provider is then available for that channel. Probably the most common rationale for using this alternative would be to take advantage of the persistence made available by the store and forward approach of JMS messaging. If configured properly, the JMS-backed Message Channel also supports transactions. In other words, a producer would not actually write to a transactional JMS-backed channel if its send operation is part of a transaction that rolls back. Likewise, a consumer would not physically remove a JMS Message from the channel if the reception of that Message is part of a transaction that rolls back. Note that the producer and consumer transactions are separate in such a scenario. This is significantly different than the propagation of a transactional context across the simple, synchronous <channel/> element that has no <queue/> sub-element.

Since the example above is referencing a JMS Queue instance, it will act as a point-to-point channel. If on the other hand, publish/subscribe behavior is needed, then a separate element can be used, and a JMS Topic can be referenced instead.

 <jms:publish-subscribe-channel id="jmsChannel" topic="exampleTopic"/>

For either type of JMS-backed channel, the name of the destination may be provided instead of a reference.

 <jms:channel id="jmsQueueChannel" queue-name="exampleQueueName"/>

 <jms:publish-subscribe-channel id="jmsTopicChannel" topic-name="exampleTopicName"/>

In the examples above, the Destination names would be resolved by Spring's default DynamicDestinationResolver implementation, but any implementation of the DestinationResolver interface could be provided. Also, the JMS ConnectionFactory is a required property of the channel, but by default the expected bean name would be "connectionFactory". The example below provides both a custom instance for resolution of the JMS Destination names and a different name for the ConnectionFactory.

 <jms:channel id="jmsChannel" queue-name="exampleQueueName"
              destination-resolver="customDestinationResolver"
              connection-factory="customConnectionFactory"/>

To experiment with these JMS adapters, check out the samples available within the "samples/jms" directory in the distribution. There are two samples included. One provides inbound and outbound Channel Adapters, and the other provides inbound and outbound Gateways. They are configured to run with an embedded ActiveMQ process, but the "common.xml" file can easily be modified to support either a different JMS provider or a standalone ActiveMQ process. In other words, you can split the configuration so that the inbound and outbound adapters are running in separate JVMs. If you have ActiveMQ installed, simply modify the "brokerURL" property within the configuration to use "tcp://localhost:61616" for example (instead of "vm://localhost"). Both of the samples accept input via stdin and then echo back to stdout. Look at the configuration to see how these messages are routed over JMS.

To invoke a Web Service upon sending a message to a channel, there are two options - both of which build upon the Spring Web Services project: SimpleWebServiceOutboundGateway and MarshallingWebServiceOutboundGateway. The former will accept either a String or javax.xml.transform.Source as the message payload. The latter provides support for any implementation of the Marshaller and Unmarshaller interfaces. Both require a Spring Web Services DestinationProvider for determining the URI of the Web Service to be called.

 simpleGateway = new SimpleWebServiceOutboundGateway(destinationProvider);

 marshallingGateway = new MarshallingWebServiceOutboundGateway(destinationProvider, marshaller);

For more detail on the inner workings, see the Spring Web Services reference guide's chapter covering client access as well as the chapter covering Object/XML mapping.

To send a message to a channel upon receiving a Web Service invocation, there are two options again: SimpleWebServiceInboundGateway and MarshallingWebServiceInboundGateway. The former will extract a javax.xml.transform.Source from the WebServiceMessage and set it as the message payload. The latter provides support for implementation of the Marshaller and Unmarshaller interfaces. If the incoming web service message is a SOAP message the SOAP Action header will be added to the headers of the Message that is forwarded onto the request channel.

 simpleGateway = new SimpleWebServiceInboundGateway();
 simpleGateway.setRequestChannel(forwardOntoThisChannel);
 simpleGateway.setReplyChannel(listenForResponseHere); //Optional

 marshallingGateway = new MarshallingWebServiceInboundGateway(marshaller);
 //set request and optionally reply channel

Both gateways implement the Spring Web Services MessageEndpoint interface, so they can be configured with a MessageDispatcherServlet as per standard Spring Web Services configuration.

For more detail on how to use these components, see the Spring Web Services reference guide's chapter covering creating a Web Service. The chapter covering Object/XML mapping is also applicable again.

To configure an outbound Web Service Gateway, use the "outbound-gateway" element from the "ws" namespace:

<ws:outbound-gateway id="simpleGateway"
                     request-channel="inputChannel"
                     uri="http://example.org"/>

To set up an inbound Web Service Gateway, use the "inbound-gateway":

<ws:inbound-gateway id="simpleGateway"
                    request-channel="inputChannel"/>

To use Spring OXM Marshallers and/or Unmarshallers, provide bean references. For outbound:

<ws:outbound-gateway id="marshallingGateway"
                     request-channel="requestChannel"
                     uri="http://example.org"
                     marshaller="someMarshaller"
                     unmarshaller="someUnmarshaller"/>

And for inbound:

<ws:inbound-gateway id="marshallingGateway"
                    request-channel="requestChannel"
                    marshaller="someMarshaller"
                    unmarshaller="someUnmarshaller"/>

For either outbound gateway type, a "destination-provider" attribute can be specified instead of the "uri" (exactly one of them is required). You can then reference any Spring Web Services DestinationProvider implementation (e.g. to lookup the URI at runtime from a registry).

For either outbound gateway type, the "message-factory" attribute can also be configured with a reference to any Spring Web Services WebServiceMessageFactory implementation.

For the simple inbound gateway type, the "extract-payload" attribute can be set to false to forward the entire WebServiceMessage instead of just its payload as a Message to the request channel. This might be useful, for example, when a custom Transformer works against the WebServiceMessage directly.

This Chapter explains how to use RMI specific channel adapters to distribute a system over multiple JVMs. The first section will deal with sending messages over RMI. The second section shows how to receive messages over RMI. The last section shows how to define rmi channel adapters through the namespace support.

To send messages from a channel over RMI, simply define an RmiOutboundGateway. This gateway will use Spring's RmiProxyFactoryBean internally to create a proxy for a remote gateway. Note that to invoke a remote interface that doesn't use Spring Integration you should use a service activator in combination with Spring's RmiProxyFactoryBean.

To configure the outbound gateway write a bean definition like this:

	<bean id="rmiOutGateway" class=org.spf.integration.rmi.RmiOutboundGateway>
		<constructor-arg value="rmi://host"/>
		<property name="replyChannel" value="replies"/>
	</bean>
	

To receive messages over RMI you need to use a RmiInboundGateway. This gateway can be configured like this

 <bean id="rmiOutGateway" class=org.spf.integration.rmi.RmiInboundGateway>
     <property name="requestChannel" value="requests"/>
 </bean>
		

To configure the inbound gateway you can choose to use the namespace support for it. The following code snippet shows the different configuration options that are supported.

 <rmi:inbound-gateway id="gatewayWithDefaults" request-channel="testChannel"/>

 <rmi:inbound-gateway id="gatewayWithCustomProperties" request-channel="testChannel"
                      expect-reply="false" request-timeout="123" reply-timeout="456"/>

 <rmi:inbound-gateway id="gatewayWithHost" request-channel="testChannel"
                      registry-host="localhost"/>

 <rmi:inbound-gateway id="gatewayWithPort" request-channel="testChannel"
                      registry-port="1234"/>

 <rmi:inbound-gateway id="gatewayWithExecutorRef" request-channel="testChannel"
                      remote-invocation-executor="invocationExecutor"/>

To configure the outbound gateway you can use the namespace support as well. The following code snippet shows the different configuration for an outbound rmi gateway.

 <rmi:outbound-gateway id="gateway"
                       request-channel="localChannel"
                       remote-channel="testChannel"
                       host="localhost"/>

HttpInvoker is a Spring-specific remoting option that essentially enables Remote Procedure Calls (RPC) over HTTP. In order to accomplish this, an outbound representation of a method invocation is serialized using standard Java serialization and then passed within an HTTP POST request. After being invoked on the target system, the method's return value is then serialized and written to the HTTP response. There are two main requirements. First, you must be using Spring on both sides since the marshalling to and from HTTP requests and responses is handled by the client-side invoker and server-side exporter. Second, the Objects that you are passing must implement Serializable and be available on both the client and server.

While traditional RPC provides physical decoupling, it does not offer nearly the same degree of logical decoupling as a messaging-based system. In other words, both participants in an RPC-based invocation must be aware of a specific interface and specific argument types. Interestingly, in Spring Integration, the "parameter" being sent is a Spring Integration Message, and the interface is an internal detail of Spring Integration's implementation. Therefore, the RPC mechanism is being used as a transport so that from the end user's perspective, it is not necessary to consider the interface and argument types. It's just another adapter to enable messaging between two systems.

To receive messages over http you can use an HttpInvokerInboundGateway. Here is an example bean definition:

<bean id="inboundGateway"
      class="org.springframework.integration.httpinvoker.HttpInvokerInboundGateway">
    <property name="requestChannel" ref="requestChannel"/>
    <property name="replyChannel" ref="replyChannel"/>
    <property name="requestTimeout" value="30000"/>
    <property name="replyTimeout" value="10000"/>
</bean>

Because the inbound gateway must be able to receive HTTP requests, it must be configured within a Servlet container. The easiest way to do this is to provide a servlet definition in web.xml:

<servlet>
    <servlet-name>inboundGateway</servlet-name>
    <servlet-class>org.springframework.web.context.support.HttpRequestHandlerServlet</servlet-class>
</servlet>

Notice that the servlet name matches the bean name.

To configure the HttpInvokerOutboundGateway write a bean definition like this:

<bean id="outboundGateway"
      class="org.springframework.integration.httpinvoker.HttpInvokerOutboundGateway">
    <property name="replyChannel" ref="replyChannel"/>
</bean>

The outbound gateway is a MessageHandler and can therefore be registered with either a PollingConsumer or EventDrivenConsumer. The URL must match that defined by an inbound HttpInvoker Gateway as described in the previous section.

Spring Integration provides an "httpinvoker" namespace and schema definition. To include it in your configuration, simply provide the following URI within a namespace declaration: 'http://www.springframework.org/schema/integration/httpinvoker'. The schema location should then map to 'http://www.springframework.org/schema/integration/httpinvoker/spring-integration-httpinvoker-1.0.xsd'.

To configure the inbound gateway you can choose to use the namespace support for it. The following code snippet shows the different configuration options that are supported.

<httpinvoker:inbound-gateway id="inboundGateway"
                             request-channel="requestChannel"
                             request-timeout="10000" 
                             expect-reply="false"
                             reply-timeout="30000"/>

To configure the outbound gateway you can use the namespace support as well. The following code snippet shows the different configuration for an outbound HttpInvoker gateway. Only the 'url' and 'request-channel' are required.

<httpinvoker:outbound-gateway id="outboundGateway"
                              url="http://localhost:8080/example"
                              request-channel="requestChannel"
                              request-timeout="5000"
                              reply-channel="replyChannel"
                              reply-timeout="10000"/>

The HTTP support allows for the execution of HTTP requests and the processing of inbound HTTP requests. Because interaction over HTTP is always synchronous, even if all that is returned is a 200 status code, the HTTP support consists of two gateway implementations: HttpInboundEndpoint and HttpRequestExecutingMessageHandler.

To receive messages over HTTP you need to use an HttpInboundEndpoint. In common with the HttpInvoker support the Http Inbound Gateway needs to be deployed within a servlet container. The easiest way to do this is to provide a servlet definition in web.xml, see Section 23.2, “HttpInvoker Inbound Gateway” for further details. Below is an example bean definition for a simple HttpInboundEndpoint

<bean id="httpInbound" class="org.springframework.integration.http.HttpInboundEndpoint">
	<property name="requestChannel" ref="httpRequestChannel" />
	<property name="replyChannel" ref="httpReplyChannel" />	
</bean>

The HttpInboundEndpoint accepts an instance of InboundRequestMapper which allows customisation of the mapping from HttpServletRequest to Message. If none is provided an instance of DefaultInboundRequestMapper will be used. This encapsulates a simple strategy, which for example will create a String message for a POST request where the content type starts with "text", see the Javadoc for full details.

Starting with this release MultiPart File support was implemented. If the request has been wrapped as a MultipartHttpServletRequest, then the 'content type' can be checked. If it is known, and begins with "text", then the MultipartFile will be copied to a String in the parameter map. If the content type does not begin with "text", then the MultipartFile will be copied to a byte array within the parameter map instead.

In sending a response to the client there are a number of ways to customize the behavior of the gateway. By default the gateway will simply acknowledge that the request was received by sending a 200 status code back. It is possible to customize this response by providing an implementation of the Spring MVC View which will be invoked with the created Message. In the case that the gateway should expect a reply to the Message then setting the expectReply flag will cause the gateway to wait for a response Message before creating an Http response. Below is an example of a gateway configured to use a custom view and to wait for a response. It also shows how to customize the Http methods accepted by the gateway, which are POST and GET by default.

<bean id="httpInbound" class="org.springframework.integration.http.HttpInboundEndpoint">
	<property name="requestChannel" ref="httpRequestChannel" />
	<property name="replyChannel" ref="httpReplyChannel" />
	<property name="view" ref="jsonView" />
	<property name="supportedMethods" >
		<list>
			<value>GET</value>
			<value>DELETE</value>
		</list>
	</property>
	<property name="expectReply" value="true" />
	<property name="requestMapper" ref="customRequestMapper" />
</bean>

The message created from the request will be available in the Model map. The key that is used for that map entry by default is 'requestMessage', but this can be overridden by setting the 'requestKey' property on the endpoint's configuration.

To configure the HttpRequestExecutingMessageHandler write a bean definition like this:

<bean id="httpOutbound" class="org.springframework.integration.http.HttpRequestExecutingMessageHandler" >
	<constructor-arg value="http://localhost:8080/example" />
	<property name="outputChannel" ref="responseChannel" />
</bean>

This bean definition will execute HTTP requests by delegating to a RestTemplate. That template in turn delegates to a list of HttpMessageConverters to generate the HTTP request body from the Message payload. You can configure those converters as well as the ClientHttpRequestFactory instance to use:

<bean id="httpOutbound" class="org.springframework.integration.http.HttpRequestExecutingMessageHandler" >
	<constructor-arg value="http://localhost:8080/example" />
	<property name="outputChannel" ref="responseChannel" />
	<property name="messageConverters" ref="messageConverterList" />
	<property name="requestFactory" ref="customRequestFactory" />
</bean>

By default the HTTP request will be generated using an instance of SimpleClientHttpRequestFactory which uses the JDK HttpURLConnection. Use of the Apache Commons HTTP Client is also supported through the provided CommonsClientHttpRequestFactory which can be injected as shown above.

Spring Integration provides an "http" namespace and schema definition. To include it in your configuration, simply provide the following URI within a namespace declaration: 'http://www.springframework.org/schema/integration/http'. The schema location should then map to 'http://www.springframework.org/schema/integration/http/spring-integration-http.xsd'.

To configure an inbound http channel adapter which is an instance of HttpInboundEndpoint configured not to expect a response.

 <http:inbound-channel-adapter id="httpChannelAdapter" channel="requests" supported-methods="PUT, DELETE"/>

To configure an inbound http gateway which expects a response.

 <http:inbound-gateway id="inboundGateway" request-channel="requests" reply-channel="responses"/>

To configure the outbound gateway you can use the namespace support as well. The following code snippet shows the different configuration options for an outbound Http gateway. Most importantly, notice that the 'http-method' and 'expected-response-type' are provided. Those are two of the most commonly configured values. The default http-method is POST, and the default response type is null. With a null response type, the payload of the reply Message would only contain the status code (e.g. 200) as long as it's a successful status (non-successful status codes will throw Exceptions). If you are expecting a different type, such as a String, then provide that fully-qualified class name as shown below.

<http:outbound-gateway id="example"
		request-channel="requests"
		url="http://localhost/test"
		http-method="POST"
		extract-request-payload="false"
		expected-response-type="java.lang.String"
		charset="UTF-8"
		request-factory="requestFactory"
		request-timeout="1234"
		reply-channel="replies"/>

If your outbound adapter is to be used in a unidirectional way, then you can use an outbound-channel-adapter instead. This means that a successful response will simply execute without sending any Messages to a reply channel. In the case of any non-successful response status code, it will throw an exception. The configuration looks very similar to the gateway:

<http:outbound-channel-adapter id="example"
			url="http://localhost/example"
			http-method="GET"
			channel="requests"
			charset="UTF-8"
			extract-payload="false"
			expected-response-type="java.lang.String"
			request-factory="someRequestFactory"
			order="3"
			auto-startup="false"/>

Two flavors each of UDP inbound and outbound adapters are provided UnicastSendingMessageHandler sends a datagram packet to a single destination. UnicastReceivingChannelAdapter receives incoming datagram packets. MulticastSendingMessageHandler sends (broadcasts) datagram packets to a multicast address. MulticastReceivingChannelAdapter receives incoming datagram packets by joining to a multicast address.

TCP inbound and outbound adapters are provided TcptSendingMessageHandler sends messages over TCP. TcpNetReceivingChannelAdapter receives messages over TCP.

A simple inbound TCP gateway is provided; this allows for simple request/response processing. While the gateway can support any number of connections, each connection can only process serially. The thread that reads from the socket waits for, and sends, the response before reading again.

A simple outbound TCP gateway is provided; this allows for simple request/response processing. Each request is processed serially. The calling thread blocks on the socket until either a response is received or an I/O error occurs. Requests are single-threaded over the socket.

 <ip:outbound-channel-adapter id="udpOut"
		protocol="udp"
		host="somehost"
		port="11111"
		multicast="false"
		channel="exampleChannel" />

A simple UDP outbound channel adapter.

UDP is an efficient, but unreliable protocol. Two attributes are added to improve reliability. When check-length is set to true, the adapter precedes the message data with a length field (4 bytes in network byte order). This enables the receiving side to verify the length of the packet received. If a receiving system uses a buffer that is too short the contain the packet, the packet can be truncated. The length header provides a mechanism to detect this.

  <ip:outbound-channel-adapter id="udpOut"
		protocol="udp"
		host="somehost"
		port="11111"
		multicast="false"
		check-length="true"
		channel="exampleChannel" />

An outbound channel adapter that adds length checking to the datagram packets.

The second reliability improvement allows an application-level acknowledgment protocol to be used. The receiver must send an acknowledgment to the sender within a specified time.

  <ip:outbound-channel-adapter id="udpOut"
		protocol="udp"
		host="somehost"
		port="11111"
		multicast="false"
		check-length="true"
		acknowledge="true"
		ack-host="thishost"
		ack-port="22222"
		ack-timeout="10000"
		channel="exampleChannel" />

An outbound channel adapter that adds length checking to the datagram packets and waits for an acknowledgment.

For even more reliable networking, TCP can be used.

  <ip:inbound-channel-adapter id="udpReceiver"
		channel="udpOutChannel"
		protocol="udp"
		port="11111"
		receive-buffer-size="500"
		multicast="false"
		check-length="true" />

A basic unicast inbound udp channel adapter.

  <ip:inbound-channel-adapter id="udpReceiver"
		channel="udpOutChannel"
		protocol="udp"
		port="11111"
		receive-buffer-size="500"
		multicast="true"
		multicast-address="225.6.7.8"
		check-length="true" />

A basic multicast inbound udp channel adapter.

For TCP, the configuration of the underlying connection is provided using a Connection Factory. Two types of connection factory are provided; a client connection factory and a server connection factory. Client connection factories are used to establish outgoing connections; Server connection factories listen for incoming connections.

A client connection factory is used by an outbound channel adapter but a reference to a client connection factory can also be provided to an inbound channel adapter and that adapter will receive any incoming messages received on connections created by the outbound adapter.

A server connection factory is used by an inbound channel adapter (in fact the connection factory will not function without one). A reference to a server connection factory can also be provided to an outbound adapter; that adapter can then be used to send replies to incoming messages to the same connection.

A maximum of one adapter of each type may be given a reference to a connection factory.

Connection factories using java.net.Socket and java.nio.channel.SocketChannel are provided.

	<ip:tcp-connection-factory id="server"
		type="server"
		port="1234"
	/>

A simple server connection factory that uses java.net.Socket connections.

	<ip:tcp-connection-factory id="server"
		type="server"
		port="1234"
                using-nio="true"
	/>

A simple server connection factory that uses java.nio.channel.SocketChannel connections.

	<ip:tcp-connection-factory id="client"
		type="client"
		host="localhost"
		port="1234"
		single-use="true"
		so-timeout="10000"
	/>

A client connection factory that uses java.net.Socket connections and creates a new connection for each message.

	<ip:tcp-connection-factory id="client"
		type="client"
		host="localhost"
		port="1234"
		single-use="true"
		so-timeout="10000"
                using-nio=true
	/>

A client connection factory that uses java.nio.channel.Socket connections and creates a new connection for each message.

TCP is a streaming protocol; this means that some structure has to be provided to data transported over TCP, so the receiver can demarcate the data into discrete messages. Connection factories are configured to use converters to convert between the message payload and the bits that are sent over TCP. This is accomplished by providing an input converter and output converter for inbound and outbound messages respectively. Four standard converters are provided; the first is ByteArrayCrlfConverter, which can convert a String or byte array to a stream of bytes followed by carriage return and linefeed characters (\r\n). This is the default converter and can be used with telnet as a client, for example. The second is is ByteArrayStxEtxConverter, which can convert a String or byte array to a stream of bytes preceded by an STX (0x02) and followed by an ETX (0x03). The third is ByteArrayLengthHeaderConverter, which can convert a String or byte array to a stream of bytes preceded by a 4 byte binary length in network byte order. Each of these converts an input stream containing the corresponding format to a byte array payload. The fourth converter is JavaSerializationConverter which can be used to convert any Serializable objects. We expect to provide other serialization technologies but you may also supply your own by implementing the InputStreamingConverter and OutputStreamingConverter interfaces. If you do not wish to use the default converters, you must supply input-converter and output-converter attributes on the connection factory (example below).

        <bean id="serial" 
              class="org.springframework.commons.serializer.JavaSerializationConverter" />

	<ip:tcp-connection-factory id="server"
		type="server"
		port="1234"
                input-converter="serial"
                output-converter="serial"
	/>

A server connection factory that uses java.net.Socket connections and uses Java serialization on the wire.

For a full reference of the attributes available on connection factories, see the reference at the end of this section.

TCP inbound and outbound channel adapters that utilize the above connection factories are provided. These adapters have just 2 attributes connection-factory and channel. The channel attribute specifies the channel on which messages arrive at an outbound adapter and on which messages are placed by an inbound adapter. The connection factory indicates which connection factory is to be used to manage connections for the adapter. While both inbound and outbound adapters can share a connection factory, server connection factories are always 'owned' by an inbound adapter; client connection factories are always 'owned' by an outbound adapter. One, and only one, adapter of each type may get a reference to a connection factory.

	<bean id="serializer" class="org.springframework.commons.serializer.JavaSerializationConverter" />

	<int-ip:tcp-connection-factory id="server"
		type="server"
		port="1234"
		input-converter="serializer"
		output-converter="serializer"
		using-nio="true"
		single-use="true"
	/>

	<int-ip:tcp-connection-factory id="client"
		type="client"
		host="localhost"
		port="#{server.port}"
		single-use="true"
		so-timeout="10000"
		input-converter="serializer"
		output-converter="serializer"
	/>
	
	<int:channel id="input" />
	
	<int:channel id="replies">
		<int:queue/>
	</int:channel>

	<int-ip:tcp-outbound-channel-adapter id="outboundClient"
		channel="input"
		connection-factory="client"/>
		
	<int-ip:tcp-inbound-channel-adapter id="inboundClient"
		channel="replies"
		connection-factory="client"/>		
	
	<int-ip:tcp-inbound-channel-adapter id="inboundServer"
		channel="loop"
		connection-factory="server"/>
		
	<int-ip:tcp-outbound-channel-adapter id="outboundServer"
		channel="loop"
		connection-factory="server"/>

	<int:channel id="loop" />

In this configuration, messages arriving in channel 'input' are serialized over connections created by 'client' received at the server and placed on channel 'loop'. Since 'loop' is the input channel for 'outboundServer' the message is simply looped back over the same connection and received by 'inboundClient' and deposited in channel 'replies'.

Two versions of TCP inbound and outbound channel adapters are provided; these adapters use either java.net.Socket IO, or java.nio.channels.SocketChannel IO. The choice of which to use depends on the application. The TcpNet* adapters use java.net.Socket and the TcpNio* adapters use java.nio.channels.ChannelSocket. It is not anticipated that much difference in performance, if any, would exist between these technologies on the outbound side. This is because each outbound adapter sends data over only one socket. On the receiving side however, consideration should be given to the number of connections. For the TcpNetReceivingChannelAdapter a thread is dedicated to receiving data on each connected socket; the pool size must therefore be set large enough to handle the expected number of connections. For the TcpNioReceivingChannelAdapter threads are used on an as-needed basis and it is likely that many fewer threads would be needed. If a small number of connections is expected, we expect that the the TcpNetReceivingChannelAdapter will give the best performance. For large number of connections, the TcpNioReceivingChannelAdapter will likely give the best performance. In addition, the TcpNioReceivingChannelAdapter provides an attribute using-direct-buffers which attempts to use direct buffers. See java.nio.ByteBuffer for more information about direct buffers.

TCP is a streaming protocol; this means that some structure has to be provided to data transported over TCP, so the receiver can demarcate the data into discrete messages. Four standard message formats are provided for this purpose; you can also provide code for your own custom format. The first of the four standard formats is 'length-header', in which case a 4 byte length header precedes the data; this is the default. The second is 'stx-etx' in which the message data is preceded by an STX (0x02) character and terminated with an ETX (0x03) character. The third is 'crlf' in which the message is terminated with a carriage return and line feed (\r\n). These three formats require a byte array or String payload outbound endpoints; inbound endpoints produce messages with byte array payloads. The fourth format is 'serialized' wherby standard java serialization is used; payloads must implement Serializable. For the simple formats, the first (the default) is likely to be the most performant. This is because we can determine exactly how many bytes to read to obtain the complete message. The other two formats require examining each byte to determine if the end of the message has been received. The length-header format can also handle binary data. The other two formats can only handle text data (specifcally, data that does not contain characters 0x02 and 0x03 for stx-etx and 0x0d and 0x0a for crlf). This limitation can be avoided by appropriate character escaping techniques in the application layer. No such escaping is provided by the adapters; therefore it is not recommened that these formats be used without some transformation if the data may contain these characters.

 <ip:outbound-channel-adapter id="tcpOut"
		channel="inChannel"
		protocol="tcp"
		host="somehost"
		port="11111"
		message-format="length-header"
		using-nio="true"
		using-direct-buffers="false"
		so-keep-alive="true"
		so-timeout="10000"
		/>

A basic outbound tcp channel adapter. This adapter uses java.nio.channels.SocketChannel. To use a java.net.Socket, set using-nio to false and using-direct-buffers is not relevant.

	<ip:inbound-channel-adapter id="tcp1"
		channel="channel"
		protocol="tcp"
		port="11111"
		message-format="length-header"
		using-nio="true"
		using-direct-buffers="false"
		pool-size="2"
		so-keep-alive="true"
		so-timeout="10000"
		/>

A basic inbound tcp channel adapter. This adapter uses java.nio.channels.SocketChannel. To use a java.net.Socket, set using-nio to false and using-direct-buffers is not relevant.

The simple inbound TCP gateway SimpleTcpNetInboundGateway and simple oubound TCP gateway SimpleTcpNetOutboundGateway use java.net.Socket for communications. Each connection can process a single request/response at a time.

The inbound gateway delegates to a subclass of the TcpNetReceivingChannelAdapter described above, so please read that section for more information. After constructing a message with the incoming payload and sending it to the requestChannel, it waits for a response and sends the payload from the response message by writing it to the socket, using the same message format configured for the incoming message.

The outbound gateway delegates to a TcpNetSendingMessageHandler described above, so please read that section for more information. After sending a message over the socket, the thread waits for a response and constructs a response message with a byte[] payload The incoming response is decoded using the same message format configured for the outgoing message. Communications over the socket are single-threaded. Users should be aware that only one message can be handled at a time and if another thread attempts to send a message before the current response has been received, it will block. Only when the inprocess message receives a response (or times out based on the socket timeout option) will it proceed. If an error occurs while reading the response, the socket will be closed, regardless of the close attribute.

 	<ip:inbound-gateway id="gatewayCrLf"
		port="1234"
		request-channel="someChannel"
		message-format="crlf"
        />

A simple inbound TCP gateway; it uses '/r/n' delimited data and can be used by a simple client such as telnet.

 	<ip:outbound-gateway id="simpleOutGateway"
		request-channel="tcpChannel"
		message-format="crlf"
		host="localhost"
		port="1234"
		/>

A simple oubound TCP gateway; it uses '/r/n' delimited data.

Spring Integration provides support for outbound email with the MailSendingMessageHandler. It delegates to a configured instance of Spring's JavaMailSender:

 JavaMailSender mailSender = (JavaMailSender) context.getBean("mailSender");

 MailSendingMessageHandler mailSendingHandler = new MailSendingMessageHandler(mailSender);

MailSendingMessageHandler has various mapping strategies that use Spring's MailMessage abstraction. If the received Message's payload is already a MailMessage instance, it will be sent directly. Therefore, it is generally recommended to precede this consumer with a Transformer for non-trivial MailMessage construction requirements. However, a few simple Message mapping strategies are supported out-of-the-box. For example, if the message payload is a byte array, then that will be mapped to an attachment. For simple text-based emails, you can provide a String-based Message payload. In that case, a MailMessage will be created with that String as the text content. If you are working with a Message payload type whose toString() method returns appropriate mail text content, then consider adding Spring Integration's ObjectToStringTransformer prior to the outbound Mail adapter (see the example within Section 9.2, “The <transformer> Element” for more detail).

The outbound MailMessage may also be configured with certain values from the MessageHeaders. If available, values will be mapped to the outbound mail's properties, such as the recipients (TO, CC, and BCC), the from/reply-to, and the subject. The header names are defined by the following constants:

 MailHeaders.SUBJECT
 MailHeaders.TO
 MailHeaders.CC
 MailHeaders.BCC
 MailHeaders.FROM
 MailHeaders.REPLY_TO

Spring Integration also provides support for inbound email with the MailReceivingMessageSource. It delegates to a configured instance of Spring Integration's own MailReceiver interface, and there are two implementations: Pop3MailReceiver and ImapMailReceiver. The easiest way to instantiate either of these is by passing the 'uri' for a Mail store to the receiver's constructor. For example:

 MailReceiver receiver = new Pop3MailReceiver("pop3://usr:pwd@localhost/INBOX");

Another option for receiving mail is the IMAP "idle" command (if supported by the mail server you are using). Spring Integration provides the ImapIdleChannelAdapter which is itself a Message-producing endpoint. It delegates to an instance of the ImapMailReceiver but enables asynchronous reception of Mail Messages. There are examples in the next section of configuring both types of inbound Channel Adapter with Spring Integration's namespace support in the 'mail' schema.

Spring Integration provides a namespace for mail-related configuration. To use it, configure the following schema locations.

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

To configure an outbound Channel Adapter, provide the channel to receive from, and the MailSender:

<mail:outbound-channel-adapter channel="outboundMail"
                              mail-sender="mailSender"/>

Alternatively, provide the host, username, and password:

<mail:outbound-channel-adapter channel="outboundMail"
                              host="somehost" username="someuser" password="somepassword"/>

To configure an inbound Channel Adapter, you have the choice between polling or event-driven (assuming your mail server supports IMAP IDLE - if not, then polling is the only option). A polling Channel Adapter simply requires the store URI and the channel to send inbound Messages to. The URI may begin with "pop3" or "imap":

 <mail:inbound-channel-adapter channel="mailIn"
                               store-uri="imap://usr:[email protected]/INBOX">
     <poller max-messages-per-poll="3">
         <interval-trigger interval="30" time-unit="SECONDS"/>
     </poller>
 </mail:inbound-channel-adapter>

If you do have IMAP idle support, then you may want to configure the "imap-idle-channel-adapter" element instead. Since the "idle" command enables event-driven notifications, no poller is necessary for this adapter. It will send a Message to the specified channel as soon as it receives the notification that new mail is available:

 <mail:imap-idle-channel-adapter channel="mailIn"
                                 store-uri="imaps://usr:[email protected]:993/INBOX"/>

When using the namespace support, a header-enricher Message Transformer is also available. This simplifies the application of the headers mentioned above to any Message prior to sending to the Mail-sending Channel Adapter.

<mail:header-enricher subject="Example Mail"
                     to="[email protected]"
                     cc="[email protected]"
                     bcc="[email protected]"
                     from="[email protected]"
                     reply-to="[email protected]"
                     overwrite="false"/>

The Notification-listening Channel Adapter requires a JMX ObjectName for the MBean that publishes Notifications to which this listener should be registered. A very simple configuration might look like this:

 <jmx:notification-listening-channel-adapter id="adapter"
                   channel="channel"
                   object-name="example.domain:name=publisher"/>

The adapter can also accept a reference to a NotificationFilter and a "handback" Object to provide some context that is passed back with each Notification. Both of those attributes are optional. Extending the above example to include those attributes as well as an explicit MBeanServer bean name would produce the following:

 <jmx:notification-listening-channel-adapter id="adapter"
                   channel="channel"
                   mbean-server="someServer"
                   object-name="example.domain:name=somePublisher"
                   notification-fliter="notificationFilter"
                   handback="myHandback"/>

Since the notification-listening adapter is registered with the MBeanServer directly, it is event-driven and does not require any poller configuration.

The Notification-publishing Channel Adapter is relatively simple. It only requires a JMX ObjectName in its configuration as shown below.

 <context:mbean:export/>

 <jmx:notification-publishing-channel-adapter id="adapter" 
                              channel="channel"
                              object-name="example.domain:name=publisher"/>

It does also require that an MBeanExporter be present in the context. That is why the <context:mbean-export/> element is shown above as well.

When Messages are sent to the channel for this adapter, the Notification is created from the Message content. If the payload is a String it will be passed as the "message" text for the Notification. Any other payload type will be passed as the "userData" of the Notification.

JMX Notifications also have a "type", and it should be a dot-delimited String. There are two ways to provide the type. Precedence will always be given to a Message header value associated with the JmxHeaders.NOTIFICATION_TYPE key. On the other hand, you can rely on a fallback "default-notification-type" attribute provided in the configuration.

 <context:mbean:export/>

 <jmx:notification-publishing-channel-adapter id="adapter" 
                              channel="channel"
                              object-name="example.domain:name=publisher"
                              default-notification-type="some.default.type"/>

The attribute polling adapter is useful when you have a requirement to periodically check on some value that is available through an MBean as a managed attribute. The poller can be configured in the same way as any other polling adapter in Spring Integration (or it's possible to rely on the default poller). The "object-name" and "attribute-name" are required. An MBeanServer reference is also required, but it will automatically check for a bean named "mbeanServer" by default just like the notification-listening-channel-adapter described above.

 <jmx:attribute-polling-channel-adapter id="adapter"
                channel="channel"
                object-name="example.domain:name=someService"
                attribute-name="InvocationCount">
     <si:poller max-messages-per-poll="1">
         <si:interval-trigger interval="5000"/>
     </si:poller>
 </jmx:attribute-polling-channel-adapter>

The operation-invoking-channel-adapter enables Message-driven invocation of any managed operation exposed by an MBean. Each invocation requires the operation name to be invoked and the ObjectName of the target MBean. Both of these must be explicitly provided via adapter configuration:

 <jmx:operation-invoking-channel-adapter id="adapter"
                object-name="example.domain:name=TestBean"
                operation-name="ping"/>

Then the adapter only needs to be able to discover the "mbeanServer" bean. If a different bean name is required, then provide the "mbean-server" attribute with a reference.

The payload of the Message will be mapped to the parameters of the operation, if any. A Map-typed payload with String keys is treated as name/value pairs whereas a List or array would be passed as a simple argument list (with no explicit parameter names). If the operation requires a single parameter value, then the payload can represent that single value, and if the operation requires no parameters, then the payload would be ignored.

If you want to expose a channel for a single common operation to be invoked by Messages that need not contain headers, then that option works well.

Similar to operation-invoking-channel-adapter Spring Integration also provides operation-invoking-outbound-gateway which could be used when dealing with non-void operations and return value is required. Such return value will be sent as message payload to the 'reply-channel' specified by this Gateway.

 <jmx:operation-invoking-outbound-gateway request-channel="requestChannel"
			reply-channel="replyChannel"
			object-name="org.springframework.integration.jmx.config:type=TestBean,name=testBeanGateway"
			operation-name="testWithReturn"/>

Another way of provideing the 'reply-channel' is by setting MessageHeaders.REPLY_CHANNEL Message Header

Spring Integration components themselves may be exposed as MBeans when the Control Bus is configured. 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. Internally, the Control Bus uses a Spring MBeanExporter instance to expose the various endpoints and channels. To create an instance of the Control Bus, define a bean and provide a reference to an MBeanServer and a domain name (we will be providing namespace support). The domain can be left out in which case the default domain is "org.springframework.integration".

 <bean id="controlBus" class="org.springframework.integration.control.ControlBus">
     <constructor-arg ref="mbeanServer"/>
     <constructor-arg value="example.domain"/>
 </bean>

 <bean id="mbeanServer" class="org.springframework.jmx.support.MBeanServerFactoryBean">
     <property name="locateExistingServerIfPossible" value="true"/>
 </bean>

The Control Bus has an "operationChannel" that can be accessed for invoking operations on the MBeans that it has exported. This will also be covered by namespace support soon to make it easier to configure references to that channel for other producers. We will likely add some other channels for notifications and attribute polling as well.

The Control Bus functionality is a work in progress. At this time, one can perform some basic monitoring of Message Channels and/or invoke Lifecycle operations (start/stop) on Message Endpoints. Now that the foundation is available, however, we will be able to extend the attributes and operations that are being exposed.

Spring Integration provides adapters for sending and receiving both XMPP messages and status changes from other entries in your roster as well as XMPP.

XMPP describes a way for multiple agents to communicate with each other in a distributed system. The canonical use case is to send and receive instant messages, though XMPP can be, and is, used for far more applications. XMPP is used to describe a network of actors. Within that network, actors may address each other directly, as well as broadcast status changes.

XMPP provides the messaging fabric that underlies some of the biggest Instant Messaging networks in the world, including Google Talk (GTalk) - which is also available from within GMail - and Facebook Chat. There are many good open-source XMPP servers available. Two popular implementations are Openfire and ejabberd .

In XMPP, rosters (the roster corresponds to the notion of a "buddy list" in your typical IM client) are used to manage a list of other agents ("contacts", or "buddies", in an IM client) in the system, calledroster items. The roster item contains - at a minimum - the roster item's JID which is its unique ID on the network. An actor may subscribe to the state changes of another actor in the system. The subscription can be bidirectional, as well. The subscription settings determine whose status updates are broadcast, and to whom. These subscriptions are stored on the XMPP server, and are thus durable.

Using the Spring Integration XMPP namespace support is simple. Its use is like any other module in the Spring framework: import the XML schema, and use it to define elements. A prototypical XMPP-based integration might feature the following header. We won't repeat this in subsequent examples, because it is uninteresting.

<?xml version="1.0" encoding="UTF-8"?>

<beans:beans
    xmlns="http://www.springframework.org/schema/integration"
    xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:p="http://www.springframework.org/schema/p"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:util="http://www.springframework.org/schema/util"
    xmlns:xmpp="http://www.springframework.org/schema/integration/xmpp"
    xmlns:tool="http://www.springframework.org/schema/tool"
    xmlns:lang="http://www.springframework.org/schema/lang"
    xsi:schemaLocation="
http://www.springframework.org/schema/integration/xmpp
http://www.springframework.org/schema/integration/xmpp/spring-integration-xmpp.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.0.xsd
">
    ...

</beans:beans>

To participate in the network, an actor must connect to an XMPP server. Typically this requires - at a minimum - a user, apassword, ahost, and aport. To create an XMPP connection, you may use the XML namespace.

<xmpp:xmpp-connection
  id="myConnection"
  user="user"
  password="password"
  host="host"
  port="port"
  resource="theNameOfTheResource"
  subscription-mode="accept_all"
/>
      

XMPP also supports broadcasting state. You can use this capability to let people who have you on their roster see your state changes. This happens all the time with your IM clients - you change your away status, and then set an away message, and everybody who has you on their roster sees your icon or username change to reflect this new state, and additionally might see your new "away" message. If you would like to receive notification, or notify others, of state changes, you can use Spring Integration's "presence" adapters.

The most important data for these adapters resides in the headers. The header keys are enumerated on the org.springframework.integration.xmpp.XmppHeaders class. The header keys specific to these "presence" adapters start with the token "PRESENCE_". Not all headers are available for both inbound and outbound.

In many cases application data is obtained from a stream. It is not recommended to send a reference to a Stream as a message payload to a consumer. Instead messages are created from data that is read from an input stream and message payloads are written to an output stream one by one.

Spring Integration provides two adapters for streams. Both ByteStreamReadingMessageSource and CharacterStreamReadingMessageSource implement MessageSource. By configuring one of these within a channel-adapter element, the polling period can be configured, and the Message Bus can automatically detect and schedule them. The byte stream version requires an InputStream, and the character stream version requires a Reader as the single constructor argument. The ByteStreamReadingMessageSource also accepts the 'bytesPerMessage' property to determine how many bytes it will attempt to read into each Message. The default value is 1024

<bean class="org.springframework.integration.stream.ByteStreamReadingMessageSource">
	<constructor-arg ref="someInputStream"/>
	<property name="bytesPerMessage" value="2048"/>
</bean>

<bean class="org.springframework.integration.stream.CharacterStreamReadingMessageSource">
	<constructor-arg ref="someReader"/>
</bean>
		

For target streams, there are also two implementations: ByteStreamWritingMessageHandler and CharacterStreamWritingMessageHandler. Each requires a single constructor argument - OutputStream for byte streams or Writer for character streams, and each provides a second constructor that adds the optional 'bufferSize'. Since both of these ultimately implement the MessageHandler interface, they can be referenced from a channel-adapter configuration as described in more detail in Chapter 6, Channel Adapter.

<bean class="org.springframework.integration.stream.ByteStreamWritingMessageHandler">
	<constructor-arg ref="someOutputStream"/>
	<constructor-arg value="1024"/>
</bean>

<bean class="org.springframework.integration.stream.CharacterStreamWritingMessageHandler">
	<constructor-arg ref="someWriter"/>
</bean>
	

To reduce the configuration needed for stream related channel adapters there is a namespace defined. The following schema locations are needed to use it.

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

To configure the inbound channel adapter the following code snippet shows the different configuration options that are supported.

<stdin-channel-adapter id="adapterWithDefaultCharset"/>

<stdin-channel-adapter id="adapterWithProvidedCharset" charset="UTF-8"/>

To configure the outbound channel adapter you can use the namespace support as well. The following code snippet shows the different configuration for an outbound channel adapters.

<stdout-channel-adapter id="stdoutAdapterWithDefaultCharset" channel="testChannel"/>

<stdout-channel-adapter id="stdoutAdapterWithProvidedCharset" charset="UTF-8" channel="testChannel"/>

<stderr-channel-adapter id="stderrAdapter" channel="testChannel"/>

<stdout-channel-adapter id="newlineAdapter" append-newline="true" channel="testChannel"/>
		

To receive events and send them to a channel, simply define an instance of Spring Integration's ApplicationEventListeningChannelAdapter. This class is an implementation of Spring's ApplicationListener interface. By default it will pass all received events as Spring Integration Messages. To limit based on the type of event, configure the list of event types that you want to receive with the 'eventTypes' property.

For convenience namespace support was provided to configure ApplicationEventListeningChannelAdapter via inbound-channel-adapter

<int-event:inbound-channel-adapter channel="input"/>
	
<int:publish-subscribe-channel id="sampleEventChannel"/>

In the above sample, all Application Context events will be delivered as Spring Integration Messages to 'sampleEventChannel'.

To send Spring ApplicationEvents, create an instance of the ApplicationEventPublishingMessageHandler and register it within an endpoint. This implementation of the MessageHandler interface also implements Spring's ApplicationEventPublisherAware interface and thus acts as a bridge between Spring Integration Messages and ApplicationEvents.

For convenience namespace support was provided to configure ApplicationEventPublishingMessageHandler via outbound-channel-adapter element

<int:channel id="input"/>
	
<int-event:outbound-channel-adapter channel="input"/>

If you are using PollableChannel (e.g., Queue), you can also provide poller as sub-element of outbound-channel-adapter, optionally providing task-executor

<int:channel id="input">
	<int:queue/>
</int:channel>
	
<int-event:outbound-channel-adapter channel="input">
	<int:poller max-messages-per-poll="1" task-executor="executor">
		<int:interval-trigger interval="100" time-unit="MILLISECONDS"/>
	</int:poller>
</int-event:outbound-channel-adapter>

<task:executor id="executor" pool-size="5"/>

In the above sample, all messages sent to an 'input' channel will be published as ApplicationEvents to Spring Application sContext

Spring Integration's XML support extends the Spring Integration Core with implementations of splitter, transformer, selector and router designed to make working with xml messages in Spring Integration simple. The provided messaging components are designed to work with xml represented in a range of formats including instances of java.lang.String, org.w3c.dom.Document and javax.xml.transform.Source. It should be noted however that where a DOM representation is required, for example in order to evaluate an XPath expression, the String payload will be converted into the required type and then converted back again to String. Components that require an instance of DocumentBuilder will create a namespace aware instance if one is not provided. Where greater control of the document being created is required an appropriately configured instance of DocumentBuilder should be provided.

This section will explain the workings of UnmarshallingTransformer, MarshallingTransformer, XsltPayloadTransformer and how to configure them as beans. All of the provided xml transformers extend AbstractTransformer or AbstractPayloadTransformer and therefore implement Transformer. When configuring xml transformers as beans in Spring Integration you would normally configure the transformer in conjunction with either a MessageTransformingChannelInterceptor or a MessageTransformingHandler. This allows the transformer to be used as either an interceptor, which transforms the message as it is sent or received to the channel, or as an endpoint. Finally the namespace support will be discussed which allows for the simple configuration of the transformers as elements in XML.

UnmarshallingTransformer allows an xml Source to be unmarshalled using implementations of Spring OXM Unmarshaller. Spring OXM provides several implementations supporting marshalling and unmarshalling using JAXB, Castor and JiBX amongst others. Since the unmarshaller requires an instance of Source where the message payload is not currently an instance of Source, conversion will be attempted. Currently String and org.w3c.dom.Document payloads are supported. Custom conversion to a Source is also supported by injecting an implementation of SourceFactory.

<bean id="unmarshallingTransformer"
      class="org.springframework.integration.xml.transformer.UnmarshallingTransformer">
    <constructor-arg>
        <bean class="org.springframework.oxm.jaxb.Jaxb1Marshaller">
		    <property name="contextPath" value="org.example" />
        </bean>
    </constructor-arg>
</bean>

The MarshallingTransformer allows an object graph to be converted into xml using a Spring OXM Marshaller. By default the MarshallingTransformer will return a DomResult. However the type of result can be controlled by configuring an alternative ResultFactory such as StringResultFactory. In many cases it will be more convenient to transform the payload into an alternative xml format. To achieve this configure a ResultTransformer. Two implementations are provided, one which converts to String and another which converts to Document.

<bean id="marshallingTransformer"
      class="org.springframework.integration.xml.transformer.MarshallingTransformer">
    <constructor-arg>
        <bean class="org.springframework.oxm.jaxb.Jaxb1Marshaller">
            <property name="contextPath" value="org.example" />
        </bean>
    </constructor-arg>
    <constructor-arg>
        <bean class="org.springframework.integration.xml.transformer.ResultToDocumentTransformer" />
    </constructor-arg>
</bean>

By default, the MarshallingTransformer will pass the payload Object to the Marshaller, but if its boolean "extractPayload" property is set to "false", the entire Message instance will be passed to the Marshaller instead. That may be useful for certain custom implementations of the Marshaller interface, but typically the payload is the appropriate source Object for marshalling when delegating to any of the various out-of-the-box Marshaller implementations.

XsltPayloadTransformer transforms xml payloads using xsl. The transformer requires an instance of either Resource or Templates. Passing in a Templates instance allows for greater configuration of the TransformerFactory used to create the template instance. As in the case of XmlPayloadMarshallingTransformer by default XsltPayloadTransformer will create a message with a Result payload. This can be customised by providing a ResultFactory and/or a ResultTransformer.

<bean id="xsltPayloadTransformer"
      class="org.springframework.integration.xml.transformer.XsltPayloadTransformer">
    <constructor-arg value="classpath:org/example/xsl/transform.xsl" />
    <constructor-arg>
        <bean class="org.springframework.integration.xml.transformer.ResultToDocumentTransformer" />
    </constructor-arg>
</bean>

Namespace support for all xml transformers is provided in the Spring Integration xml namespace, a template for which can be seen below. The namespace support for transformers creates an instance of either EventDrivenConsumer or PollingConsumer according to the type of the provided input channel. The namespace support is designed to reduce the amount of xml configuration by allowing the creation of an endpoint and transformer using one element.

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

The namespace support for UnmarshallingTransformer is shown below. Since the namespace is now creating an endpoint instance rather than a transformer, a poller can also be nested within the element to control the polling of the input channel.

<si-xml:unmarshalling-transformer id="defaultUnmarshaller"
    input-channel="input"
    output-channel="output"
    unmarshaller="unmarshaller"/>

<si-xml:unmarshalling-transformer id="unmarshallerWithPoller"
    input-channel="input"
    output-channel="output"
    unmarshaller="unmarshaller">
    <si:poller>
        <si:interval-trigger interval="2000"/>
    </si:poller>
<si-xml:unmarshalling-transformer/>    
    

The namespace support for the marshalling transformer requires an input channel, output channel and a reference to a marshaller. The optional result-type attribute can be used to control the type of result created, valid values are StringResult or DomResult (the default). Where the provided result types are not sufficient a reference to a custom implementation of ResultFactory can be provided as an alternative to setting the result-type attribute using the result-factory attribute. An optional result-transformer can also be specified in order to convert the created Result after marshalling.

<si-xml:marshalling-transformer
     input-channel="marshallingTransformerStringResultFactory"
     output-channel="output"
     marshaller="marshaller"
     result-type="StringResult" />
     
<si-xml:marshalling-transformer
    input-channel="marshallingTransformerWithResultTransformer"
    output-channel="output"
    marshaller="marshaller"
    result-transformer="resultTransformer" />
     
<bean id="resultTransformer"
      class="org.springframework.integration.xml.transformer.ResultToStringTransformer"/>

Namespace support for the XsltPayloadTransformer allows either a resource to be passed in in order to create the Templates instance or alternatively a precreated Templates instance can be passed in as a reference. In common with the marshalling transformer the type of the result output can be controlled by specifying either the result-factory or result-type attribute. A result-transfomer attribute can also be used to reference an implementation of ResultTransfomer where conversion of the result is required before sending.

<si-xml:xslt-transformer id="xsltTransformerWithResource"
    input-channel="withResourceIn"
    output-channel="output"
    xsl-resource="org/springframework/integration/xml/config/test.xsl"/>
<si-xml:xslt-transformer id="xsltTransformerWithTemplatesAndResultTransformer"
    input-channel="withTemplatesAndResultTransformerIn"
    output-channel="output"
    xsl-templates="templates"
    result-transformer="resultTransformer"/>

XPathMessageSplitter supports messages with either String or Document payloads. The splitter uses the provided XPath expression to split the payload into a number of nodes. By default this will result in each Node instance becoming the payload of a new message. Where it is preferred that each message be a Document the createDocuments flag can be set. Where a String payload is passed in the payload will be converted then split before being converted back to a number of String messages. The XPath splitter implements MessageHandler and should therefore be configured in conjunction with an appropriate endpoint (see the namespace support below for a simpler configuration alternative).

<bean id="splittingEndpoint"
      class="org.springframework.integration.endpoint.EventDrivenConsumer">
    <constructor-arg ref="orderChannel" />
    <constructor-arg>
        <bean class="org.springframework.integration.xml.splitter.XPathMessageSplitter">
            <constructor-arg value="/order/items" />
            <property name="documentBuilder" ref="customisedDocumentBuilder" />
            <property name="outputChannel" ref="orderItemsChannel" />
        </bean>
    </constructor-arg>
</bean>

Two Router implementations based on XPath are provided XPathSingleChannelRouter and XPathMultiChannelRouter. The implementations differ in respect to how many channels any given message may be routed to, exactly one in the case of the single channel version or zero or more in the case of the multichannel router. Both evaluate an XPath expression against the xml payload of the message, supported payload types by default are Node, Document and String. For other payload types a custom implementation of XmlPayloadConverter can be provided. The router implementations use ChannelResolver to convert the result(s) of the XPath expression to a channel name. By default a BeanFactoryChannelResolver strategy will be used, this means that the string returned by the XPath evaluation should correspond directly to the name of a channel. Where this is not the case an alternative implementation of ChannelResolver can be used. Where there is a simple mapping from Xpath result to channel name the provided MapBasedChannelResolver can be used.

<!-- Expects a channel for each value of order type to exist  -->
<bean id="singleChannelRoutingEndpoint"
      class="org.springframework.integration.endpoint.EventDrivenConsumer">
    <constructor-arg ref="orderChannel" />
    <constructor-arg>
        <bean class="org.springframework.integration.xml.router.XPathSingleChannelRouter">
            <constructor-arg value="/order/@type" />
        </bean>
    </constructor-arg>
</bean>
	
	
<!-- Multi channel router which uses a map channel resolver to resolve the channel name
     based on the XPath evaluation result Since the router is multi channel it may deliver
     message to one or both of the configured channels -->
<bean id="multiChannelRoutingEndpoint"
      class="org.springframework.integration.endpoint.EventDrivenConsumer">
    <constructor-arg ref="orderChannel" />
    <constructor-arg>
        <bean class="org.springframework.integration.xml.router.XPathMultiChannelRouter">
            <constructor-arg value="/order/recipient" />
            <property name="channelResolver">
                <bean class="org.springframework.integration.channel.MapBasedChannelResolver">
                    <constructor-arg>
                        <map>
                            <entry key="accounts"
                                   value-ref="accountConfirmationChannel" />
                            <entry key="humanResources"
                                   value-ref="humanResourcesConfirmationChannel" />
                        </map>
                     </constructor-arg>
	             </bean>
            </property>
        </bean>
    </constructor-arg>
</bean>

Two MessageSelector implementations are provided, BooleanTestXPathMessageSelector and StringValueTestXPathMessageSelector. BooleanTestXPathMessageSelector requires an XPathExpression which evaluates to a boolean, for example boolean(/one/two) which will only select messages which have an element named two which is a child of a root element named one. StringValueTestXPathMessageSelector evaluates any XPath expression as a String and compares the result with the provided value.

<!-- Interceptor which rejects messages that do not have a root element order  -->
<bean id="orderSelectingInterceptor"
      class="org.springframework.integration.channel.interceptor.MessageSelectingInterceptor">
    <constructor-arg>
        <bean class="org.springframework.integration.xml.selector.BooleanTestXPathMessageSelector">
            <constructor-arg value="boolean(/order)" />
        </bean>
    </constructor-arg>
</bean>
	
<!--  Interceptor which rejects messages that are not version one orders --> 
<bean id="versionOneOrderSelectingInterceptor"
      class="org.springframework.integration.channel.interceptor.MessageSelectingInterceptor">
    <constructor-arg>
        <bean class="org.springframework.integration.xml.selector.StringValueTestXPathMessageSelector">
            <constructor-arg value="/order/@version" index="0"/>
            <constructor-arg value="1" index="1"/>
        </bean>
    </constructor-arg>
</bean>

All XPath based components have namespace support allowing them to be configured as Message Endpoints with the exception of the XPath selectors which are not designed to act as endpoints. Each component allows the XPath to either be referenced at the top level or configured via a nested xpath-expression element. So the following configurations of an xpath-selector are all valid and represent the general form of XPath namespace support. All forms of XPath expression result in the creation of an XPathExpression using the Spring XPathExpressionFactory

<si-xml:xpath-selector id="xpathRefSelector"
                       xpath-expression="refToXpathExpression"
                       evaluation-result-type="boolean" />

<si-xml:xpath-selector id="selectorWithNoNS" evaluation-result-type="boolean" >
    <si-xml:xpath-expression expression="/name"/>
</si-xml:xpath-selector>	
	
<si-xml:xpath-selector id="selectorWithOneNS" evaluation-result-type="boolean" >
    <si-xml:xpath-expression expression="/ns1:name"
                             ns-prefix="ns1" ns-uri="www.example.org" />
</si-xml:xpath-selector>
	
<si-xml:xpath-selector id="selectorWithTwoNS" evaluation-result-type="boolean" >
    <si-xml:xpath-expression expression="/ns1:name/ns2:type">
        <map>
            <entry key="ns1" value="www.example.org/one" />
            <entry key="ns2" value="www.example.org/two" />
        </map>
    </si-xml:xpath-expression>
</si-xml:xpath-selector>
	
<si-xml:xpath-selector id="selectorWithNamespaceMapRef" evaluation-result-type="boolean" >
    <si-xml:xpath-expression expression="/ns1:name/ns2:type"
                             namespace-map="defaultNamespaces"/>
</si-xml:xpath-selector>
	
<util:map id="defaultNamespaces">
    <util:entry key="ns1" value="www.example.org/one" />
    <util:entry key="ns2" value="www.example.org/two" />
</util:map>

XPath splitter namespace support allows the creation of a Message Endpoint with an input channel and output channel.

<!-- Split the order into items creating a new message for each item node -->
<si-xml:xpath-splitter id="orderItemSplitter"
                       input-channel="orderChannel"
                       output-channel="orderItemsChannel">
    <si-xml:xpath-expression expression="/order/items"/>
</si-xml:xpath-splitter>

<!-- Split the order into items creating a new document for each item-->
<si-xml:xpath-splitter id="orderItemDocumentSplitter"
                       input-channel="orderChannel" 
                       output-channel="orderItemsChannel"
                       create-documents="true">
    <si-xml:xpath-expression expression="/order/items"/>
    <si:poller>
        <si:interval-trigger interval="2000"/>
    </si:poller>
</si-xml:xpath-splitter>

XPath router namespace support allows for the creation of a Message Endpoint with an input channel but no output channel since the output channel is determined dynamically. The multi-channel attribute causes the creation of a multi channel router capable of routing a single message to many channels when true and a single channel router when false.

<!-- route the message according to exactly one order type channel -->
<si-xml:xpath-router id="orderTypeRouter" input-channel="orderChannel" multi-channel="false">
    <si-xml:xpath-expression expression="/order/type"/>
</si-xml:xpath-router>

<!-- route the order to all responders-->
<si-xml:xpath-router id="responderRouter" input-channel="orderChannel" multi-channel="true">
    <si-xml:xpath-expression expression="/request/responders"/>
    <si:poller>
        <si:interval-trigger interval="2000"/>
    </si:poller>
</si-xml:xpath-router>

Spring Integration provides integration with the Spring Security project to allow role based security checks to be applied to channel send and receive invocations.

Spring Integration provides the interceptor ChannelSecurityInterceptor, which extends AbstractSecurityInterceptor and intercepts send and receive calls on the channel. Access decisions are then made with reference to ChannelInvocationDefinitionSource which provides the definition of the send and receive security constraints. The interceptor requires that a valid SecurityContext has been established by authenticating with Spring Security, see the Spring Security reference documentation for details.

Namespace support is provided to allow easy configuration of security constraints. This consists of the secured channels tag which allows definition of one or more channel name patterns in conjunction with a definition of the security configuration for send and receive. The pattern is a java.util.regexp.Pattern.

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

<si-security:secured-channels>
    <si-security:access-policy pattern="admin.*" send-access="ROLE_ADMIN"/>
    <si-security:access-policy pattern="user.*" receive-access="ROLE_USER"/>
</si-security:secured-channels>
	

By default the secured-channels namespace element expects a bean named authenticationManager which implements AuthenticationManager and a bean named accessDecisionManager which implements AccessDecisionManager. Where this is not the case references to the appropriate beans can be configured as attributes of the secured-channels element as below.

<si-security:secured-channels access-decision-manager="customAccessDecisionManager"
                              authentication-manager="customAuthenticationManager">
    <si-security:access-policy pattern="admin.*" send-access="ROLE_ADMIN"/>
    <si-security:access-policy pattern="user.*" receive-access="ROLE_USER"/>
</si-security:secured-channels>