The separate Spring Integration Java DSL project has now been merged into the core Spring Integration project. The IntegrationComponentSpec implementations for channel adapters and gateways are distributed to their specific modules. See Chapter 9, Java DSL for more information about Java DSL support. Also see the 4.3 to 5.0 Migration Guide for the required steps to move to Spring Integration 5.0.

A new Spring Integration Test Framework has been created to assist with testing Spring Integration applications. Now, with the @SpringIntegrationTest annotation on test class and MockIntegration factory you can make your JUnit tests for integration flows somewhat easier.

See Appendix F, Testing support for more information.

The new MongoDbOutboundGateway allows you to make queries to the database on demand by sending a message to its request channel.

See Section 23.6, “MongoDB Outbound Gateway” for more information.

The new WebFlux support module has been introduced for Spring WebFlux Framework gateways and channel adapters.

See Chapter 34, WebFlux Support for more information.

Now that we use the new InvocableHandlerMethod -based infrastructure for service method invocations, we can perform contentType conversion from payload to target method argument.

See Section 8.1.7, “Content Type Conversion” for more information.

The ErrorMessagePublisher and the ErrorMessageStrategy are provided for creating ErrorMessage instances.

See Section E.4, “Error Handling” for more information.

A JDBC implementation of MetadataStore implementation is now provided. This is useful when it is necessary to ensure transactional boundaries for metadata.

See Section 19.7, “JDBC Metadata Store” for more information.

The @Poller annotation now has the errorChannel attribute for easier configuration of the underlying MessagePublishingErrorHandler.

See Section E.6, “Annotation Support” for more information.

All the request-reply endpoints (based on AbstractReplyProducingMessageHandler) can now start transaction and, therefore, make the whole downstream flow transactional.

See Section 8.9.6, “Transaction Support” for more information.

The SmartLifecycleRoleController now provides methods to obtain status of endpoints in roles.

See Section 8.2, “Endpoint Roles” for more information.

POJO methods are now invoked using an InvocableHandlerMethod by default, but can be configured to use SpEL as before.

See Section 3.8, “POJO Method invocation” for more information.

When targeting POJO methods as message handlers, one of the service methods can now be marked with the @Default annotation to provide a fallback mechanism for non-matched conditions.

See Section 8.5.2, “Configuring Service Activator” for more information.

A simple PassThroughTransactionSynchronizationFactory is provided to always store a polled message in the current transaction context. That message is used as a failedMessage property of the MessagingException which wraps a raw exception thrown during transaction completion.

See Section C.3, “Transaction Synchronization” for more information.

The aggregator expression-based ReleaseStrategy now evaluates the expression against the MessageGroup instead of just the collection of Message<?>.

See the section called “Aggregators and Spring Expression Language (SpEL)” for more information.

The ObjectToMapTransformer can now be supplied with a customised JsonObjectMapper.

See the section called “Aggregators and Spring Expression Language (SpEL)” for more information.

The @GlobalChannelInterceptor annotation and <int:channel-interceptor> now support negative patterns (via ! prepending) for component names matching.

See the section called “Global Channel Interceptor Configuration” for more information.

A new OnFailedToAcquireMutexEvent is emitted now via DefaultLeaderEventPublisher by the LockRegistryLeaderInitiator, when candidate is failed to acquire the lock.

See Section 8.3, “Leadership Event Handling” for more information.

The gateway now correctly sets the errorChannel header when the gateway method has a void return type and an error channel is provided. Previously, the header was not populated. This had the effect that synchronous downstream flows (running on the calling thread) would send the exception to the configured channel but an exception on an async downstream flow would be sent to the default errorChannel instead.

The RequestReplyExchanger interface now has a throws MessagingException clause to meet all the proposed messages exchange contract.

The request and reply timeouts can now be specified as SpEL expressions.

See Section 8.4, “Messaging Gateways” for more information.

Aggregators now use a SimpleSequenceSizeReleaseStrategy by default, which is more efficient, especially with large groups. Empty groups are now scheduled for removal after empty-group-min-timeout.

See Section 6.4, “Aggregator” for more information.

The Splitter component now can handle and split Java Stream and Reactive Streams Publisher objects. If the output channel is a ReactiveStreamsSubscribableChannel, the AbstractMessageSplitter builds a Flux for subsequent iteration instead of a regular Iterator independent of object being split. In addition, AbstractMessageSplitter provides protected obtainSizeIfPossible() methods to allow the determination of the size of the Iterable and Iterator objects if that is possible.

See Section 6.3, “Splitter” for more information.

Previously, Spring Integration JMS XML configuration used a default bean name connectionFactory for the JMS Connection Factory, allowing the property to be omitted from component definitions. It has now been renamed to jmsConnectionFactory, which is the bean name used by Spring Boot to auto-configure the JMS Connection Factory bean.

If your application is relying on the previous behavior, rename your connectionFactory bean to jmsConnectionFactory, or specifically configure your components to use your bean using its current name.

See Chapter 21, JMS Support for more information.

Some inconsistencies with rendering IMAP mail content have been resolved.

See the note in the Mail-Receiving Channel Adapter Section for more information.

Instead of the com.rometools.fetcher.FeedFetcher, which is deprecated in ROME, a new Resource property has been introduced to the FeedEntryMessageSource.

See Chapter 14, Feed Adapter for more information.

The new FileHeaders.RELATIVE_PATH Message header has been introduced to represent relative path in the FileReadingMessageSource.

The tail adapter now supports idleEventInterval to emit events when there is no data in the file during that period.

The flush predicates for the FileWritingMessageHandler now have an additional parameter.

The file outbound channel adapter and gateway (FileWritingMessageHandler) now support the REPLACE_IF_MODIFIED FileExistsMode.

They also now support setting file permissions on the newly written file.

A new FileSystemMarkerFilePresentFileListFilter is now available; see Section 15.2.7, “Dealing With Incomplete Data” for more information.

The FileSplitter now provides a firstLineAsHeader option to carry the first line of content as a header in the messages emitted for the remaining lines.

See Chapter 15, File Support for more information.

The Inbound Channel Adapters now have a property max-fetch-size which is used to limit the number of files fetched during a poll when there are no files currently in the local directory. They also are configured with a FileSystemPersistentAcceptOnceFileListFilter in the local-filter by default.

You can also provide a custom DirectoryScanner implementation to Inbound Channel Adapters via the newly introduced scanner attribute.

The regex and pattern filters can now be configured to always pass directories. This can be useful when using recursion in the outbound gateways.

All the Inbound Channel Adapters (streaming and synchronization-based) now use an appropriate AbstractPersistentAcceptOnceFileListFilter implementation by default to prevent remote files duplicate downloads.

The FTP and SFTP outbound gateways now support the REPLACE_IF_MODIFIED FileExistsMode when fetching remote files.

The (S)FTP streaming inbound channel adapters now add remote file information in a message header.

The FTP and SFTP outbound channel adapters, as well as PUT command of the outbound gateways, now support InputStream as payload, too.

The inbound channel adapters now can build file tree locally using a newly introduced RecursiveDirectoryScanner. See scanner option for injection. Also these adapters can now be switched to the WatchService instead.

The NLST command has been added to the AbstractRemoteFileOutboundGateway to perform only list files names remote command.

The FtpOutboundGateway can now be supplied with workingDirExpression to change the FTP client working directory for the current request message.

The RemoteFileTemplate is supplied now with the invoke(OperationsCallback<F, T> action) to perform several RemoteFileOperations calls in the scope of the same, thread-bounded, Session.

New filters for detecting incomplete remote files are now provided.

The FtpOutboundGateway and SftpOutboundGateway now support an option to remove the remote file after a successful transfer using the GET or MGET commands.

See Chapter 16, FTP/FTPS Adapters and Chapter 28, SFTP Adapters for more information.

Since version 4.3.2 a new spring.integration.readOnly.headers global property has been added to customize the list of headers which should not be copied to a newly created Message by the MessageBuilder.

See Section E.5, “Global Properties” for more information.

There is a new option on the CharacterStreamReadingMessageSource to allow it to be used to "pipe" stdin and publish an application event when the pipe is closed.

See Section 30.2, “Reading from streams” for more information.

The BarrierMessageHandler now supports a discard channel to which late-arriving trigger messages are sent.

See Section 6.8, “Thread Barrier” for more information.

The AMQP outbound endpoints now support setting a delay expression for when using the RabbitMQ Delayed Message Exchange plugin.

The inbound endpoints now support the Spring AMQP DirectMessageListenerContainer.

Pollable AMQP-backed channels now block the poller thread for the poller’s configured receiveTimeout (default 1 second).

Headers, such as contentType that are added to message properties by the message converter are now used in the final message; previously, it depended on the converter type as to which headers/message properties appeared in the final message. To override headers set by the converter, set the headersMappedLast property to true.

See Chapter 12, AMQP Support for more information.

The DefaultHttpHeaderMapper.userDefinedHeaderPrefix property is now an empty string by default instead of X-.

See Section 18.8, “HTTP Header Mappings” for more information.

Inbound messages are now mapped with headers RECEIVED_TOPIC, RECEIVED_QOS and RECEIVED_RETAINED to avoid inadvertent propagation to outbound messages when an application is relaying messages.

The outbound channel adapter now supports expressions for the topic, qos and retained properties; the defaults remain the same.

See Chapter 24, MQTT Support for more information.

The STOMP module has been changed to use ReactorNettyTcpStompClient, based on the Project Reactor 3.1 and reactor-netty extension. The Reactor2TcpStompSessionManager has been renamed to the ReactorNettyTcpStompSessionManager according to the ReactorNettyTcpStompClient foundation.

See Chapter 29, STOMP Support for more information.

See Chapter 36, Web Services Support for more information.

The RedisStoreWritingMessageHandler is supplied now with additional String-based setters for SpEL expressions - for convenience with Java configuration. The zsetIncrementExpression can now be configured on the RedisStoreWritingMessageHandler, as well. In addition this property has been changed from true to false since INCR option on ZADD Redis command is optional.

The RedisInboundChannelAdapter can now be supplied with an Executor for executing Redis listener invokers. In addition the received messages now contains a RedisHeaders.MESSAGE_SOURCE header to indicate the source of the message - topic or pattern.

See Chapter 25, Redis Support for more information.

A new ThreadAffinityClientConnectionFactory is provided that binds TCP connections to threads.

You can now configure the TCP connection factories to support PushbackInputStream s, allowing deserializers to "unread" (push back) bytes after "reading ahead".

A ByteArrayElasticRawDeserializer has been added without maxMessageSize control and buffer incoming data as needed.

See Chapter 32, TCP and UDP Support for more information.

The GemfireMetadataStore now implements ListenableMetadataStore, allowing users to listen to cache events by providing MetadataStoreListener instances to the store.

See Chapter 17, GemFire Support for more information.

The JdbcMessageChannelStore now provides setter for the ChannelMessageStorePreparedStatementSetter allowing users to customize a message insertion in the store.

The ExpressionEvaluatingSqlParameterSourceFactory now provides setter for the sqlParameterTypes allowing users to customize sql types of the parameters.

See Chapter 19, JDBC Support for more information.

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.

Figure 3.1. Message

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.

Figure 3.2. Message Channel

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

Figure 3.3. Router

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 CorrelationStrategy, a ReleaseStrategy and configurable settings for: timeout, whether to send partial results upon timeout, and a 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.

Figure 3.4. Service Activator

	Note
	As discussed in Message Channel above, channels can be Pollable or Subscribable; in this diagram, this is depicted by the "clock" symbol and the solid arrow (poll) and the dotted arrow (subscribe).

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.

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

	Note
	Message sources can be Pollable (e.g. POP3) or Message-Driven (e.g. IMAP Idle); in this diagram, this is depicted by the "clock" symbol and the solid arrow (poll) and the dotted arrow (message-driven).

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

	Note
	As discussed in Message Channel above, channels can be Pollable or Subscribable; in this diagram, this is depicted by the "clock" symbol and the solid arrow (poll) and the dotted arrow (subscribe).

When using XML configuration, to avoid getting false schema validation errors, you should use a "Spring-aware" IDE, such as the Spring Tool Suite (STS) (or eclipse with the Spring IDE plugins) or IntelliJ IDEA, for example. These IDEs know how to resolve the correct XML schema from the classpath (using the META-INF/spring.schemas file in the jar(s)). When using STS, or eclipse with the plugin, be sure to enable Spring Project Nature on the project.

The schemas hosted on the internet for certain legacy modules (those that existed in version 1.0) are the 1.0 versions for compatibility reasons; if your IDE uses these schemas, you will likely see false errors.

Each of these online schemas has a warning similar to this:

Important

This schema is for the 1.0 version of Spring Integration Core. We cannot update it to the current schema because that will break any applications using 1.0.3 or lower. For subsequent versions, the unversioned schema is resolved from the classpath and obtained from the jar. Please refer to github: https://github.com/spring-projects/spring-integration/tree/master/spring-integration-core/src/main/resources/org/springframework/integration/config

The affected modules are

With XML configuration and Spring Integration Namespace support, the XML Parsers hide how target beans are declared and wired together. For Java & Annotation Configuration, it is important to understand the Framework API for target end-user applications.

The first class citizens for EIP implementation are Message, Channel and Endpoint (see Section 3.3, “Main Components” above). Their implementations (contracts) are:

The first two are simple enough to understand how to implement, configure and use, respectively; the last one deserves more review.

The AbstractEndpoint is widely used throughout the Framework for different component implementations; its main implementations are:

Using Messaging Annotations and/or Java DSL, you shouldn’t worry about these components, because the Framework produces them automatically via appropriate annotations and BeanPostProcessor s. When building components manually, the ConsumerEndpointFactoryBean should be used to help to determine the target AbstractEndpoint consumer implementation to create, based on the provided inputChannel property.

On the other hand, the ConsumerEndpointFactoryBean delegates to an another first class citizen in the Framework - org.springframework.messaging.MessageHandler. The goal of the implementation of this interface is to handle the message consumed by the endpoint from the channel. All EIP components in Spring Integration are MessageHandler implementations, e.g. AggregatingMessageHandler, MessageTransformingHandler, AbstractMessageSplitter etc.; as well as the target protocol outbound adapters are implementations too, e.g. FileWritingMessageHandler, HttpRequestExecutingMessageHandler, AbstractMqttMessageHandler etc. When you develop Spring Integration applications with Java & Annotation Configuration, you should take a look into the Spring Integration module to find an appropriate MessageHandler implementation to be used for the @ServiceActivator configuration. For example to send an XMPP message (see Chapter 38, XMPP Support) we should configure something like this:

@Bean
@ServiceActivator(inputChannel = "input")
public MessageHandler sendChatMessageHandler(XMPPConnection xmppConnection) {
    ChatMessageSendingMessageHandler handler = new ChatMessageSendingMessageHandler(xmppConnection);

    DefaultXmppHeaderMapper xmppHeaderMapper = new DefaultXmppHeaderMapper();
    xmppHeaderMapper.setRequestHeaderNames("*");
    handler.setHeaderMapper(xmppHeaderMapper);

    return handler;
}

The MessageHandler implementations represent the outbound and processing part of the message flow.

The inbound message flow side has its own components, which are divided to polling and listening behaviors. The listening (message-driven) components are simple and typically require only one target class implementation to be ready to produce messages. Listening components can be one-way MessageProducerSupport implementations, e.g. AbstractMqttMessageDrivenChannelAdapter and ImapIdleChannelAdapter; and request-reply - MessagingGatewaySupport implementations, e.g. AmqpInboundGateway and AbstractWebServiceInboundGateway.

Polling inbound endpoints are for those protocols which don’t provide a listener API or aren’t intended for such a behavior. For example any File based protocol, as an FTP, any data bases (RDBMS or NoSQL) etc.

These inbound endpoints consist with two components: the poller configuration, to initiate the polling task periodically, and message source class to read data from the target protocol and produce a message for the downstream integration flow. The first class, for the poller configuration, is a SourcePollingChannelAdapter. It is one more AbstractEndpoint implementation, but especially for polling to initiate an integration flow. Typically, with the Messaging Annotations or Java DSL, you shouldn’t worry about this class, the Framework produces a bean for it, based on the @InboundChannelAdapter configuration or a Java DSL Builder spec.

Message source components are more important for the target application development and they all implement the MessageSource interface, e.g. MongoDbMessageSource and AbstractTwitterMessageSource. With that in mind, our config for reading data from an RDBMS table with JDBC may look like:

@Bean
@InboundChannelAdapter(value = "fooChannel", poller = @Poller(fixedDelay="5000"))
public MessageSource<?> storedProc(DataSource dataSource) {
    return new JdbcPollingChannelAdapter(dataSource, "SELECT * FROM foo where status = 0");
}

All the required inbound and outbound classes for the target protocols you can find in the particular Spring Integration module, in most cases in the respective package. For example spring-integration-websocket adapters are:

If you are familiar with Spring Integration XML configuration, starting with version 4.3, we provide information in the XSD element definitions about which target classes are used to declare beans for the adapter or gateway, for example:

<xsd:element name="outbound-async-gateway">
    <xsd:annotation>
		<xsd:documentation>
Configures a Consumer Endpoint for the 'o.s.i.amqp.outbound.AsyncAmqpOutboundGateway'
that will publish an AMQP Message to the provided Exchange and expect a reply Message.
The sending thread returns immediately; the reply is sent asynchronously; uses 'AsyncRabbitTemplate.sendAndReceive()'.
       </xsd:documentation>
	</xsd:annotation>

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 21, 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 convenience 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 Message s 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);

    void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex);

    boolean preReceive(MessageChannel channel);

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

    void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex);
}

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(..), postReceive(..) and afterReceiveCompletion(..) 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.

Starting with Spring Framework 4.1 and Spring Integration 4.1, the ChannelInterceptor provides new methods - afterSendCompletion() and afterReceiveCompletion(). They are invoked after send()/receive() calls, regardless of any exception that is raised, thus allowing for resource cleanup. Note, the Channel invokes these methods on the ChannelInterceptor List in the reverse order of the initial preSend()/preReceive() calls.

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 8.4.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 4.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 8.1.6, “Payload Type Conversion”.

Beginning with version 4.0, the integrationConversionService is invoked by the DefaultDatatypeChannelMessageConverter, which looks up the conversion service in the application context. To use a different conversion technique, you can specify the message-converter attribute on the channel. This must be a reference to a MessageConverter implementation. Only the fromMessage method is used, which provides the converter with access to the message headers (for example if the conversion might need information from the headers, such as content-type). The method can return just the converted payload, or a full Message object. If the latter, the converter must be careful to copy all the headers from the inbound message.

Alternatively, declare a <bean/> of type MessageConverter with an id "datatypeChannelMessageConverter" and that converter will be used by all channels with a datatype.

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 10.4, “Message Store”.

	Important
	The capacity attribute is not allowed when the message-store attribute is used.

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, a QueueChannel stores its Messages in an in-memory Queue and can therefore lead to the lost message scenario mentioned above. However Spring Integration provides persistent stores, such as the JdbcChannelMessageStore.

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="channelStore"/>
</int:channel>

