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.2 Filter

A Message Filter determines whether a Message should be passed to an output channel at all. This simply requires a boolean test method that may check for a particular payload content type, a property value, the presence of a header, etc. If the Message is accepted, it is sent to the output channel, but if not it will be dropped (or for a more severe implementation, an Exception could be thrown). Message Filters are often used in conjunction with a Publish Subscribe channel, where multiple consumers may receive the same Message and use the filter to narrow down the set of Messages to be processed based on some criteria.

Note

Be careful not to confuse the generic use of "filter" within the Pipes-and-Filters architectural pattern with this specific endpoint type that selectively narrows down the Messages flowing between two channels. The Pipes-and-Filters concept of "filter" matches more closely with Spring Integration's Message Endpoint: any component that can be connected to Message Channel(s) in order to send and/or receive Messages.

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 {
	...
}

Note

Even though the MessageHeaders implements Map, it is effectively a read-only implementation. Any attempt to put a value in the Map will result in an UnsupportedOperationException. The same applies for remove and clear. Since Messages may be passed to multiple consumers, the structure of the Map cannot be modified. Likewise, the Message's payload Object can not be set after the initial creation. However, the mutability of the header values themselves (or the payload Object) is intentionally left as a decision for the framework user.

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(MessageConsumer consumer);

    boolean unsubscribe(MessageConsumer consumer);

}

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

public QueueChannel(int capacity)

3.2.5 DirectChannel

The DirectChannel has point-to-point semantics, but otherwise is more similar to the PublishSubscribeChannel than any of the queue-based channel implementations described above. It implements the SubscribableChannel interface instead of the PollableChannel interface, so it dispatches Messages directly to a subscriber. As a point-to-point channel, however, it differs from the PublishSubscribeChannel in that it will only send each Message to a single subscribed MessageConsumer. Its primary purpose is to enable a single thread to perform the operations on "both sides" of the channel. For example, if a consumer is subscribed to a DirectChannel, then sending a Message to that channel will trigger invocation of that consumer's onMessage(Message) method directly in the sender's thread. The key motivation for providing a channel implementation with this behavior is to support transactions that must span across the channel while still benefiting from the abstraction and loose coupling that the channel provides. If the send call is invoked within the scope of a transaction, then the outcome of the consumer's invocation (e.g. updating a database record) can play a role in determining the ultimate result of that transaction (commit or rollback).

Note

Since the DirectChannel is the simplest option and does not add any additional overhead that would be required for scheduling and managing the threads of a poller, it is the default channel type within Spring Integration. The general idea is to define the channels for an application and then to consider which of those needs to provide buffering to throttle input, and to modify those to be queue-based PollableChannels. Likewise, if a channel needs to broadcast messages, it should not be a DirectChannel but rather a PublishSubscribeChannel. Below you will see how these can be configured.

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 an Exception). Also, the preReceive method can return 'false' to prevent the receive operation from proceeding.

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 parameter 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;
    }
}

Note

Keep in mind that receive() calls are only relevant for PollableChannels. In fact the SubscribableChannel interface does not even define a receive() method. The reason for this is that when a Message is sent to a SubscribableChannel it will be sent directly to one or more subscribers depending on the type of channel (e.g. a PublishSubscribeChannel sends to all of its subscribers). Therefore, the preReceive(..) and postReceive(..) interceptor methods are only invoked when the interceptor is applied to a PollableChannel.

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 also provide a variety of "queue" sub-elements to create the channel types (as described in Section 3.2, “Message Channel Implementations”). Examples of each are shown below.

<channel id="exampleChannel"/>

3.5.2 QueueChannel Configuration

To create a QueueChannel, use the "queue" sub-element. You may specify the channel's capacity:

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

	Note
	If you do not provide a value for the 'capacity' attribute on this <queue/> sub-element, the resulting queue will be unbounded. To avoid issues such as OutOfMemoryErrors, it's highly recommended to set an explicit value for a bounded queue.

<publish-subscribe-channel id="exampleChannel" task-executor="someTaskExecutor"/>

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

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

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

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

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

Message channels may also have interceptors as described in Section 3.3, “Channel Interceptors”. One or more <interceptor> elements can be added as sub-elements of <channel> (or the more specific element types). Provide the "ref" attribute to reference any Spring-managed object that implements the ChannelInterceptor interface:

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

In general, it is a good idea to define the interceptor implementations in a separate location since they usually provide common behavior that can be reused across multiple channels.

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("exampleSubscribableChannel");

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("examplePollableChannel");

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

A Polling Consumer may even delegate to a Spring TaskExecutor and 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.

Note

Many of the MessageHandler implementations are also capable of generating reply Messages. As mentioned above, sending Messages is trivial when compared to the Message reception. Nevertheless, when and how many reply Messages are sent depends on the handler type. For example, an Aggregator waits for a number of Messages to arrive and is often configured as a downstream consumer for a Splitter which may generate multiple replies for each Message it handles. When using the namespace configuration, you do not strictly need to know all of the details, but it still might be worth knowing that several of these components share a common base class, the AbstractReplyProducingMessageHandler, and it provides a setOutputChannel(..) method.

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 will 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="REQUIRES_NEW"
                   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 a convenience, there is also namespace support for creating a simple thread pool executor. The <thread-pool-task-executor/> element defines attributes for common concurrency settings such as core-size, max-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>

 <thread-pool-task-executor id="pool"
                            core-size="5"
                            max-size="25"
                            queue-capacity="20"
                            keep-alive-seconds="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 MessageBus' 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 thread pool elements is simply provided for convenience.

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 RETURN_ADDRESS 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.

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 'cronExpression' for an individual channel-adapter's schedule, 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 * * * * MON-FRI"/>
    </poller>
</channel-adapter>

An "outbound-channel-adapter" element can also connect a MessageChannel to any method that should be invoked with the payload of any Message 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>

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>

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

	Note
	The router implementations share some common properties, such as "defaultOutputChannel" and "resolutionRequired". If "resolutionRequired" is set to "true", and the router is unable to determine a target channel (e.g. there is no matching payload for a PayloadTypeRouter and no "defaultOutputChannel" has been specified), then an Exception will be thrown.

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 to one of the implementations described above:

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

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

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

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 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. Otherwise, it will be dropped.

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

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

Note

A common usage for Message Filters is in conjunction with a publish/subscribe channel. Many filter endpoints may be subscribed to the same channel, and they decide whether or not to pass the Message for the next endpoint which could be any of the supported types (e.g. Service Activator). This provides a reactive alternative to the more proactive approach of using a Message Router with a single point-to-point input channel and multiple output channels.

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.

	Note
	For the same reason of maximizing flexibility, Spring does not require XML-based Message payloads. Nevertheless, the framework does provide some convenient Transformers for dealing with XML-based payloads if that is indeed the right choice for your application. For more information on those transformers, see Chapter 25, Dealing with XML Payloads.

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.

The method that is used for transformation may expect either the Message type or the payload type of inbound Messages. 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 value is null, then no reply Message will be sent (effectively the same behavior as a Message Filter). Otherwise, the return value will be sent as the payload of a Message.

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 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 MessageConsumer 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: