The HTTP module now provides powerful Request Mapping support for Inbound Endpoints. Class UriPathHandlerMapping was replaced by IntegrationRequestMappingHandlerMapping, which is registered under the bean name integrationRequestMappingHandlerMapping in the application context. Upon parsing of the HTTP Inbound Endpoint, a new IntegrationRequestMappingHandlerMapping bean is either registered or an existing bean is being reused. To achieve flexible Request Mapping configuration, Spring Integration provides the <request-mapping/> sub-element for the <http:inbound-channel-adapter/> and the <http:inbound-gateway/>. Both HTTP Inbound Endpoints are now fully based on the Request Mapping infrastructure that was introduced with Spring MVC 3.1. For example, multiple paths are supported on a single inbound endpoint. For more information see Section 16.4, “HTTP Namespace Support”.

A new IntegrationEvaluationContextFactoryBean is provided to allow configuration of custom PropertyAccessors and functions for use in SpEL expressions throughout the framework. For more information see Appendix A, Spring Expression Language (SpEL).

To customize the SpEL EvaluationContext with static Method functions, the new <spel-function/> component is introduced. Two built-in functions are also provided (#jsonPath and #xpath). For more information see Section A.3, “SpEL Functions”.

To customize the SpEL EvaluationContext with PropertyAccessor implementations the new <spel-property-accessors/> component is introduced. For more information see Section A.4, “PropertyAccessors”.

A new Redis-based MetadataStore implementation has been added. The RedisMetadataStore can be used to maintain state of a MetadataStore across application restarts. This new MetadataStore implementation can be used with adapters such as:

New queue-based components have been added. The <int-redis:queue-inbound-channel-adapter/> and the <int-redis:queue-outbound-channel-adapter/> components are provided to perform 'right pop' and 'left push' operations on a Redis List, respectively.

For more information see Chapter 22, Redis Support.

It is now possible to instruct the framework to store reply and error channels in a registry for later resolution. This is useful for cases where the replyChannel or errorChannel might be lost; for example when serializing a message. See Section 6.2.2, “Header Enricher” for more information.

In addition to the existing eMongoDbMessageStore, a new ConfigurableMongoDbMessageStore has been introduced. This provides a more robust and flexible implementation of MessageStore for MongoDB. It does not have backward compatibility, with the existing store, but it is recommended to use it for new applications. Existing applications can use it, but messages in the old store will not be available. See Chapter 21, MongoDb Support for more information.

Building on the 2.2 SyslogToMapTransformer Spring Integration 3.0 now introduces UDP and TCP inbound channel adapters especially tailored for receiving SYSLOG messages. For more information, see Chapter 27, Syslog Support.

File 'tail'ing inbound channel adapters are now provided to generate messages when lines are added to the end of text files; see Section 13.2.1, “'Tail'ing Files”.

For more information, see Section 8.1, “JMX Support”.

TcpConnections now emit ApplicationEvents (specifically TcpConnectionEvents) when connections are opened, closed, or an exception occurs. This allows applications to be informed of changes to TCP connections using the normal Spring ApplicationListener mechanism.

AbstractTcpConnection has been renamed TcpConnectionSupport; custom connections that are subclasses of this class, can use its methods to publish events. Similarly, AbstractTcpConnectionInterceptor has been renamed to TcpConnectionInterceptorSupport.

In addition, a new <int-ip:tcp-connection-event-inbound-channel-adapter/> is provided; by default, this adapter sends all TcpConnectionEvents to a Channel.

Further, the TCP Connection Factories, now provide a new method getOpenConnectionIds(), which returns a list of identifiers for all open connections; this allows applications, for example, to broadcast to all open connections.

Finally, the connection factories also provide a new method closeConnection(String connectionId) which allows applications to explicitly close a connection using its ID.

For more information see Section 28.5, “TCP Connection Events”.

The <int:inbound-channel-adapter/> now supports <expression/> and <script/> sub-elements to create a MessageSource; see Section 3.3.3, “Channel Adapter Expressions and Scripts”.

The Content Enricher now provides configuration for <header/> sub-elements, to enrich the outbound Message with headers based on the reply Message from the underlying message flow. For more information see Section 6.2.3, “Payload Enricher”.

Previously, message ids were generated using the JDK UUID.randomUUID() method. With this release, the default mechanism has been changed to use a more efficient algorithm which is significantly faster. In addition, the ability to change the strategy used to generate message ids has been added. For more information see the section called “Message ID Generation”.

For more information see Section 7.2, “Messaging Gateways”.

For more information see Chapter 16, HTTP Support.

For more information, see 'JSON Transformers' in Section 6.1, “Transformer”.

Previously, the id attribute for elements within a <chain> was ignored and, in some cases, disallowed. Now, the id attribute is allowed for all elements within a <chain>. The bean names of chain elements is a combination of the surrounding chain's id and the id of the element itself. For example: 'fooChain$child.fooTransformer.handler'. For more information see Section 5.6, “Message Handler Chain”.

The AbstractCorrelatingMessageHandler provides a new property empty-group-min-timeout to allow empty group expiry to run on a longer schedule than expiring partial groups. Empty groups will not be removed from the MessageStore until they have not been modified for at least this number of milliseconds. For more information see Section 5.4.4, “Configuring an Aggregator”.

New FileListFilters that use a persistent MetadataStore are now available. These can be used to prevent duplicate files after a system restart. See Section 13.2, “Reading Files”, Section 14.3, “FTP Inbound Channel Adapter”, and Section 25.5, “SFTP Inbound Channel Adapter” for more information.

A new variables attribute has been introduced for scripting components. In addition, variable bindings are now allowed for inline scripts. See Section 7.6, “Groovy support” and Section 7.5, “Scripting support” for more information.

Previously, when configuring LoadBalancingStrategy on the channel's 'dispatcher' sub-element, the only available option was to use a pre-defined enumeration of values which did not allow one to set a custom implementation of the LoadBalancingStrategy. You can now use load-balancer-ref to provide a reference to a custom implementation of the LoadBalancingStrategy. For more information see the section called “DirectChannel”.

Previously, sending to a <publish-subscribe-channel/> that had no subscribers would return a false result. If used in conjunction with a MessagingTemplate, this would result in an exception being thrown. Now, the PublishSubscribeChannel has a property minSubscribers (default 0). If the message is sent to at least the minimum number of subscribers, the send is deemed to be successful (even if zero). If an application is expecting to get an exception under these conditions, set the minimum subscribers to at least 1.

The FTP, SFTP and FTPS endpoints no longer cache sessions by default

The deprecated cached-sessions attribute has been removed from all endpoints. Previously, the embedded caching mechanism controlled by this attribute's value didn't provide a way to limit the size of the cache, which could grow indefinitely. The CachingConnectionFactory was introduced in release 2.1 and it became the preferred (and is now the only) way to cache sessions.

The CachingConnectionFactory now provides a new method resetCache(). This immediately closes idle sessions and causes in-use sessions to be closed as and when they are returned to the cache.

The DefaultSftpSessionFactory (in conjunction with a CachingSessionFactory) now supports multiplexing channels over a single SSH connection (SFTP Only).

FTP, SFTP and FTPS Inbound Adapters

Previously, there was no way to override the default filter used to process files retrieved from a remote server. The filter attribute determines which files are retrieved but the FileReadingMessageSource uses an AcceptOnceFileListFilter. This means that if a new copy of a file is retrieved, with the same name as a previously copied file, no message was sent from the adapter.

With this release, a new attribute local-filter allows you to override the default filter, for example with an AcceptAllFileListFilter, or some other custom filter.

For users that wish the behavior of the AcceptOnceFileListFilter to be maintained across JVM executions, a custom filter that retains state, perhaps on the file system, can now be configured.

Inbound Channel Adapters now support the preserve-timestamp attribute, which sets the local file modified timestamp to the timestamp from the server (default false).

FTP, SFTP and FTPS Gateways

Remote File Template

A new higher-level abstraction (RemoteFileTemplate) is provided over the Session implementations used by the FTP and SFTP modules. While it is used internally by endpoints, this abstraction can also be used programmatically and, like all Spring *Template implementations, reliably closes the underlying session while allowing low level access to the session when needed.

For more information, see Chapter 14, FTP/FTPS Adapters and Chapter 25, SFTP Adapters.

All Outbound Gateways (e.g. <jdbc:outbound-gateway/> or <jms:outbound-gateway/>) are designed for 'request-reply' scenarios. A response is expected from the external service and will be published to the reply-channel, or the replyChannel message header. However, there are some cases where the external system might not always return a result, e.g. a <jdbc:outbound-gateway/>, when a SELECT ends with an empty ResultSet or, say, a Web Service is One-Way. An option is therefore needed to configure whether or not a reply is required. For this purpose, the requires-reply attribute has been introduced for Outbound Gateway components. In most cases, the default value for requires-reply is true and, if there is not any result, a ReplyRequiredException will be thrown. Changing the value to false means that, if an external service doesn't return anything, the message-flow will end at that point, similar to an Outbound Channel Adapter.

Note

The WebService outbound gateway has an additional attribute ignore-empty-responses; this is used to treat an empty String response as if no response was received. It is true by default but can be set to false to allow the application to receive an empty String in the reply message payload. When the attribute is true an empty string is treated as no response for the purposes of the requires-reply attribute. requires-reply is false by default for the WebService outbound gateway.

Note, the requiresReply property was previously present in the AbstractReplyProducingMessageHandler but set to false, and there wasn't any way to configure it on Outbound Gateways using the XML namespace.

	Important
	Previously, a gateway receiving no reply would silently end the flow (with a DEBUG log message); with this change an exception will now be thrown by default by most gateways. To revert to the previous behavior, set requires-reply to false.

Previously, the <int-amqp:outbound-gateway/> mapped headers before invoking the message converter, and the converter could overwrite headers such as content-type. The outbound adapter maps the headers after the conversion, which means headers like content-type from the outbound Message (if present) are used.

Starting with this release, the gateway now maps the headers after the message conversion, consistent with the adapter. If your application relies on the previous behavior (where the converter's headers overrode the mapped headers), you either need to filter those headers (before the message reaches the gateway) or set them appropriately. The headers affected by the SimpleMessageConverter are content-type and content-encoding. Custom message converters may set other headers.

For more complex database-specific types, not supported by the standard CallableStatement.getObject method, 2 new additional attributes were introduced to the <sql-parameter-definition/> element with OUT-direction:

The row-mapper attribute of the Stored Procedure Inbound Channel Adapter <returning-resultset/> sub-element now supports a reference to a RowMapper bean definition. Previously, it contained just a class name (which is still supported).

For more information see Section 17.5, “Stored Procedures”.

Web Service Outbound Gateway 'uri' attribute now supports <uri-variable/> substitution for all URI-schemes supported by Spring Web Services. For more information see Section 30.4, “Outbound URI Configuration”.

For more information, see Chapter 22, Redis Support.

Previously, when a <filter/> had a <request-handler-advice-chain/>, the discard action was all performed within the scope of the advice chain (including any downstream flow on the discard-channel). The filter element now has an attribute discard-within-advice (default true), to allow the discard action to be performed after the advice chain completes. See Section 7.7.4, “Advising Filters”.

Request Handler Advice Chains can now be configured using annotations. See Section 7.7.5, “Advising Endpoints Using Annotations”.

This transformer now correctly transforms byte[] and char[] payloads to String. For more information see Section 6.1, “Transformer”.

Payloads to persist or merge can now be of type java.lang.Iterable.

In that case, each object returned by the Iterable is treated as an entity and persisted or merged using the underlying EntityManager. NULL values returned by the iterator are ignored.

The JPA adapters now have additional attributes to optionally 'flush' and 'clear' entities from the associated persistence context after performing persistence operations.

Retrieving gateways had no mechanism to specify the first record to be retrieved which is a common use case. The retrieving gateways now support specifying this parameter using a first-result and first-result-expression attributes to the gateway definition. Section 18.6.3, “Retrieving Outbound Gateway”.

The JPA retrieving gateway and inbound adapter now have an attribute to specify the maximum number of results in a result set as an expression. In addition, the max-results attribute has been introduced to replace max-number-of-results, which has been deprecated. max-results and max-results-expression are used to provide the maximum number of results, or an expression to compute the maximum number of results, respectively, in the result set.

For more information see Chapter 18, JPA Support.

Previously, the <delayer> provided a delay-header-name attribute to determine the delay value at runtime. In complex cases it was necessary to precede the <delayer> with a <header-enricher>. Spring Integration 3.0 introduced the expression attribute and expression sub-element for dynamic delay determination. The delay-header-name attribute is now deprecated because the header evaluation can be specified in the expression. In addition, the ignore-expression-failures was introduced to control the behavior when an expression evaluation fails. For more information see Section 7.4, “Delayer”.

Spring Integration 3.0 adds a new set of DDL scripts for MySQL version 5.6.4 and higher. Now MySQL supports fractional seconds and is thus improving the FIFO ordering when polling from a MySQL-based Message Store. For more information, please see Section 17.4.1, “The Generic JDBC Message Store”.

Previously, if an IMAP idle connection failed, it was logged but there was no mechanism to inform an application. Such exceptions now generate ApplicationEvents. Applications can obtain these events using an <int-event:inbound-channel-adapter> or any ApplicationListener configured to receive an ImapIdleExceptionEvent or one of its super classes.

The TCP connection factories now enable the configuration of a flexible mechanism to transfer selected headers (as well as the payload) over TCP. A new TcpMessageMapper enables the selection of the headers, and an appropriate (de)serializer needs to be configured to write the resulting Map to the TCP stream. A MapJsonSerializer is provided as a convenient mechanism to transfer headers and payload over TCP. For more information see Section 28.8.4, “Transferring Headers”.

Previously, when configuring a <message-driven-channel-adapter/>, if you wished to use a specific TaskExecutor, it was necessary to declare a container bean and provide it to the adapter using the container attribute. The task-executor is now provided, allowing it to be set directly on the adapter. This is in addition to several other container attributes that were already available.

The RMI Inbound Gateway now supports an error-channel attribute. See Section 24.3, “Inbound RMI”.

You can now specify the transformer factory class name using the transformer-factory-class attribute. See the section called “XsltPayloadTransformer”

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, correlation id, 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.

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.1.2, “Message Channel Implementations”.

One of the primary goals of Spring Integration is to simplify the development of enterprise integration solutions through inversion of control. This means that you should not have to implement consumers and producers directly, and you should not even have to build Messages and invoke send or receive operations on a Message Channel. Instead, you should be able to focus on your specific domain model with an implementation based on plain Objects. Then, by providing declarative configuration, you can "connect" your domain-specific code to the messaging infrastructure provided by Spring Integration. The components responsible for these connections are Message Endpoints. This does not mean that you will necessarily connect your existing application code directly. Any real-world enterprise integration solution will require some amount of code focused upon integration concerns such as routing and transformation. The important thing is to achieve separation of concerns between such integration logic and business logic. In other words, as with the Model-View-Controller paradigm for web applications, the goal should be to provide a thin but dedicated layer that translates inbound requests into service layer invocations, and then translates service layer return values into outbound replies. The next section will provide an overview of the Message Endpoint types that handle these responsibilities, and in upcoming chapters, you will see how Spring Integration's declarative configuration options provide a non-invasive way to use each of these.

A Message Transformer is responsible for converting a Message's content or structure and returning the modified Message. Probably the most common type of transformer is one that converts the payload of the Message from one format to another (e.g. from XML Document to java.lang.String). Similarly, a transformer may be used to add, remove, or modify the Message's header values.

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.

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.

A Splitter is another type of Message Endpoint whose responsibility is to accept a Message from its input channel, split that Message into multiple Messages, and then send each of those to its output channel. This is typically used for dividing a "composite" payload object into a group of Messages containing the sub-divided payloads.

Basically a mirror-image of the Splitter, the Aggregator is a type of Message Endpoint 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, and to timeout if necessary. Furthermore, in case of a timeout, the Aggregator needs to know whether to send the partial results or to discard them to a separate channel. Spring Integration provides a CompletionStrategy as well as configurable settings for timeout, whether to send partial results upon timeout, and the discard channel.

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.

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.

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

public interface MessageChannel {

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

}

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.

PublishSubscribeChannel

The PublishSubscribeChannel implementation broadcasts any Message sent to it to all of its subscribed handlers. This is most often used for sending Event Messages whose primary role is notification as opposed to Document Messages which are generally intended to be processed by a single handler. Note that the PublishSubscribeChannel is intended for sending only. Since it broadcasts to its subscribers directly when its send(Message) method is invoked, consumers cannot poll for Messages (it does not implement PollableChannel and therefore has no receive() method). Instead, any subscriber must be a MessageHandler itself, and the subscriber's handleMessage(Message) method will be invoked in turn.

Prior to version 3.0, invoking the send method on a PublishSubscribeChannel that had no subscribers returned false. When used in conjunction with a MessagingTemplate, a MessageDeliveryException was thrown. Starting with version 3.0, the behavior has changed such that a send is always considered successful if at least the minimum subscribers are present (and successfully handle the message). This behavior can be modified by setting the minSubscribers property, which defaults to 0.

	Note
	If a TaskExecutor is used, only the presence of the correct number of subscribers is used for this determination, because the actual handling of the message is performed asynchronously.

public QueueChannel(int capacity)

RendezvousChannel

The RendezvousChannel enables a "direct-handoff" scenario where a sender will block until another party invokes the channel's receive() method or vice-versa. Internally, this implementation is quite similar to the QueueChannel except that it uses a SynchronousQueue (a zero-capacity implementation of BlockingQueue). This works well in situations where the sender and receiver are operating in different threads but simply dropping the message in a queue asynchronously is not appropriate. In other words, with a RendezvousChannel at least the sender knows that some receiver has accepted the message, whereas with a QueueChannel, the message would have been stored to the internal queue and potentially never received.

Tip

Keep in mind that all of these queue-based channels are storing messages in-memory only by default. When persistence is required, you can either provide a 'message-store' attribute within the 'queue' element to reference a persistent MessageStore implementation, or you can replace the local channel with one that is backed by a persistent broker, such as a JMS-backed channel or Channel Adapter. The latter option allows you to take advantage of any JMS provider's implementation for message persistence, and it will be discussed in Chapter 19, JMS Support. However, when buffering in a queue is not necessary, the simplest approach is to rely upon the DirectChannel discussed next.

The RendezvousChannel is also useful for implementing request-reply operations. The sender can create a temporary, anonymous instance of RendezvousChannel which it then sets as the 'replyChannel' header when building a Message. After sending that Message, the sender can immediately call receive (optionally providing a timeout value) in order to block while waiting for a reply Message. This is very similar to the implementation used internally by many of Spring Integration's request-reply components.

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

In addition to being the simplest point-to-point channel option, one of its most important features is that it enables a single thread to perform the operations on "both sides" of the channel. For example, if a handler is subscribed to a DirectChannel, then sending a Message to that channel will trigger invocation of that handler's handleMessage(Message) method directly in the sender's thread, before the send() method invocation can return.

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 handler's invocation (e.g. updating a database record) will 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 need to provide buffering or to throttle input, and then 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 each of these can be configured.

The DirectChannel internally delegates to a Message Dispatcher to invoke its subscribed Message Handlers, and that dispatcher can have a load-balancing strategy exposed via load-balancer or load-balancer-ref attributes (mutually exclusive). The load balancing strategy is used by the Message Dispatcher to help determine how Messages are distributed amongst Message Handlers in the case that there are multiple Message Handlers subscribed to the same channel. As a convinience the load-balancer attribute exposes enumeration of values pointing to pre-existing implementations of LoadBalancingStrategy. The "round-robin" (load-balances across the handlers in rotation) and "none" (for the cases where one wants to explicitely disable load balancing) are the only available values. Other strategy implementations may be added in future versions. However, since version 3.0 you can provide your own implementation of the LoadBalancingStrategy and inject it using load-balancer-ref attribute which should point to a bean that implements LoadBalancingStrategy.

<int:channel id="lbRefChannel">
  <int:dispatcher load-balancer-ref="lb"/>
</int:channel>
  
<bean id="lb" class="foo.bar.SampleLoadBalancingStrategy"/>

Note that load-balancer or load-balancer-ref attributes are mutually exclusive.

The load-balancing also works in combination with a boolean failover property. If the "failover" value is true (the default), then the dispatcher will fall back to any subsequent handlers as necessary when preceding handlers throw Exceptions. The order is determined by an optional order value defined on the handlers themselves or, if no such value exists, the order in which the handlers are subscribed.

If a certain situation requires that the dispatcher always try to invoke the first handler, then fallback in the same fixed order sequence every time an error occurs, no load-balancing strategy should be provided. In other words, the dispatcher still supports the failover boolean property even when no load-balancing is enabled. Without load-balancing, however, the invocation of handlers will always begin with the first according to their order. For example, this approach works well when there is a clear definition of primary, secondary, tertiary, and so on. When using the namespace support, the "order" attribute on any endpoint will determine that order.

	Note
	Keep in mind that load-balancing and failover only apply when a channel has more than one subscribed Message Handler. When using the namespace support, this means that more than one endpoint shares the same channel reference in the "input-channel" attribute.

ExecutorChannel

The ExecutorChannel is a point-to-point channel that supports the same dispatcher configuration as DirectChannel (load-balancing strategy and the failover boolean property). The key difference between these two dispatching channel types is that the ExecutorChannel delegates to an instance of TaskExecutor to perform the dispatch. This means that the send method typically will not block, but it also means that the handler invocation may not occur in the sender's thread. It therefore does not support transactions spanning the sender and receiving handler.

Tip

Note that there are occasions where the sender may block. For example, when using a TaskExecutor with a rejection-policy that throttles back on the client (such as the ThreadPoolExecutor.CallerRunsPolicy), the sender's thread will execute the method directly anytime the thread pool is at its maximum capacity and the executor's work queue is full. Since that situation would only occur in a non-predictable way, that obviously cannot be relied upon for transactions.

<int:channel id="threadScopedChannel" scope="thread">
     <int:queue />
</int:channel>

<bean class="org.springframework.beans.factory.config.CustomScopeConfigurer">
    <property name="scopes">
        <map>
            <entry key="thread" value="org.springframework.context.support.SimpleThreadScope" />
        </map>
    </property>
</bean>

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.

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.

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 the section called “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;
    }
}

Tip

The order of invocation for the interceptor methods depends on the type of channel. As described above, the queue-based channels are the only ones where the receive method is intercepted in the first place. Additionally, the relationship between send and receive interception depends on the timing of separate sender and receiver threads. For example, if a receiver is already blocked while waiting for a message the order could be: preSend, preReceive, postReceive, postSend. However, if a receiver polls after the sender has placed a message on the channel and already returned, the order would be: preSend, postSend, (some-time-elapses) preReceive, postReceive. The time that elapses in such a case depends on a number of factors and is therefore generally unpredictable (in fact, the receive may never happen!). Obviously, the type of queue also plays a role (e.g. rendezvous vs. priority). The bottom line is that you cannot rely on the order beyond the fact that preSend will precede postSend and preReceive will precede postReceive.

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

MessagingTemplate template = new MessagingTemplate();

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

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 MessageChannel channel, final Message<?> message) { ... }

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

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

	Note
	A less invasive approach that allows you to invoke simple interfaces with payload and/or header values instead of Message instances is described in Section 7.2.1, “Enter the GatewayProxyFactoryBean”.

To create a Message Channel instance, you can use the <channel/> element:

<int:channel id="exampleChannel"/>

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

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

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.1.2, “Message Channel Implementations”). Examples of each are shown below.

<int:channel id="directChannel"/>

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

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

Datatype Channel Configuration

There are times when a consumer can only process a particular type of payload and you need to therefore ensure the payload type of input Messages. Of course the first thing that comes to mind is Message Filter. However all that Message Filter will do is filter out Messages that are not compliant with the requirements of the consumer. Another way would be to use a Content Based Router and route Messages with non-compliant data-types to specific Transformers to enforce transformation/conversion to the required data-type. This of course would work, but a simpler way of accomplishing the same thing is to apply the Datatype Channel pattern. You can use separate Datatype Channels for each specific payload data-type.

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:

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

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

So the 'numberChannel' above will only accept Messages with a data-type of java.lang.Number. But what happens if the payload of the Message is not of the required type? It depends on whether you have defined a bean named "integrationConversionService" that is an instance of Spring's Conversion Service. If not, then an Exception would be thrown immediately, but if you do have an "integrationConversionService" bean defined, it will be used in an attempt to convert the Message's payload to the acceptable type.

You can even register custom converters. For example, let's say you are sending a Message with a String payload to the 'numberChannel' we configured above.

MessageChannel inChannel = context.getBean("numberChannel", MessageChannel.class);
inChannel.send(new GenericMessage<String>("5"));

Typically this would be a perfectly legal operation, however since we are using Datatype Channel the result of such operation would generate an exception:

Exception in thread "main" org.springframework.integration.MessageDeliveryException:
Channel 'numberChannel'
expected one of the following datataypes [class java.lang.Number],
but received [class java.lang.String]
…

And rightfully so since we are requiring the payload type to be a Number while sending a String. So we need something to convert String to a Number. All we need to do is implement a Converter.

public static class StringToIntegerConverter implements Converter<String, Integer> {
	public Integer convert(String source) {
		return Integer.parseInt(source);
	}
}

Then, register it as a Converter with the Integration Conversion Service:

<int:converter ref="strToInt"/>

<bean id="strToInt" class="org.springframework.integration.util.Demo.StringToIntegerConverter"/>

When the 'converter' element is parsed, it will create the "integrationConversionService" bean on-demand if one is not already defined. With that Converter in place, the send operation would now be successful since the Datatype Channel will use that Converter to convert the String payload to an Integer.

	Note
	For more information regarding Payload Type Conversion, please read Section 7.1.6, “Payload Type Conversion”.

QueueChannel Configuration

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

<int:channel id="queueChannel">
    <queue capacity="25"/>
</int: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 is highly recommended to set an explicit value for a bounded queue.

Persistent QueueChannel Configuration

Since a QueueChannel provides the capability to buffer Messages, but does so in-memory only by default, it also introduces a possibility that Messages could be lost in the event of a system failure. To mitigate this risk, a QueueChannel may be backed by a persistent implementation of the MessageGroupStore strategy interface. For more details on MessageGroupStore and MessageStore see Section 8.3, “Message Store”.

When a QueueChannel receives a Message, it will add it to the Message Store, and when a Message is polled from a QueueChannel, it is removed from the Message Store.

By default any QueueChannel only stores its Messages in an in-memory Queue and can therefore lead to the lost message scenario mentioned above. However Spring Integration provides a JdbcMessageStore to allow a QueueChannel to be backed by an RDBMS.

You can configure a Message Store for any QueueChannel by adding the message-store attribute as shown in the next example.

<int:channel id="dbBackedChannel">
    <int:queue message-store="messageStore"/>
</int:channel>

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

The above example also shows that JdbcMessageStore can be configured with the namespace support provided by the Spring Integration JDBC module. All you need to do is inject any javax.sql.DataSource instance. The Spring Integration JDBC module also provides schema DDL for most popular databases. These schemas are located in the org.springframework.integration.jdbc package of that module (spring-integration-jdbc).

	Important
	One important feature is that with any transactional persistent store (e.g., JdbcMessageStore), as long as the poller has a transaction configured, a Message removed from the store will only be permanently removed if the transaction completes successfully, otherwise the transaction will roll back and the Message will not be lost.

Many other implementations of the Message Store will be available as the growing number of Spring projects related to "NoSQL" data stores provide the underlying support. Of course, you can always provide your own implementation of the MessageGroupStore interface if you cannot find one that meets your particular needs.

PublishSubscribeChannel Configuration

To create a PublishSubscribeChannel, use the <publish-subscribe-channel/> element. When using this element, you can also specify the task-executor used for publishing Messages (if none is specified it simply publishes in the sender's thread):

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

If you are providing a Resequencer or Aggregator downstream from a PublishSubscribeChannel, then you can set the 'apply-sequence' property on the channel to true. That will indicate that the channel should set the sequence-size and sequence-number Message headers as well as the correlation id prior to passing the Messages along. For example, if there are 5 subscribers, the sequence-size would be set to 5, and the Messages would have sequence-number header values ranging from 1 to 5.

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

	Note
	The apply-sequence value is false by default so that a Publish Subscribe Channel can send the exact same Message instances to multiple outbound channels. Since Spring Integration enforces immutability of the payload and header references, the channel creates new Message instances with the same payload reference but different header values when the flag is set to true.

ExecutorChannel

To create an ExecutorChannel, add the <dispatcher> sub-element along with a task-executor attribute. Its value can reference any TaskExecutor within the context. For example, this enables configuration of a thread-pool for dispatching messages to subscribed handlers. As mentioned above, this does break the "single-threaded" execution context between sender and receiver so that any active transaction context will not be shared by the invocation of the handler (i.e. the handler may throw an Exception, but the send invocation has already returned successfully).

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

Note

The load-balancer and failover options are also both available on the <dispatcher/> sub-element as described above in the section called “DirectChannel Configuration”. The same defaults apply as well. So, the channel will have a round-robin load-balancing strategy with failover enabled unless explicit configuration is provided for one or both of those attributes. <int:channel id="executorChannelWithoutFailover"> <int:dispatcher task-executor="someExecutor" failover="false"/> </int:channel>

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

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

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

<int:channel id="threadLocalChannel" scope="thread"/>

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

Global Channel Interceptor Configuration

Channel Interceptors provide a clean and concise way of applying cross-cutting behavior per individual channel. If the same behavior should be applied on multiple channels, configuring the same set of interceptors for each channel would not be the most efficient way. To avoid repeated configuration while also enabling interceptors to apply to multiple channels, Spring Integration provides Global Interceptors. Look at the example below:

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

or

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

<bean id="myInterceptor" class="foo.barSampleInterceptor"/>

Each <channel-interceptor/> element allows you to define a global interceptor which will be applied on all channels that match any patterns defined via the pattern attribute. In the above case the global interceptor will be applied on the 'foo' channel and all other channels that begin with 'bar' or 'input'. The order attribute allows you to manage where this interceptor will be injected if there are multiple interceptors on a given channel. For example, channel 'inputChannel' could have individual interceptors configured locally (see below):

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

A reasonable question is how will a global interceptor be injected in relation to other interceptors configured locally or through other global interceptor definitions? The current implementation provides a very simple mechanism for defining the order of interceptor execution. A positive number in the order attribute will ensure interceptor injection after any existing interceptors and a negative number will ensure that the interceptor is injected before existing interceptors. This means that in the above example, the global interceptor will be injected AFTER (since its order is greater than 0) the 'wire-tap' interceptor configured locally. If there were another global interceptor with a matching pattern, its order would be determined by comparing the values of the order attribute. To inject a global interceptor BEFORE the existing interceptors, use a negative value for the order attribute.

	Note
	Note that both the order and pattern attributes are optional. The default value for order will be 0 and for pattern, the default is '*' (to match all channels).

Wire Tap

As mentioned above, Spring Integration provides a simple Wire Tap interceptor out of the box. You can configure a Wire Tap on any channel within an <interceptors/> element. This is especially useful for debugging, and can be used in conjunction with Spring Integration's logging Channel Adapter as follows:

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

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

Tip

The 'logging-channel-adapter' also accepts an 'expression' attribute so that you can evaluate a SpEL expression against 'payload' and/or 'headers' variables. Alternatively, to simply log the full Message toString() result, provide a value of "true" for the 'log-full-message' attribute. That is false by default so that only the payload is logged. Setting that to true enables logging of all headers in addition to the payload. The 'expression' option does provide the most flexibility, however (e.g. expression="payload.user.name").

A little more on Wire Tap

One of the common misconceptions about the wire tap and other similar components (Section B.1, “Message Publishing Configuration”) is that they are automatically asynchronous in nature. Wire-tap as a component is not invoked asynchronously be default. Instead, Spring Integration focuses on a single unified approach to configuring asynchronous behavior: the Message Channel. What makes certain parts of the message flow sync or async is the type of Message Channel that has been configured within that flow. That is one of the primary benefits of the Message Channel abstraction. From the inception of the framework, we have always emphasized the need and the value of the Message Channel as a first-class citizen of the framework. It is not just an internal, implicit realization of the EIP pattern, it is fully exposed as a configurable component to the end user. So, the Wire-tap component is ONLY responsible for performing the following 3 tasks:

• intercept a message flow by tapping into a channel (e.g., channelA)

• grab each message

• send the message to another channel (e.g., channelB)

It is essentially a variation of the Bridge, but it is encapsulated within a channel definition (and hence easier to enable and disable without disrupting a flow). Also, unlike the bridge, it basically forks another message flow. Is that flow synchronous or asynchronous? The answer simply depends on the type of Message Channel that 'channelB' is. And, now you know that we have: Direct Channel, Pollable Channel, and Executor Channel as options. The last two do break the thread boundary making communication via such channels asynchronous simply because the dispatching of the message from that channel to its subscribed handlers happens on a different thread than the one used to send the message to that channel. That is what is going to make your wire-tap flow sync or async. It is consistent with other components within the framework (e.g., Message Publisher) and actually brings a level of consistency and simplicity by sparing you from worrying in advance (other than writing thread safe code) whether a particular piece of code should be implemented as sync or async. The actual wiring of two pieces of code (component A and component B) via Message Channel is what makes their collaboration sync or async. You may even want to change from sync to async in the future and Message Channel is what's going to allow you to do it swiftly without ever touching the code.

One final point regarding the Wire Tap is that, despite the rationale provided above for not being async be default, one should keep in mind it is usually desirable to hand off the Message as soon as possible. Therefore, it would be quite common to use an asynchronous channel option as the wire-tap's outbound channel. Nonetheless, another reason that we do not enforce asynchronous behavior by default is that you might not want to break a transactional boundary. Perhaps you are using the Wire Tap for auditing purposes, and you DO want the audit Messages to be sent within the original transaction. As an example, you might connect the wire-tap to a JMS outbound-channel-adapter. That way, you get the best of both worlds: 1) the sending of a JMS Message can occur within the transaction while 2) it is still a "fire-and-forget" action thereby preventing any noticeable delay in the main message flow.

Global Wire Tap Configuration

It is possible to configure a global wire tap as a special case of the Global Channel Interceptor. Simply configure a top level wire-tap element. Now, in addition to the normal wire-tap namespace support, the pattern and order attributes are supported and work in exactly the same way as with the channel-interceptor

<int:wire-tap pattern="input*, bar*, foo" order="3" channel="wiretapChannel"/>

	Tip
	A global wire tap provides a convenient way to configure a single channel wire tap externally without modifying the existing channel configuration. Simply set the pattern attribute to the target channel name. For example, This technique may be used to configure a test case to verify messages on a channel.

If namespace support is enabled, there are two special channels defined within the application context by default: errorChannel and nullChannel. The 'nullChannel' acts like /dev/null, simply logging any Message sent to it at DEBUG level and returning immediately. Any time you face channel resolution errors for a reply that you don't care about, you can set the affected component's output-channel attribute to 'nullChannel' (the name 'nullChannel' is reserved within the application context). The 'errorChannel' is used internally for sending error messages and may be overridden with a custom configuration. This is discussed in greater detail in Section F.4, “Error Handling”.

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 one of the scheduling attributes, such as 'fixed-rate' or 'cron'.

<int:inbound-channel-adapter ref="source1" method="method1" channel="channel1">
    <int:poller fixed-rate="5000"/>
</int:inbound-channel-adapter>

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

Also see Section 3.3.3, “Channel Adapter Expressions and Scripts”.

	Note
	If no poller is provided, then a single default poller must be registered within the context. See Section 7.1.4, “Namespace Support” for more detail.

Important

Poller Configuration Some inbound-channel-adapter types are backed by a SourcePollingChannelAdapter which means they contain Poller configuration which will poll the MessageSource (invoke a custom method which produces the value that becomes a Message payload) based on the configuration specified in the Poller. For example: <int:poller max-messages-per-poll="1" fixed-rate="1000"/> <int:poller max-messages-per-poll="10" fixed-rate="1000"/> In the the first configuration the polling task will be invoked once per poll and during such task (poll) the method (which results in the production of the Message) will be invoked once based on the max-messages-per-poll attribute value. In the second configuration the polling task will be invoked 10 times per poll or until it returns 'null' thus possibly producing 10 Messages per poll while each poll happens at 1 second intervals. However what if the configuration looks like this: <int:poller fixed-rate="1000"/> Note there is no max-messages-per-poll specified. As you'll learn later the identical poller configuration in the PollingConsumer (e.g., service-activator, filter, router etc.) would have a default value of -1 for max-messages-per-poll which means "execute poling task non-stop unless polling method returns null (e.g., no more Messages in the QueueChannel)" and then sleep for 1 second. However in the SourcePollingChannelAdapter it is a bit different. The default value for max-messages-per-poll will be set to 1 by default unless you explicitly set it to a negative value (e.g., -1). It is done so to make sure that poller can react to a LifeCycle events (e.g., start/stop) and prevent it from potentially spinning in the infinite loop if the implementation of the custom method of the MessageSource has a potential to never return null and happened to be non-interruptible. However if you are sure that your method can return null and you need the behavior where you want to poll for as many sources as available per each poll, then you should explicitly set max-messages-per-poll to negative value. <int:poller max-messages-per-poll="-1" fixed-rate="1000"/>

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.

<int:outbound-channel-adapter channel="channel1" ref="target" method="handle"/>

<beans:bean id="target" class="org.Foo"/>

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

<int:outbound-channel-adapter channel="channel2" ref="target" method="handle">
    <int:poller fixed-rate="3000"/>
</int:outbound-channel-adapter>

<beans:bean id="target" class="org.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 is only referenced by a single definition of the <outbound-channel-adapter>, you can define it as inner bean:

<int:outbound-channel-adapter channel="channel" method="handle">
    <beans:bean class="org.Foo"/>
</int:outbound-channel-adapter>

	Note
	Using both the "ref" attribute and an inner handler definition in the same <outbound-channel-adapter> configuration is not allowed as it creates an ambiguous condition. Such a configuration will result in an Exception being thrown.

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.

Like many other Spring Integration components, the <inbound-channel-adapter> and <outbound-channel-adapter> also provide support for SpEL expression evaluation. To use SpEL, provide the expression string via the 'expression' attribute instead of providing the 'ref' and 'method' attributes that are used for method-invocation on a bean. When an Expression is evaluated, it follows the same contract as method-invocation where: the expression for an <inbound-channel-adapter> will generate a message anytime the evaluation result is a non-null value, while the expression for an <outbound-channel-adapter> must be the equivalent of a void returning method invocation.

Starting with Spring Integration 3.0, an <int:inbound-channel-adapter/> can also be configured with a SpEL <expression/> (or even with <script/>) sub-element, for when more sophistication is required than can be achieved with the simple 'expression' attribute. If you provide a script as a Resource using the location attribute, you can also set the refresh-check-delay allowing the resource to be refreshed periodically. If you want the script to be checked on each poll, you would need to coordinate this setting with the poller's trigger:

<int:inbound-channel-adapter ref="source1" method="method1" channel="channel1">
    <int:poller max-messages-per-poll="1" fixed-delay="5000"/>
    <script:script lang="ruby" location="Foo.rb" refresh-check-delay="5000"/>
</int:inbound-channel-adapter>

Also see the cacheSeconds property on the ReloadableResourceBundleExpressionSource when using the <expression/> sub-element. For more information regarding expressions see Appendix A, Spring Expression Language (SpEL), and for scripts - Section 7.6, “Groovy support” and Section 7.5, “Scripting support”.

Important

The <int:inbound-channel-adapter/> is an endpoint that starts a message flow via periodic triggering to poll some underlying MessageSource. Since, at the time of polling, there is not yet a message object, expressions and scripts don't have access to a root Message, so there are no payload or headers properties that are available in most other messaging SpEL expressions. Of course, the script can generate and return a complete Message object with headers and payload, or just a payload, which will be added to a message with basic headers.

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:

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

<int:bridge input-channel="pollable" output-channel="subscribable">
     <int:poller max-messages-per-poll="10" fixed-rate="5000"/>
 </int: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.

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

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

 <int: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.

	Note
	If no 'output-channel' is defined on a bridge, the reply channel provided by the inbound Message will be used, if available. If neither output or reply channel is available, an Exception will be thrown.

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.

Message ID Generation

When a message transitions through an application, each time it is mutated (e.g. by a transformer) a new message id is assigned. The message id is a UUID. Beginning with Spring Integration 3.0, the default strategy used for id generation is more efficient than the previous java.util.UUID.randomUUID() implementation. It uses simple random numbers based on a secure random seed, instead of creating a secure random number each time.

A different UUID generation strategy can be selected by declaring a bean that implements MessageHeaders.IdGenerator in the application context.

Important

Only one UUID generation strategy can be used in a classloader. This means that if two or more application contexts are running in the same classloader, they will share the same strategy. If one of the contexts changes the strategy, it will be used by all contexts. If two or more contexts in the same classloader declare a bean of type MessageHeaders.IdGenerator, they must all be an instance of the same class, otherwise the context attempting to replace a custom strategy will fail to initialize. If the strategy is the same, but parameterized, the strategy in the first context to initialize will be used.

In addition to the default strategy, two additional IdGenerators are provided; MessageHeaders.JdkIdGenerator uses the previous UUID.randomUUID() mechanism; MessageHeaders.SimpleIncrementingIdGenerator can be used in cases where a UUID is not really needed and a simple incrementing value is sufficient.

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 is also a convenient implementation of Message designed to communicate error conditions. This implementation takes Throwable object as its payload:

ErrorMessage message = new ErrorMessage(someThrowable);

Throwable t = message.getPayload();

Notice that this implementation takes 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(5)
        .build();

assertEquals(5, importantMessage.getHeaders().getPriority());

Message<Integer> lessImportantMessage = MessageBuilder.fromMessage(importantMessage)
        .setHeaderIfAbsent(MessageHeaders.PRIORITY, 2)
        .build();

assertEquals(2, lessImportantMessage.getHeaders().getPriority());

The priority header is only considered when using a PriorityChannel (as described in the next chapter). It is defined as java.lang.Integer.

Routers are a crucial element in many messaging architectures. They consume Messages from a Message Channel and forward each consumed message to one or more different Message Channel depending on a set of conditions.

Spring Integration provides the following routers out-of-the-box:

Router implementations share many configuration parameters. Yet, certain differences exist between routers. Furthermore, the availability of configuration parameters depends on whether Routers are used inside or outside of a chain. In order to provide a quick overview, all available attributes are listed in the 2 tables below.

Table 5.1. Routers Outside of a Chain

Attribute	router	header value router	xpath router	payload type router	recipient list router	exception type router
apply-sequence
default-output-channel
resolution-required
ignore-send-failures
timeout
id
auto-startup
input-channel
order
method
ref
expression
header-name
evaluate-as-string
xpath-expression-ref
converter

Table 5.2. Routers Inside of a Chain

Attribute	router	header value router	xpath router	payload type router	recipient list router	exception type router
apply-sequence
default-output-channel
resolution-required
ignore-send-failures
timeout
id
auto-startup
input-channel
order
method
ref
expression
header-name
evaluate-as-string
xpath-expression-ref
converter

Important

Router parameters have been more standardized across all router implementations with Spring Integration 2.1. Consequently, there are a few minor changes that leave the possibility of breaking older Spring Integration based applications. Since Spring Integration 2.1 the ignore-channel-name-resolution-failures attribute is removed in favor of consolidating its behavior with the resolution-required attribute. Also, the resolution-required attribute now defaults to true. Prior to these changes, the resolution-required attribute defaulted to false causing messages to be silently dropped when no channel was resolved and no default-output-channel was set. The new behavior will require at least one resolved channel and by default will throw an MessageDeliveryException if no channel was determined (or an attempt to send was not successful). If you do desire to drop messages silently simply set default-output-channel="nullChannel".

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 common requirements.

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

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

HeaderValueRouter

A HeaderValueRouter will send Messages to the channel based on the individual header value mappings. When a HeaderValueRouter is created it is initialized with the name of the header to be evaluated. The value of the header could be one of two things:

1. Arbitrary value

2. Channel name

If arbitrary then additional mappings for these header values to channel names is required, otherwise no additional configuration is needed.

Spring Integration provides a simple namespace-based XML configuration to configure a HeaderValueRouter. The example below demonstrates two types of namespace-based configuration for the HeaderValueRouter.

1. Configuration where mapping of header values to channels is required

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

During the resolution process this router may encounter channel resolution failures, causing an exception. If you want to suppress such exceptions and send unresolved messages to the default output channel (identified with the default-output-channel attribute) set resolution-required to false.

Normally, messages for which the header value is not explicitly mapped to a channel will be sent to the default-output-channel. However, in cases where the header value is mapped to a channel name but the channel cannot be resolved, setting the resolution-required attribute to false will result in routing such messages to the default-output-channel.

	Important
	With Spring Integration 2.1 the attribute was changed from ignore-channel-name-resolution-failures to resolution-required. Attribute resolution-required will default to true.

2. Configuration where mapping of header values to channel names is not required since header values themselves represent channel names

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

Note

Since Spring Integration 2.1 the behavior of resolving channels is more explicit. For example, if you ommit the default-output-channel attribute and the Router was unable to resolve at least one valid channel, and any channel name resolution failures were ignored by setting resolution-required to false, then a MessageDeliveryException is thrown. Basically, by default the Router must be able to route messages successfully to at least one channel. If you really want to drop messages, you must also have default-output-channel set to nullChannel.

RecipientListRouter

A RecipientListRouter will send each received Message to a statically defined list of Message Channels:

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

Spring Integration also provides namespace support for the RecipientListRouter configuration (see Section F.2, “Namespace Support”) as the example below demonstrates.

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

	Note
	The 'apply-sequence' flag here has the same effect as it does for a publish-subscribe-channel, and like a publish-subscribe-channel, it is disabled by default on the recipient-list-router. Refer to the section called “PublishSubscribeChannel Configuration” for more information.

Another convenient option when configuring a RecipientListRouter is to use Spring Expression Language (SpEL) support as selectors for individual recipient channels. This is similar to using a Filter at the beginning of 'chain' to act as a "Selective Consumer". However, in this case, it's all combined rather concisely into the router's configuration.

<int:recipient-list-router id="customRouter" input-channel="routingChannel">
    <int:recipient channel="channel1" selector-expression="payload.equals('foo')"/>
    <int:recipient channel="channel2" selector-expression="headers.containsKey('bar')"/>
</int:recipient-list-router>

In the above configuration a SpEL expression identified by the selector-expression attribute will be evaluated to determine if this recipient should be included in the recipient list for a given input Message. The evaluation result of the expression must be a boolean. If this attribute is not defined, the channel will always be among the list of recipients.

<int:exception-type-router input-channel="inputChannel"
                           default-output-channel="defaultChannel">
    <int:mapping exception-type="java.lang.IllegalArgumentException"
                 channel="illegalChannel"/>
    <int:mapping exception-type="java.lang.NullPointerException"
                 channel="npeChannel"/>
</int:exception-type-router>

<int:channel id="illegalChannel" />
<int:channel id="npeChannel" />

So as you can see, Spring Integration provides quite a few different router configurations for common content-based routing use cases as well as the option of implementing custom routers as POJOs. For example PayloadTypeRouter provides a simple way to configure a router which computes channels based on the payload type of the incoming Message while HeaderValueRouter provides the same convenience in configuring a router which computes channels by evaluating the value of a particular Message Header. There are also expression-based (SpEL) routers where the channel is determined based on evaluating an expression. Thus, these type of routers exhibit some dynamic characteristics.

However these routers all require static configuration. Even in the case of expression-based routers, the expression itself is defined as part of the router configuration which means that the same expression operating on the same value will always result in the computation of the same channel. This is acceptable in most cases since such routes are well defined and therefore predictable. But there are times when we need to change router configurations dynamically so message flows may be routed to a different channel.

Example:

You might want to bring down some part of your system for maintenance and temporarily re-reroute messages to a different message flow. Or you may want to introduce more granularity to your message flow by adding another route to handle a more concrete type of java.lang.Number (in the case of PayloadTypeRouter).

Unfortunately with static router configuration to accomplish this, you would have to bring down your entire application, change the configuration of the router (change routes) and bring it back up. This is obviously not the solution.

The Dynamic Router pattern describes the mechanisms by which one can change/configure routers dynamically without bringing down the system or individual routers. 

Before we get into the specifics of how this is accomplished in Spring Integration, let's quickly summarize the typical flow of the router, which consists of 3 simple steps:

There is not much that can be done with regard to dynamic routing if Step 1 results in the actual instance of the MessageChannel, simply because the MessageChannel is the final product of any router's job. However, if Step 1 results in a channel identifier that is not an instance of MessageChannel, then there are quite a few possibilities to influence the process of deriving the Message Channel. Lets look at couple of the examples in the context of the 3 steps mentioned above: 

Payload Type Router

<int:payload-type-router input-channel="routingChannel">
    <int:mapping type="java.lang.String"  channel="channel1" />
    <int:mapping type="java.lang.Integer" channel="channel2" />
</int:payload-type-router>

Within the context of the Payload Type Router the 3 steps mentioned above would be realized as:

In other words, each step feeds the next step until the process completes.

Header Value Router

<int:header-value-router input-channel="inputChannel" header-name="testHeader">
    <int:mapping value="foo" channel="fooChannel" />
    <int:mapping value="bar" channel="barChannel" />
</int:header-value-router>

Similar to the PayloadTypeRouter:

The above two configurations of two different router types look almost identical. However if we look at the alternate configuration of the HeaderValueRouter we clearly see that there is no mapping sub element:

<int:header-value-router input-channel="inputChannel" header-name="testHeader">

But the configuration is still perfectly valid. So the natural question is what about the mapping in the Step 2?

What this means is that Step 2 is now an optional step. If mapping is not defined then the channel identifier value computed in Step 1 will automatically be treated as the channel name, which will now be resolved to the actual MessageChannel as in Step 3. What it also means is that Step 2 is one of the key steps to provide dynamic characteristics to the routers, since it introduces a process which allows you to change the way 'channel identifier' resolves to 'channel name', thus influencing the process of determining the final instance of the MessageChannel from the initial channel identifier. 

For Example:

In the above configuration let's assume that the testHeader value is 'kermit' which is now a channel identifier (Step 1). Since there is no mapping in this router, resolving this channel identifier to a channel name (Step 2) is impossible and this channel identifier is now treated as channel name. However what if there was a mapping but for a different value? The end result would still be the same and that is: if a new value cannot be determined through the process of resolving the 'channel identifier' to a 'channel name', such 'channel identifier' becomes 'channel name'.

So all that is left is for Step 3 to resolve the channel name ('kermit') to an actual instance of the MessageChannel identified by this name. That basically involves a bean lookup for the name provided. So now all messages which contain the header/value pair as testHeader=kermit are going to be routed to a MessageChannel whose bean name (id) is 'kermit'.

But what if you want to route these messages to the 'simpson' channel? Obviously changing a static configuration will work, but will also require bringing your system down. However if you had access to the channel identifier map, then you could just introduce a new mapping where the header/value pair is now kermit=simpson, thus allowing Step 2 to treat 'kermit' as a channel identifier while resolving it to 'simpson' as the channel name .

The same obviously applies for PayloadTypeRouter, where you can now remap or remove a particular payload type mapping. In fact, it applies to every other router, including expression-based routers, since their computed values will now have a chance to go through Step 2 to be additionally resolved to the actual channel name.

In Spring Integration 2.0 the router hierarchy underwent significant refactoring, so that now any router that is a subclass of the AbstractMessageRouter (which includes all framework defined routers) is a Dynamic Router simply because the channelIdentiferMap is defined at the AbstractMessageRouter level. That map's setter method is exposed as a public method along with 'setChannelMapping' and 'removeChannelMapping' methods. These allow you to change/add/remove router mappings at runtime as long as you have a reference to the router itself. It also means that you could expose these same configuration options via JMX (see Section 8.1, “JMX Support”) or the Spring Integration ControlBus (see Section 8.5, “Control Bus”) functionality. 

Manage Router Mappings using the Control Bus

One way to manage the router mappings is through the Control Bus pattern which exposes a Control Channel where you can send control messages to manage and monitor Spring Integration components, including routers.

	Note
	For more information about the Control Bus, please see chapter Section 8.5, “Control Bus”.

Typically you would send a control message asking to invoke a particular operation on a particular managed component (e.g. router). The two managed operations (methods) that are specific to changing the router resolution process are:

• public void setChannelMapping(String channelIdentifier, String channelName) - will allow you to add a new or modify an existing mapping between channel identifier and channel name

• public void removeChannelMapping(String channelIdentifier) - will allow you to remove a particular channel mapping, thus disconnecting the relationship between channel identifier and channel name

Manage Router Mappings using JMX

You can also expose a router instance with Spring's JMX support, and then use your favorite JMX client (e.g., JConsole) to manage those operations (methods) for changing the router's configuration.

	Note
	For more information about Spring Integration's JMX suppor, please see chapter JMX Support.

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 Message content 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 the Message to but only decides whether to send.

	Note
	As you will see momentarily, the Filter also supports a discard channel, so in certain cases it can play the role of a very simple router (or "switch") based on a boolean condition.

In Spring Integration, a Message Filter may be configured as a Message Endpoint that delegates to an 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 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 of 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 enables tracking 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);

}

To implement a specific Splitter in an application, extend AbstractMessageSplitter and implement the splitMessage method, which contains logic for splitting the messages. The return value may 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.

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, an Aggregator is often a downstream consumer in a pipeline that includes a Splitter.

Technically, the Aggregator is more complex than a Splitter, because it is stateful as it must hold the Messages to be aggregated and determine when the complete group of Messages is ready to be aggregated. 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 the aggregated message as output.

Implementing an Aggregator requires providing the logic to perform the aggregation (i.e., the creation of a single message from many). Two related concepts are correlation and release.

Correlation determines how messages are grouped for aggregation. In Spring Integration correlation is done by default based on the MessageHeaders.CORRELATION_ID message header. Messages with the same MessageHeaders.CORRELATION_ID will be grouped together. However, the correlation strategy may be customized to allow other ways of specifying how the messages should be grouped together by implementing a CorrelationStrategy (see below).

To determine the point at which a group of messages is ready to be processed, a ReleaseStrategy is consulted. The default release strategy for the Aggregator will release a group when all messages included in a sequence are present, based on the MessageHeaders.SEQUENCE_SIZE header. This default strategy may be overridden by providing a reference to a custom ReleaseStrategy implementation.

The Aggregation API consists of a number of classes:

AggregatingMessageHandler

The AggregatingMessageHandler (subclass of AbstractCorrelatingMessageHandler) is a MessageHandler implementation, encapsulating the common functionalities of an Aggregator (and other correlating use cases), which are:

• correlating messages into a group to be aggregated

• maintaining those messages in a MessageStore until the group can be released

• deciding when the group can be released

• aggregating the released group into a single message

• recognizing and responding to an expired group

The responsibility of deciding how the messages should be grouped together is delegated to a CorrelationStrategy instance. The responsibility of deciding whether the message group can be released is delegated to a ReleaseStrategy instance.

Here is a brief highlight of the base AbstractAggregatingMessageGroupProcessor (the responsibility of implementing the aggregatePayloads method is left to the developer):

public abstract class AbstractAggregatingMessageGroupProcessor
              implements MessageGroupProcessor {

    protected Map<String, Object> aggregateHeaders(MessageGroup group) {
        // default implementation exists
    }

    protected abstract Object aggregatePayloads(MessageGroup group, Map<String, Object> defaultHeaders);

}

The CorrelationStrategy is owned by the AbstractCorrelatingMessageHandler and it has a default value based on the MessageHeaders.CORRELATION_ID message header:

public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
        CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
    ...
    this.correlationStrategy = correlationStrategy == null ?
        new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID) : correlationStrategy;
    this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy;
    ...
}

As for actual processing of the message group, the default implementation is the DefaultAggregatingMessageGroupProcessor. It creates a single Message whose payload is a List of the payloads received for a given group. This works well for simple Scatter Gather implementations with either a Splitter, Publish Subscribe Channel, or Recipient List Router upstream.

Note

When using a Publish Subscribe Channel or Recipient List Router in this type of scenario, be sure to enable the flag to apply-sequence. That will add the necessary headers (CORRELATION_ID, SEQUENCE_NUMBER and SEQUENCE_SIZE). That behavior is enabled by default for Splitters in Spring Integration, but it is not enabled for the Publish Subscribe Channel or Recipient List Router because those components may be used in a variety of contexts in which these headers are not necessary.

When implementing a specific aggregator strategy for an application, a developer can extend AbstractAggregatingMessageGroupProcessor and implement the aggregatePayloads method. However, there are better solutions, less coupled to the API, for implementing the aggregation logic which can be configured easily either through XML or through annotations.

In general, any POJO can implement the aggregation algorithm if it provides a method that accepts a single java.util.List as an argument (parameterized lists are supported as well). This method will be invoked for aggregating messages as follows:

• if the argument is a java.util.List<T>, and the parameter type T is assignable to Message, then the whole list of messages accumulated for aggregation will be sent to the aggregator

• if the argument is a non-parameterized java.util.List or the parameter type is not assignable to Message, then the method will receive the payloads of the accumulated messages

• if the return type is not assignable to Message, then it will be treated as the payload for a Message that will be created automatically by the framework.

	Note
	In the interest of code simplicity, and promoting best practices such as low coupling, testability, etc., the preferred way of implementing the aggregation logic is through a POJO, and using the XML or annotation support for configuring it in the application.

ReleaseStrategy

The ReleaseStrategy interface is defined as follows:

public interface ReleaseStrategy {

  boolean canRelease(MessageGroup group);

}

In general, any POJO can implement the completion decision logic if it provides a method that accepts a single java.util.List as an argument (parameterized lists are supported as well), and returns a boolean value. This method will be invoked after the arrival of each new message, to decide whether the group is complete or not, as follows:

• if the argument is a java.util.List<T>, and the parameter type T is assignable to Message, then the whole list of messages accumulated in the group will be sent to the method

• if the argument is a non-parametrized java.util.List or the parameter type is not assignable to Message, then the method will receive the payloads of the accumulated messages

• the method must return true if the message group is ready for aggregation, and false otherwise.

For example:

public class MyReleaseStrategy {

    @ReleaseStrategy
    public boolean canMessagesBeReleased(List<Message<?>>) {...}
}

public class MyReleaseStrategy {

    @ReleaseStrategy
    public boolean canMessagesBeReleased(List<String>) {...}
}

As you can see based on the above signatures, the POJO-based Release Strategy will be passed a Collection of not-yet-released Messages (if you need access to the whole Message) or a Collection of payload objects (if the type parameter is anything other than Message). Typically this would satisfy the majority of use cases. However if, for some reason, you need to access the full MessageGroup then you should simply provide an implementation of the ReleaseStrategy interface.

When the group is released for aggregation, all its not-yet-released messages are processed and removed from the group. If the group is also complete (i.e. if all messages from a sequence have arrived or if there is no sequence defined), then the group is marked as complete. Any new messages for this group will be sent to the discard channel (if defined). Setting expire-groups-upon-completion to true (default is false) removes the entire group and any new messages, with the same correlation id as the removed group, will form a new group. Partial sequences can be released by using a MessageGroupStoreReaper together with send-partial-result-on-expiry being set to true.

Important

To facilitate discarding of late-arriving messages, the aggregator must maintain state about the group after it has been released. This can eventually cause out of memory conditions. To avoid such situations, you should consider configuring a MessageGroupStoreReaper to remove the group metadata; the expiry parameters should be set to expire groups after it is not expected that late messages will arrive. For information about configuring a reaper, see Section 5.4.5, “Managing State in an Aggregator: MessageGroupStore”.

Spring Integration provides an out-of-the box implementation for ReleaseStrategy, the SequenceSizeReleaseStrategy. This implementation consults the SEQUENCE_NUMBER and SEQUENCE_SIZE headers of each arriving message to decide when a message group is complete and ready to be aggregated. As shown above, it is also the default strategy.

public interface CorrelationStrategy {

  Object getCorrelationKey(Message<?> message);

}

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 (whether defined by the framework or a user) should be able to remain stateless. All state is carried by the MessageGroup and its management is delegated to the MessageGroupStore.

public interface MessageGroupStore {
    int getMessageCountForAllMessageGroups();

    int getMarkedMessageCountForAllMessageGroups();

    int getMessageGroupCount();

    MessageGroup getMessageGroup(Object groupId);

    MessageGroup addMessageToGroup(Object groupId, Message<?> message);

    MessageGroup markMessageGroup(MessageGroup group);

    MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove);

    MessageGroup markMessageFromGroup(Object key, Message<?> messageToMark);

    void removeMessageGroup(Object groupId);

    void registerMessageGroupExpiryCallback(MessageGroupCallback callback);

    int expireMessageGroups(long timeout);
}

For more information please refer to the JavaDoc.

The MessageGroupStore accumulates state information in MessageGroups while waiting for a release strategy to be triggered, and that event might not ever happen. So to prevent stale messages from lingering, and for volatile stores to provide a hook for cleaning up when the application shuts down, the MessageGroupStore allows the user to register callbacks to apply to its MessageGroups when they expire. The interface is very straightforward:

public interface MessageGroupCallback {

    void execute(MessageGroupStore messageGroupStore, MessageGroup group);

}

The callback has direct access 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, on demand, to all messages whose timestamp is earlier than a time supplied as a parameter (see the registerMessageGroupExpiryCallback(..) and expireMessageGroups(..) methods above).

The expireMessageGroups method can be called with a timeout value. Any message older than the current time minus this value will 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="30000"/>
</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 in the example above is that the message group store's expire method is being called once every 10 seconds. The timeout itself is 30 seconds.

Note

It is important to understand that the 'timeout' property of the MessageGroupStoreReaper is an approximate value and is impacted by the the rate of the task scheduler since this property will only be checked on the next scheduled execution of the MessageGroupStoreReaper task. For example if the timeout is set for 10 min, but the MessageGroupStoreReaper task is scheduled to run every 60 min and the last execution of the MessageGroupStoreReaper task happened 1 min before the timeout, the MessageGroup will not expire for the next 59 min. So it is recommended to set the rate at least equal to the value of the timeout or shorter.

In addition to the reaper, the expiry callbacks are invoked when the application shuts down via a lifecycle callback in the AbstractCorrelatingMessageHandler.

The AbstractCorrelatingMessageHandler 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, any unmarked messages in groups that are not yet released can be sent on to the output channel.

Important

When using a MessageGroupStoreReaper, it is generally recommended to use a separate MessageStore for each correlating endpoint. Otherwise, unexpected results may occur because one endpoint may remove another endpoint's groups. Some MessageStore implementations allow using the same physical resources, by partitioning the data; for example, the JdbcMessageStore has a region property; the MongoDbMessageStore has a collectionName property. For more information about MessageStore interface and its implementations, please read Section 8.3, “Message Store”.

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.

<int:channel id="inputChannel"/>

<int:channel id="outputChannel"/>

<int:resequencer id="completelyDefinedResequencer" 
  input-channel="inputChannel" 
  output-channel="outputChannel" 
  discard-channel="discardChannel" 
  release-partial-sequences="true" 
  message-store="messageStore" 
  send-partial-result-on-expiry="true" 
  send-timeout="86420000" 
  correlation-strategy="correlationStrategyBean" 
  correlation-strategy-method="correlate" 
  correlation-strategy-expression="headers['foo']" 
  release-strategy="releaseStrategyBean" 
  release-strategy-method="release" 
  release-strategy-expression="size() == 10" 
  empty-group-min-timeout="60000" />

	Note
	Since there is no custom behavior to be implemented in Java classes for resequencers, there is no annotation support for it.

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 eliminating the need to define channels for each individual component.

Tip

Spring Integration's Filter provides a boolean property throwExceptionOnRejection. When providing multiple Selective Consumers on the same point-to-point channel with different acceptance criteria, this value should be set to 'true' (the default is false) so that the dispatcher will know that the Message was rejected and as a result will attempt to pass the Message on to other subscribers. If the Exception were not thrown, then it would appear to the dispatcher as if the Message had been passed on successfully even though the Filter had dropped the Message to prevent further processing. If you do indeed want to "drop" the Messages, then the Filter's 'discard-channel' might be useful since it does give you a chance to perform some operation with the dropped message (e.g. send to a JMS queue or simply write to a log).

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 anonymous 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 required to implement the MessageProducer interface (which provides a 'setOutputChannel()' method). The last handler only needs an output channel if the outputChannel on the MessageHandlerChain is set.

	Note
	As with other endpoints, the output-channel is optional. If there is a reply Message at the end of the chain, the output-channel takes precedence, but if not available, the chain handler will check for a reply channel header on the inbound Message as a fallback.

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 or an outbound-channel-adapter.

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

The <header-enricher> element used in the above example will set a message header named "foo" with a value of "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, but the header-enricher is obviously a simpler option.

The <chain> can be configured as the last 'black-box' consumer of the message flow. For this solution it is enough to put at the end of the <chain> some <outbound-channel-adapter>:

<int:chain input-channel="input">
    <si-xml:marshalling-transformer marshaller="marshaller" result-type="StringResult" />
    <int:service-activator ref="someService" method="someMethod"/>
    <int:header-enricher>
        <int:header name="foo" value="bar"/>
    </int:header-enricher>
    <int:logging-channel-adapter level="INFO" log-full-message="true"/>
</int:chain>

Disallowed Attributes and Elements

It is important to note that certain attributes, such as order and input-channel are not allowed to be specified on components used within a chain. The same is true for the poller sub-element.

Important

For the Spring Integration core components, the XML Schema itself will enforce some of these constraints. However, for non-core components or your own custom components, these constraints are enforced by the XML namespace parser, not by the XML Schema. These XML namespace parser constraints were added with Spring Integration 2.2. The XML namespace parser will throw an BeanDefinitionParsingException if you try to use disallowed attributes and elements.

'id' Attribute

Beginning with Spring Integration 3.0, if a chain element is given an id, the bean name for the element is a combination of the chain's id and the id of the element itself. Elements without an id are not registered as beans, but they are given componentNames that include the chain id. For example:

<int:chain id="fooChain" input-channel="input">
    <int:service-activator id="fooService" ref="someService" method="someMethod"/>
    <int:object-to-json-transformer/>
</int:chain>

The id attribute for <chain> elements allows them to be eligible for JMX export and they are trackable via Message History. They can also be accessed from the BeanFactory using the appropriate bean name as discussed above.

	Tip
	It is useful to provide an explicit id attribute on <chain>s to simplify the identification of sub-components in logs, and to provide access to them from the BeanFactory etc.

Calling a Chain from within a Chain

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

<int:chain id="main-chain" input-channel="in" output-channel="out">
    <int:header-enricher>
      <int:header name="name" value="Many" />
    </int:header-enricher>
    <int:service-activator>
      <bean class="org.foo.SampleService" />
    </int:service-activator>
    <int:gateway request-channel="inputA"/>  
</int:chain>

<int:chain id="nested-chain-a" input-channel="inputA">
    <int:header-enricher>
        <int:header name="name" value="Moe" />
    </int:header-enricher>
    <int:gateway request-channel="inputB"/> 
    <int:service-activator>
        <bean class="org.foo.SampleService" />
    </int:service-activator>
</int:chain>

<int:chain id="nested-chain-b" input-channel="inputB">
    <int:header-enricher>
        <int:header name="name" value="Jack" />
    </int:header-enricher>
    <int:service-activator>
        <bean class="org.foo.SampleService" />
    </int:service-activator>
</int:chain>