<bean id="channelStore" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
    <property name="dataSource" ref="dataSource"/>
    <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
</bean>

The Spring Integration JDBC module also provides schema DDL for a number of popular databases. These schemas are located in the org.springframework.integration.jdbc.store.channel package of that module (spring-integration-jdbc).

	Important
	One important feature is that with any transactional persistent store (e.g., JdbcChannelMessageStore), 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.

Since version 4.0, it is recommended that QueueChannel s are configured to use a ChannelMessageStore if possible. These are generally optimized for this use, when compared with a general message store. If the ChannelMessageStore is a ChannelPriorityMessageStore the messages will be received in FIFO within priority order. The notion of priority is determined by the message store implementation. For example the Java Configuration for the Section 23.3.1, “MongoDB Channel Message Store”:

@Bean
public BasicMessageGroupStore mongoDbChannelMessageStore(MongoDbFactory mongoDbFactory) {
    MongoDbChannelMessageStore store = new MongoDbChannelMessageStore(mongoDbFactory);
    store.setPriorityEnabled(true);
    return store;
}

@Bean
public PollableChannel priorityQueue(BasicMessageGroupStore mongoDbChannelMessageStore) {
    return new PriorityChannel(new MessageGroupQueue(mongoDbChannelMessageStore, "priorityQueue"));
}

	Note
	Pay attention to the MessageGroupQueue class. That is a BlockingQueue implementation to utilize the MessageGroupStore operations.

The same with Java DSL may look like:

@Bean
public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) {
    return IntegrationFlows.from((Channels c) ->
            c.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup"))
            ....
            .get();
}

Another option to customize the QueueChannel environment is provided by the ref attribute of the <int:queue> sub-element or particular constructor. This attribute implies the reference to any java.util.Queue implementation. For example Hazelcast distributed IQueue:

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance(new Config()
                                           .setProperty("hazelcast.logging.type", "log4j"));
}

@Bean
public PollableChannel distributedQueue() {
    return new QueueChannel(hazelcastInstance()
                              .getQueue("springIntegrationQueue"));
}

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, !baz*" order="3">
    <bean class="foo.barSampleInterceptor"/>
</int:channel-interceptor>

or

<int:channel-interceptor ref="myInterceptor" pattern="input*, bar*, foo, !baz*" 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 and not to channel starting with baz (starting with version 5.0).

	Warning
	The addition of this syntax to the pattern causes one possible (although perhaps unlikely) problem. If you have a bean "!foo"and you included a pattern "!foo" in your channel-interceptor’s pattern patterns; it will no long match; the pattern will now match all beans not named foo. In this case, you can escape the ! in the pattern with \. The pattern "\!foo" means match a bean named "!foo".

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

Important

Starting with version 4.0, it is important to avoid circular references when an interceptor (such as WireTap) references a channel itself. You need to exclude such channels from those being intercepted by the current interceptor. This can be done with appropriate patterns or programmatically. If you have a custom ChannelInterceptor that references a channel, consider implementing VetoCapableInterceptor. That way, the framework will ask the interceptor if it’s OK to intercept each channel that is a candidate based on the pattern. You can also add runtime protection in the interceptor methods that ensures that the channel is not one that is referenced by the interceptor. The WireTap uses both of these techniques.

Starting with version 4.3, the WireTap has additional constructors that take a channelName instead of a MessageChannel instance. This can be convenient for Java Configuration and when channel auto-creation logic is being used. The target MessageChannel bean is resolved from the provided channelName later, on the first interaction with the interceptor.

	Important
	Channel resolution requires a BeanFactory so the wire tap instance must be a Spring-managed bean.

This late-binding approach also allows simplification of typical wire-tapping patterns with Java DSL configuration:

@Bean
public PollableChannel myChannel() {
    return MessageChannels.queue()
            .wireTap("loggingFlow.input")
            .get();
}

@Bean
public IntegrationFlow loggingFlow() {
    return f -> f.log();
}

Global Wire Tap Configuration

It is possible to configure a global wire tap as a special case of the the section called “Global Channel Interceptor Configuration”. 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 E.4, “Error Handling”.

See also Section 9.4, “Message Channels” in Java DSL chapter for more information about message channel and interceptors.

When Message Endpoints (Channel Adapters) are connected to channels and instantiated, they produce one of the following 2 instances:

The actual implementation depends on which type of channel these Endpoints are connected to. A channel adapter connected to a channel that implements the org.springframework.messaging.SubscribableChannel interface will produce an instance of EventDrivenConsumer. On the other hand, a channel adapter connected to a channel that implements the org.springframework.messaging.PollableChannel interface (e.g. a QueueChannel) will produce an instance of PollingConsumer.

Polling Consumers allow Spring Integration components to actively poll for Messages, rather than to process Messages in an event-driven manner.

They represent a critical cross cutting concern in many messaging scenarios. In Spring Integration, Polling Consumers are based on the pattern with the same name, which is described in the book "Enterprise Integration Patterns" by Gregor Hohpe and Bobby Woolf. You can find a description of the pattern on the book’s website at:

http://www.enterpriseintegrationpatterns.com/PollingConsumer.html

Furthermore, in Spring Integration a second variation of the Polling Consumer pattern exists. When Inbound Channel Adapters are being used, these adapters are often wrapped by a SourcePollingChannelAdapter. For example, when retrieving messages from a remote FTP Server location, the adapter described in Section 16.4, “FTP Inbound Channel Adapter” is configured with a poller to retrieve messages periodically. So, when components are configured with Pollers, the resulting instances are of one of the following types:

This means, Pollers are used in both inbound and outbound messaging scenarios. Here are some use-cases that illustrate the scenarios in which Pollers are used:

Note

AOP Advice classes can be applied to pollers, in an advice-chain. An example being a transaction advice to start a transaction. Starting with version 4.1 a PollSkipAdvice is provided. Pollers use triggers to determine the time of the next poll. The PollSkipAdvice can be used to suppress (skip) a poll, perhaps because there is some downstream condition that would prevent the message to be processed properly. To use this advice, you have to provide it with an implementation of a PollSkipStrategy. Starting with version 4.2.5, a SimplePollSkipStrategy is provided. Add an instance as a bean to the application context, inject it into a PollSkipAdvice and add that to the poller’s advice chain. To skip polling, call skipPolls(), to resume polling, call reset(). Version 4.2 added more flexibility in this area - see Section 4.2.3, “Conditional Pollers for Message Sources”.

This chapter is meant to only give a high-level overview regarding Polling Consumers and how they fit into the concept of message channels - Section 4.1, “Message Channels” and channel adapters - Section 4.3, “Channel Adapter”. For more in-depth information regarding Messaging Endpoints in general and Polling Consumers in particular, please see Section 8.1, “Message Endpoints”.

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 4.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 8.1.4, “Endpoint 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 a 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 8.8, “Groovy support” and Section 8.7, “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.

@Bean
public PollableChannel polled() {
    return new QueueChannel();
}

@Bean
@BridgeFrom(value = "polled", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "10"))
public SubscribableChannel direct() {
    return new DirectChannel();
}

or

@Bean
@BridgeTo(value = "direct", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "10"))
public PollableChannel polled() {
    return new QueueChannel();
}

@Bean
public SubscribableChannel direct() {
    return new DirectChannel();
}

Or, using a BridgeHandler:

@Bean
@ServiceActivator(inputChannel = "polled",
        poller = @Poller(fixedRate = "5000", maxMessagesPerPoll = "10"))
public BridgeHandler bridge() {
    BridgeHandler bridge = new BridgeHandler();
    bridge.setOutputChannelName("direct");
    return bridge;
}

@Bean
public IntegrationFlow bridgeFlow() {
    return IntegrationFlows.from("polled")
            .bridge(e -> e.poller(Pollers.fixedDelay(5000).maxMessagesPerPoll(10)))
            .channel("direct")
            .get();
}

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:

MessageHeaders.ID

java.util.UUID

MessageHeaders.
    TIMESTAMP

java.lang.Long

MessageHeaders.
    REPLY_CHANNEL

java.lang.Object
(String or MessageChannel)

MessageHeaders.
    ERROR_CHANNEL

java.lang.Object
(String or MessageChannel)

Many inbound and outbound adapter implementations will also provide and/or expect certain headers, and additional user-defined headers can also be configured. Constants for these headers can be found in those modules where such headers exist, for example AmqpHeaders, JmsHeaders etc.

IntegrationMessageHeaderAccessor.
    CORRELATION_ID

java.lang.Object

IntegrationMessageHeaderAccessor.
    SEQUENCE_NUMBER

java.lang.Integer

IntegrationMessageHeaderAccessor.
    SEQUENCE_SIZE

java.lang.Integer

IntegrationMessageHeaderAccessor.
    EXPIRATION_DATE

java.lang.Long

IntegrationMessageHeaderAccessor.
    PRIORITY

java.lang.Integer

IntegrationMessageHeaderAccessor.
    DUPLICATE_MESSAGE

java.lang.Boolean

IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor(message);
int sequenceNumber = accessor.getSequenceNumber();
Object correlationId = accessor.getCorrelationId();
...

IntegrationMessageHeaderAccessor.
    SEQUENCE_DETAILS

java.util.List<
List<Object>>

IntegrationMessageHeaderAccessor.
    ROUTING_SLIP

java.util.Map<
List<Object>, Integer>

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 org.springframework.util.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 org.springframework.util.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; org.springframework.util.JdkIdGenerator uses the previous UUID.randomUUID() mechanism; o.s.i.support.IdGenerators.SimpleIncrementingIdGenerator can be used in cases where a UUID is not really needed and a simple incrementing value is sufficient.

Header Propagation

When messages are processed (and modified) by message-producing endpoints (such as a service activator), in general, inbound headers are propagated to the outbound message. One exception to this is a transformer, when a complete message is returned to the framework; in that case, the user code is responsible for the entire outbound message. When a transformer just returns the payload; the inbound headers are propagated. Also, a header is only propagated if it does not already exist in the outbound message, allowing user code to change header values as needed.

Starting with version 4.3.10, you can configure message handlers (that modify messages and produce output) to suppress the propagation of specific headers. Call the setNotPropagatedHeaders() or addNotPropagatedHeaders() methods on the MessageProducingMessageHandler abstract class, to configure the header(s) you don’t want to be copied.

You can also globally suppress propagation of specific message headers by setting the readOnlyHeaders property in META-INF/spring.integration.properties to a comma-delimited list of headers.

Starting with version 5.0, the setNotPropagatedHeaders() implementation on the AbstractMessageProducingHandler applies the simple patterns (xxx*, *xxx, *xxx* or xxx*yyy) to allow filtering headers with a common suffix or prefix. See PatternMatchUtils JavaDocs for more information. When one of the patterns is * (asterisk), no headers are propagated; all other patterns are ignored. In this case the Service Activator behaves the same way as Transformer and any required headers must be supplied in the Message returned from the service method. The option notPropagatedHeaders() is available in the ConsumerEndpointSpec for Java DSL, as well as for XML configuration of the <service-activator> component as a not-propagated-headers attribute.

	Important
	Header propagation suppression does not apply to those endpoints that don’t modify the message, e.g. bridges and routers.

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(IntegrationMessageHeaderAccessor.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 6.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 6.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