The spring-integration-rsocket module is now available with channel adapter implementations for RSocket protocol support. See RSocket Support for more information.

The RateLimiterRequestHandlerAdvice is now available for limiting requests rate on handlers. See Rate Limiter Advice for more information.

The CacheRequestHandlerAdvice is now available for caching request results on handlers. See Caching Advice for more information.

The JSR223 scripting module now includes a support for Kotlin scripts. See Scripting Support for more information.

The FluxAggregatorMessageHandler is now available for grouping and windowing messages logic based on the Project Reactor Flux operators. See Flux Aggregator for more information.

The FTP and SFTP modules now provide an event listener for certain Apache Mina FTP/SFTP server events. See Apache Mina FTP Server Events and Apache Mina SFTP Server Events for more information.

Simple Apache Avro transformers are now provided. See Avro Transformers for more information.

The outbound endpoints can now be configured to synthesize a "nack" if no publisher confirm is received within a timeout. See Outbound Channel Adapter for more information.

The inbound channel adapter can now receive batched messages as a List<?> payload instead of receiving a discrete message for each batch fragment. See Batched Messages for more information.

The outbound channel adapter can now be configured to block the calling thread until a publisher confirm (acknowledgment) is received. See Outbound Channel Adapter for more information.

Some improvements to filtering remote files have been made. See Remote Persistent File List Filters for more information.

The length header used by the ByteArrayLengthHeaderSerializer can now include the length of the header in addition to the payload. See Message Demarcation (Serializers and Deserializers) for more information.

When using a TcpNioServerConnectionFactory, priority is now given to accepting new connections over reading from existing connections, but it is configurable. See About Non-blocking I/O (NIO) for more information.

The outbound gateway has a new property closeStreamAfterSend; when used with a new connection for each request/reply it signals EOF to the server, without closing the connection. This is useful for servers that use the EOF to signal end of message instead of some delimiter in the data. See TCP Gateways for more information.

The client connection factories now support connectTimeout which causes an exception to be thrown if the connection is not established in that time. See TCP Connection Factories for more information.

SoftEndOfStreamException is now a RuntimeException instead of extending IOException.

The AbstractMailReceiver has now an autoCloseFolder option (true by default), to disable an automatic folder close after a fetch, but populate IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE header instead for downstream interaction. See Mail-receiving Channel Adapter for more information.

The HTTP inbound endpoint now support a request payload validation. See HTTP Support for more information.

The WebFluxRequestExecutingMessageHandler now supports a Publisher, Resource and MultiValueMap as a request message payload. The WebFluxInboundEndpoint now supports a request payload validation. See WebFlux Support for more information.

The MongoDbMessageStore can now be configured with custom converters. See MongoDB Support for more information.

You can now disable falling back to the channel key as the channel bean name. See Dynamic Routers for more information.

The RotatingServerAdvice is decoupled now from the RotationPolicy and its StandardRotationPolicy.

The remote file information, including host/port and directory are included now into message headers in the AbstractInboundFileSynchronizingMessageSource and AbstractRemoteFileStreamingMessageSource implementations. Also this information is included into headers in the read operations results of the AbstractRemoteFileOutboundGateway implementations. The FTP outbound endpoints now support chmod to change permissions on the uploaded file. (SFTP already supported it since version 4.3). See FTP(S) Support and SFTP 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, and others) may be configured as message header values by an upstream component. Developers can also store any arbitrary key-value pairs in the headers.

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

A message channel may follow either point-to-point or publish-subscribe semantics. With a point-to-point channel, no more than one consumer can receive each message sent to the channel. Publish-subscribe channels, on the other hand, attempt to broadcast each message to all subscribers on the channel. Spring Integration supports both of these models.

Whereas “point-to-point” and "publish-subscribe" define the two options for how many consumers 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. Message Channel Implementations has a detailed discussion of the variety of channel implementations available in Spring Integration.

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 should necessarily connect your existing application code directly. Any real-world enterprise integration solution requires some amount of code focused upon integration concerns such as routing and transformation. The important thing is to achieve separation of concerns between the integration logic and the business logic. In other words, as with the Model-View-Controller (MVC) 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 provides an overview of the message endpoint types that handle these responsibilities, and, in upcoming chapters, you can 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 (such as from XML to java.lang.String). Similarly, a transformer can 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, or other conditions. If the message is accepted, it is sent to the output channel. If not, it is 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 criteria of the filter to narrow down the set of messages to be processed.

A message router is responsible for deciding what channel or channels (if any) should receive the message next. Typically, the decision is based upon the message’s content or the 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 earlier.

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 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 subdivided 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, discard them, or send 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.

The service activator invokes an operation on some service object to process the request message, extracting the request message’s payload and converting (if the method does not expect a message-typed parameter). Whenever the service object’s method returns a value, that return value is likewise converted to a reply message if necessary (if it is not already a message type). That reply message is sent to the output channel. If no output channel has been configured, the reply is sent to the channel specified in the message’s “return address”, if available.

A request-reply service activator endpoint connects a target object’s method to input and output Message Channels.

A channel adapter is an endpoint that connects a message channel to some other system or transport. Channel adapters may be either inbound or outbound. Typically, the channel adapter does some mapping between the message and whatever object or resource is received from or sent to the other system (file, HTTP Request, JMS message, and others). Depending on the transport, the channel adapter may also populate or extract message header values. Spring Integration provides a number of channel adapters, which are described in upcoming chapters.

Consuming endpoints (anything with an inputChannel) consist of two beans, the consumer and the message handler. The consumer has a reference to the message handler and invokes it as messages arrive.

Consider the following XML example:

Given the preceding example, the bean names are as follows:

When using Enterprise Integration Pattern (EIP) annotations, the names depend on several factors. Consider the following example of an annotated POJO:

Given the preceding example, the bean names are as follows:

Starting with version 5.0.4, you can modify these names by using the @EndpointId annotation, as the following example shows:

Given the preceding example, the bean names are as follows:

The @EndpointId creates names as created by the id attribute with XML configuration. Consider the following example of an annotated bean:

Given the preceding example, the bean names are as follows:

Starting with version 5.0.4, you can modify these names by using the @EndpointId annotation, as the following example shows:

The @EndpointId annotation creates names as created by the id attribute with XML configuration, as long as you use the convention of appending .handler to the @Bean name.

There is one special case where a third bean is created: For architectural reasons, if a MessageHandler @Bean does not define an AbstractReplyProducingMessageHandler, the framework wraps the provided bean in a ReplyProducingMessageHandlerWrapper. This wrapper supports request handler advice handling and emits the normal 'produced no reply' debug log messages. Its bean name is the handler bean name plus .wrapper (when there is an @EndpointId — otherwise, it is the normal generated handler name).

Similarly Pollable Message Sources create two beans, a SourcePollingChannelAdapter (SPCA) and a MessageSource.

Consider the following XML configuration:

Given the preceding XML configuration, the bean names are as follows:

Consider the following Java configuration of a POJO to define an @EndpointId:

Given the preceding Java configuration example, the bean names are as follows:

Consider the following Java configuration of a bean to define an @EndpointID:

Given the preceding example, the bean names are as follows:

Spring Integration bootstraps certain features by using Spring Framework’s SpringFactories mechanism to load several IntegrationConfigurationInitializer classes. This includes the -core jar as well as certain others, including -http and -jmx. The information for this process is stored in a META-INF/spring.factories file in each jar.

Some developers prefer to repackage their application and all dependencies into a single jar by using well known tools, such as the Apache Maven Shade Plugin.

By default, the shade plugin does not merge the spring.factories files when producing the shaded jar.

In addition to spring.factories, other META-INF files (spring.handlers and spring.schemas) are used for XML configuration. These files also need to be merged.

Even if you do not use Spring Boot, you can still use the tooling provided by Boot to enhance the shade plugin by adding transformers for the above mentioned files.

You may wish to consult the current spring-boot-starter-parent pom to see the current settings that boot uses. The following example shows how to configure the plugin:

You can add a property for ${spring.boot.version} or use an explicit version.

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), Eclipse with the Spring IDE plugins, or IntelliJ IDEA. These IDEs know how to resolve the correct XML schema from the classpath (by using the META-INF/spring.schemas file in the jars). When using STS or Eclipse with the plugin, you must 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 are likely to see false errors.

Each of these online schemas has a warning similar to the following:

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 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 Main Components, earlier in this chapter). Their implementations (contracts) are:

The first two are simple enough to understand how to implement, configure, and use. The last one deserves more attention

The AbstractEndpoint is widely used throughout the Spring Framework for different component implementations. Its main implementations are:

When you use messaging annotations or the Java DSL, you need to worry about these components, because the Framework automatically produces them with appropriate annotations and BeanPostProcessor implementations. When building components manually, you should use the ConsumerEndpointFactoryBean to help 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 (for example, AggregatingMessageHandler, MessageTransformingHandler, AbstractMessageSplitter, and others). The target protocol outbound adapters (FileWritingMessageHandler, HttpRequestExecutingMessageHandler, AbstractMqttMessageHandler, and others) are also MessageHandler implementations. When you develop Spring Integration applications with Java configuration, you should look into the Spring Integration module to find an appropriate MessageHandler implementation to use for the @ServiceActivator configuration. For example, to send an XMPP message (see XMPP Support) you should configure something like the following:

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 into 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, (such as AbstractMqttMessageDrivenChannelAdapter and ImapIdleChannelAdapter) or request-reply MessagingGatewaySupport implementations (such as AmqpInboundGateway and AbstractWebServiceInboundGateway).

Polling inbound endpoints are for those protocols that do not provide a listener API or are not intended for such a behavior, including any file based protocol (such as FTP), any data bases (RDBMS or NoSQL), and others.

These inbound endpoints consist of two components: the poller configuration, to initiate the polling task periodically, and a 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 should not 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 (for example, MongoDbMessageSource and AbstractTwitterMessageSource). With that in mind, our config for reading data from an RDBMS table with JDBC could resemble the following:

You can find all the required inbound and outbound classes for the target protocols in the particular Spring Integration module (in most cases, in the respective package). For example, the 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, as the following example shows:

Spring Integration’s top-level MessageChannel interface is defined as follows:

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

Spring Integration provides several different message channel implementations. The following sections briefly describe each one.

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 instances are sent to and received from MessageChannel instances, those channels provide an opportunity for intercepting the send and receive operations. The ChannelInterceptor strategy interface, shown in the following listing, provides methods for each of those operations:

After implementing the interface, registering the interceptor with a channel is just a matter of making the following call:

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

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

Because it is rarely necessary to implement all of the interceptor methods, the interface provides no-op methods (methods returning void method have no code, the Message-returning methods return the Message as-is, and the boolean method returns true).

Starting with Spring Framework 4.1 and Spring Integration 4.1, the ChannelInterceptor provides new methods: afterSendCompletion() and afterReceiveCompletion(). They are invoked after send()' and 'receive() calls, regardless of any exception that is raised, which allow for resource cleanup. Note that the channel invokes these methods on the ChannelInterceptor list in the reverse order of the initial preSend() and preReceive() calls.

Starting with version 5.1, global channel interceptors now apply to dynamically registered channels - such as through beans that are initialized by using beanFactory.initializeBean() or IntegrationFlowContext when using the Java DSL. Previously, interceptors were not applied when beans were created after the application context was refreshed.

Also, starting with version 5.1, ChannelInterceptor.postReceive() is no longer called when no message is received; it is no longer necessary to check for a null Message<?>. Previously, the method was called. If you have an interceptor that relies on the previous behavior, implement afterReceiveCompleted() instead, since that method is invoked, regardless of whether a message is received or not.

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, it is sometimes 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 and reply scenarios. For example, it is possible to send a request and wait for a reply, as follows:

In the preceding 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. The following listing shows the signatures for such methods:

To create a message channel instance, you can use the <channel/> element, as follows:

The equivalent Java configuration declares a DirectChannel @Bean:

The default channel type is point-to-point. To create a publish-subscribe channel, use the <publish-subscribe-channel/> element, as follows:

The Java configuration is:

When you use the <channel/> element without any sub-elements, it creates 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 Message Channel Implementations). The following sections shows examples of each channel type.

If namespace support is enabled, two special channels are defined within the application context by default: errorChannel and nullChannel. The 'nullChannel' acts like /dev/null, logging any message sent to it at the DEBUG level and returning immediately. Any time you face channel resolution errors for a reply that you do not 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 Error Handling.

See also Message Channels in the 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 instances:

The actual implementation depends on the type of channel to which these endpoints connect. A channel adapter connected to a channel that implements the org.springframework.messaging.SubscribableChannel interface produces an instance of EventDrivenConsumer. On the other hand, a channel adapter connected to a channel that implements the org.springframework.messaging.PollableChannel interface (such as a QueueChannel) produces an instance of PollingConsumer.

Polling consumers let Spring Integration components actively poll for Messages rather than 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.

Spring Integration offers a second variation of the polling consumer pattern. When inbound channel adapters are used, these adapters are often wrapped by a SourcePollingChannelAdapter. For example, when retrieving messages from a remote FTP Server location, the adapter described in FTP Inbound Channel Adapter is configured with a poller to periodically retrieve messages. So, when components are configured with pollers, the resulting instances are of one of the following types:

This means that pollers are used in both inbound and outbound messaging scenarios. Here are some use cases in which pollers are used:

This chapter is meant to only give a high-level overview of polling consumers and how they fit into the concept of message channels (see Message Channels) and channel adapters (see Channel Adapter). For more information regarding messaging endpoints in general and polling consumers in particular, see Message Endpoints.

Starting with version 5.0.1, certain modules provide MessageSource implementations that support deferring acknowledgment until the downstream flow completes (or hands off the message to another thread). This is currently limited to the AmqpMessageSource and the KafkaMessageSource provided by the spring-integration-kafka extension project.

With these message sources, the IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK header (see MessageHeaderAccessor API) is added to the message. The value of the header is an instance of AcknowledgmentCallback, as the following example shows:

Not all message sources (for example, Kafka) support the REJECT status. It is treated the same as ACCEPT.

Applications can acknowledge a message at any time, as the following example shows:

If the MessageSource is wired into a SourcePollingChannelAdapter, when the poller thread returns to the adapter after the downstream flow completes, the adapter checks whether the acknowledgment has already been acknowledged and, if not, sets its status to ACCEPT it (or REJECT if the flow throws an exception). The status values are defined in the AcknowledgmentCallback.Status enumeration.

Spring Integration provides MessageSourcePollingTemplate to perform ad-hoc polling of a MessageSource. This, too, takes care of setting ACCEPT or REJECT on the AcknowledgmentCallback when the MessageHandler callback returns (or throws an exception). The following example shows how to poll with the MessageSourcePollingTemplate:

In both cases (SourcePollingChannelAdapter and MessageSourcePollingTemplate), you can disable auto ack/nack by calling noAutoAck() on the callback. You might do this if you hand off the message to another thread and wish to acknowledge later. Not all implementations support this (for example, Apache Kafka does not, because the offset commit has to be performed on the same thread).

This section covers how to use conditional pollers.

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 the method’s output to a Message. When the adapter’s subscription is activated, a poller tries to receive messages from the source. The poller is scheduled with the TaskScheduler according to the provided configuration. To configure the polling interval or cron expression for an individual channel adapter, you can provide a 'poller' element with one of the scheduling attributes, such as 'fixed-rate' or 'cron'. The following example defines two inbound-channel-adapter instances:

See also Channel Adapter Expressions and Scripts.

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. The following example shows how to define an outbound channel adapter:

If the channel being adapted is a PollableChannel, you must provide a poller sub-element, as the following example shows:

You should use a ref attribute if the POJO consumer implementation can be reused in other <outbound-channel-adapter> definitions. However, if the consumer implementation is referenced by only a single definition of the <outbound-channel-adapter>, you can define it as an inner bean, as the following example shows:

Any channel adapter can be created without a channel reference, in which case it implicitly creates an instance of DirectChannel. The created channel’s name matches the id attribute of the <inbound-channel-adapter> or <outbound-channel-adapter> element. Therefore, if channel is not provided, 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 in 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> generates a message any time 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 a <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 by using the location attribute, you can also set refresh-check-delay, which allows the resource to be periodically refreshed. If you want the script to be checked on each poll, you would need to coordinate this setting with the poller’s trigger, as the following example shows:

See also the cacheSeconds property on the ReloadableResourceBundleExpressionSource when using the <expression/> sub-element. For more information regarding expressions, see Spring Expression Language (SpEL). For scripts, see Groovy support and Scripting Support.

You can use the <bridge> element is used to create a messaging bridge between two message channels or channel adapters. To do so, provide the input-channel and output-channel attributes, as the following example shows:

As mentioned above, a common use case for the messaging bridge is to connect a PollableChannel to a SubscribableChannel. When performing this role, the messaging bridge may also serve as a throttler:

You can use a similar mechanism to connecting channel adapters. The following example shows a simple “echo” between the stdin and stdout adapters from Spring Integration’s stream namespace:

Similar configurations work for other (potentially more useful) Channel Adapter bridges, such as file-to-JMS or mail-to-file. Upcoming chapters cover the various channel adapters.

The following example shows how to configure a bridge in Java by using the @BridgeFrom annotation:

The following example shows how to configure a bridge in Java by using the @BridgeTo annotation:

Alternately, you can use a BridgeHandler, as the following example shows:

You can use the Java Domain Specific Language (DSL) to configure a bridge, as the following example shows:

Starting with Spring Framework 4.0 and Spring Integration 4.0, the core messaging abstraction has been moved to the spring-messaging module, and the MessageHeaderAccessor API has been introduced to provide additional abstraction over messaging implementations. All (core) Spring Integration-specific message headers constants are now declared in the IntegrationMessageHeaderAccessor class. The following table describes the pre-defined message headers:

Convenient typed getters for some of these headers are provided on the IntegrationMessageHeaderAccessor class, as the following example shows:

The following table describes headers that also appear in the IntegrationMessageHeaderAccessor but are generally not used by user code (that is, they are generally used by internal parts of Spring Integration — their inclusion here is for completeness):

When a message transitions through an application, each time it is mutated (for example, 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 IS 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.

In addition to the default strategy, two additional IdGenerators are provided. org.springframework.util.JdkIdGenerator uses the previous UUID.randomUUID() mechanism. You can use o.s.i.support.IdGenerators.SimpleIncrementingIdGenerator when a UUID is not really needed and a simple incrementing value is sufficient.

The MessageHeaders.ID and MessageHeaders.TIMESTAMP are read-only headers and cannot be overridden.

Since version 4.3.2, the MessageBuilder provides the readOnlyHeaders(String…​ readOnlyHeaders) API to customize a list of headers that should not be copied from an upstream Message. Only the MessageHeaders.ID and MessageHeaders.TIMESTAMP are read only by default. The global spring.integration.readOnly.headers property (see Global Properties) is provided to customize DefaultMessageBuilderFactory for framework components. This can be useful when you would like do not populate some out-of-the-box headers, such as contentType by the ObjectToJsonTransformer (see JSON Transformers).

When you try to build a new message using MessageBuilder, this kind of header is ignored and a particular INFO message is emitted to logs.

Starting with version 5.0, Messaging Gateway, Header Enricher, Content Enricher and Header Filter do not let you configure the MessageHeaders.ID and MessageHeaders.TIMESTAMP header names when DefaultMessageBuilderFactory is used, and they throw BeanInitializationException.

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, letting you 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. To configure the header(s) you do not want to be copied, call the setNotPropagatedHeaders() or addNotPropagatedHeaders() methods on the MessageProducingMessageHandler abstract class.

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 simple patterns (xxx*, xxx, *xxx, or xxx*yyy) to allow filtering headers with a common suffix or prefix. See PatternMatchUtils Javadoc for more information. When one of the patterns is * (asterisk), no headers are propagated. All other patterns are ignored. In that case, the service activator behaves the same way as a transformer and any required headers must be supplied in the Message returned from the service method. The notPropagatedHeaders() option is available in the ConsumerEndpointSpec for the Java DSL It is also available for XML configuration of the <service-activator> component as a not-propagated-headers attribute.

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 channels depending on a set of conditions.

Spring Integration provides the following routers:

Router implementations share many configuration parameters. However, 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 two following tables .

The following table shows the configuration parameters available for a router outside of a chain:

The following table shows the configuration parameters available for a router inside of a chain:

This section describes the parameters common to all router parameters (the parameters with all their boxes ticked in the two tables shown earlier in this chapter).

Since content-based routing often requires some domain-specific logic, most use cases require Spring Integration’s options for delegating to POJOs by using either the XML namespace support or annotations. Both of these are discussed later. However, we first present a couple of implementations that fulfill common requirements.

Spring Integration provides a generic router. You can use it for general-purpose routing (as opposed to the other routers provided by Spring Integration, each of which has some form of specialization).

Sometimes, the routing logic may be simple, and writing a separate class for it and configuring it as a bean may seem like overkill. As of Spring Integration 2.0, we offer an alternative that lets you use SpEL to implement simple computations that previously required a custom POJO router.

Generally, a SpEL expression is evaluated and its result is mapped to a channel, as the following example shows:

The following example shows the equivalent router configured in Java:

The following example shows the equivalent router configured in the Java DSL:

To simplify things even more, the SpEL expression may evaluate to a channel name, as the following expression shows:

In the preceding configuration, the result channel is computed by the SpEL expression, which concatenates the value of the payload with the literal String, 'Channel'.

Another virtue of SpEL for configuring routers is that an expression can return a Collection, effectively making every <router> a recipient list router. Whenever the expression returns multiple channel values, the message is forwarded to each channel. The following example shows such an expression:

In the above configuration, if the message includes a header with a name of 'channels' and the value of that header is a List of channel names, the message is sent to each channel in the list. You may also find collection projection and collection selection expressions useful when you need to select multiple channels. For further information, see:

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

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

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

Unfortunately, with static router configuration to accomplish either of those goals, you would have to bring down your entire application, change the configuration of the router (change routes), and bring the application back up. This is obviously not a solution anyone wants.

The dynamic router pattern describes the mechanisms by which you can change or configure routers dynamically without bringing down the system or individual routers. 

Before we get into the specifics of how Spring Integration supports dynamic routing, we need to consider the typical flow of a router:

There is not much that can be done with regard to dynamic routing if Step 1 results in the actual instance of the MessageChannel, because the MessageChannel is the final product of any router’s job. However, if the first step results in a channel identifier that is not an instance of MessageChannel, you have quite a few possible ways to influence the process of deriving the MessageChannel. Consider the following example of a payload type router:

Within the context of a payload type router, the three steps mentioned earlier would be realized as follows:

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

Now consider an example of a header value router:

Now we can consider how the three steps work for a header value router:

The preceding two configurations of two different router types look almost identical. However, if you look at the alternate configuration of the HeaderValueRouter we clearly see that there is no mapping sub element, as the following listing shows:

However, the configuration is still perfectly valid. So the natural question is what about the mapping in the second step?

The second step is now optional. If mapping is not defined, then the channel identifier value computed in the first step is automatically treated as the channel name, which is now resolved to the actual MessageChannel, as in the third step.  What it also means is that the second step is one of the key steps to providing dynamic characteristics to the routers, since it introduces a process that lets you change the way channel identifier resolves to the channel name, thus influencing the process of determining the final instance of the MessageChannel from the initial channel identifier. 

For example, in the preceding configuration, assume that the testHeader value is 'kermit', which is now a channel identifier (the first step). Since there is no mapping in this router, resolving this channel identifier to a channel name (the second step) is impossible and this channel identifier is now treated as the channel name. However, what if there was a mapping but for a different value? The end result would still be the same, because, if a new value cannot be determined through the process of resolving the channel identifier to a channel name, the channel identifier becomes the channel name.

All that is left is for the third step to resolve the channel name ('kermit') to an actual instance of the MessageChannel identified by this name. That basically involves a bean lookup for the provided name. Now all messages that contain the header-value pair as testHeader=kermit are going to be routed to a MessageChannel whose bean name (its id) is 'kermit'.

But what if you want to route these messages to the 'simpson' channel? Obviously changing a static configuration works, but doing so also requires bringing your system down. However, if you had access to the channel identifier map, you could introduce a new mapping where the header-value pair is now kermit=simpson, thus letting the second step treat 'kermit' as a channel identifier while resolving it to 'simpson' as the channel name.

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

Any router that is a subclass of the AbstractMappingMessageRouter (which includes most framework-defined routers) is a dynamic router, because the channelMapping is defined at the AbstractMappingMessageRouter level. That map’s setter method is exposed as a public method along with the 'setChannelMapping' and 'removeChannelMapping' methods. These let you change, add, and remove router mappings at runtime, as long as you have a reference to the router itself. It also means that you could expose these same configuration options through JMX (see JMX Support) or the Spring Integration control bus (see Control Bus) functionality. 

Enterprise integration patterns include the process manager pattern. You can now easily implement this pattern by using custom process manager logic encapsulated in a RoutingSlipRouteStrategy within the routing slip. In addition to a bean name, the RoutingSlipRouteStrategy can return any MessageChannel object, and there is no requirement that this MessageChannel instance be a bean in the application context. This way, we can provide powerful dynamic routing logic when there is no way to predict which channel should be used. A MessageChannel can be created within the RoutingSlipRouteStrategy and returned. A FixedSubscriberChannel with an associated MessageHandler implementation is a good combination for such cases. For example, you can route to a Reactive Streams, as the following example shows:

You can use the <filter> element is used to create a message-selecting endpoint. In addition to input-channel and output-channel attributes, it requires a ref attribute. The ref can point to a MessageSelector implementation, as the following example shows:

Alternatively, you can add the method attribute. In that case, the ref attribute may refer to any object. The referenced method may expect either the Message type or the payload type of inbound messages. The method must return a boolean value. If the method returns 'true', the message is sent to the output channel. The following example shows how to configure a filter that uses the method attribute:

If the selector or adapted POJO method returns false, a few settings control the handling of the rejected message. By default (if configured as in the preceding example), rejected messages are silently dropped. If rejection should instead result in an error condition, set the throw-exception-on-rejection attribute to true, as the following example shows:

If you want rejected messages to be routed to a specific channel, provide that reference as the discard-channel, as the following example shows:

See also Advising Filters.

We recommend using a ref attribute if the custom filter implementation is referenced in other <filter> definitions. However, if the custom filter implementation is scoped to a single <filter> element, you should provide an inner bean definition, as the following example shows:

With the introduction of SpEL support, Spring Integration added the expression attribute to the filter element. It can be used to avoid Java entirely for simple filters, as the following example shows:

The string passed as the value of the expression attribute is evaluated as a SpEL expression with the message available in the evaluation context. If you must include the result of an expression in the scope of the application context, you can use the #{} notation, as defined in the SpEL reference documentation, as the following example shows:

If the expression itself needs to be dynamic, you can use an 'expression' sub-element. That provides a level of indirection for resolving the expression by its key from an ExpressionSource. That is a strategy interface that you can implement directly, or you can rely upon a version available in Spring Integration that loads expressions from a “resource bundle” and can check for modifications after a given number of seconds. All of this is demonstrated in the following configuration example, where the expression could be reloaded within one minute if the underlying file had been modified:

If the ExpressionSource bean is named expressionSource, you need not provide the` source` attribute on the <expression> element. However, in the preceding example, we show it for completeness.

The 'config/integration/expressions.properties' file (or any more-specific version with a locale extension to be resolved in the typical way that resource-bundles are loaded) can contain a key/value pair, as the following example shows:

The following example shows how to configure a filter by using annotations:

All of the configuration options provided by the XML element are also available for the @Filter annotation.

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

See also Advising Endpoints Using Annotations.

The API for performing splitting consists of one base class, AbstractMessageSplitter. It is a MessageHandler implementation that encapsulates features common to splitters, such as filling in the appropriate message headers (CORRELATION_ID, SEQUENCE_SIZE, and SEQUENCE_NUMBER) on the messages that are produced. This filling enables tracking down the messages and the results of their processing (in a typical scenario, these headers get copied to the messages that are produced by the various transforming endpoints). The values can then be used, for example, by a composed message processor.

The following example shows an excerpt from AbstractMessageSplitter:

To implement a specific splitter in an application, you can extend AbstractMessageSplitter and implement the splitMessage method, which contains logic for splitting the messages. The return value can be one of the following:

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

A splitter can be configured through XML as follows:

We recommend using a ref attribute if the custom splitter implementation can be referenced in other <splitter> definitions. However if the custom splitter handler implementation should be scoped to a single definition of the <splitter>, you can configure an inner bean definition, as the following example follows:

The @Splitter annotation is applicable to methods that expect either the Message type or the message payload type, and the return values of the method should be a Collection of any type. If the returned values are not actual Message objects, each item is wrapped in a Message as the payload of the Message. Each resulting Message is sent to the designated output channel for the endpoint on which the @Splitter is defined.

The following example shows how to configure a splitter by using the @Splitter annotation:

See also Advising Endpoints Using Annotations.

See also Splitters in the Java DSL chapter.

The Aggregator combines a group of related messages, by correlating and storing them, until the group is deemed to be complete. At that point, the aggregator creates a single message by processing the whole group and sends the aggregated message as output.

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

Correlation determines how messages are grouped for aggregation. In Spring Integration, correlation is done by default, based on the IntegrationMessageHeaderAccessor.CORRELATION_ID message header. Messages with the same IntegrationMessageHeaderAccessor.CORRELATION_ID are grouped together. However, you can customize the correlation strategy to allow other ways of specifying how the messages should be grouped together. To do so, you can implement a CorrelationStrategy (covered later in this chapter).

To determine the point at which a group of messages is ready to be processed, a ReleaseStrategy is consulted. The default release strategy for the aggregator releases a group when all messages included in a sequence are present, based on the IntegrationMessageHeaderAccessor.SEQUENCE_SIZE header. You can override this default strategy by providing a reference to a custom ReleaseStrategy implementation.

The Aggregation API consists of a number of classes:

See Aggregators and Resequencers for how to configure an aggregator in Java DSL.

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

For more information, see the Javadoc.

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

The callback has direct access to the store and the message group so that it can manage the persistent state (for example, by entirely removing the group from the store).

The MessageGroupStore maintains a list of these callbacks, which it applies, on demand, to all messages whose timestamps are earlier than a time supplied as a parameter (see the registerMessageGroupExpiryCallback(..) and expireMessageGroups(..) methods, described earlier). For more detail, see Managing State in an Aggregator: MessageGroupStore.

You can call the expireMessageGroups method with a timeout value. Any message older than the current time minus this value is expired and has the callbacks applied. Thus, it is the user of the store that defines what is meant by message group “expiry”.

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

The reaper is a Runnable. In the preceding example, the message group store’s expire method is called every ten seconds. The timeout itself is 30 seconds.

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

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

In version 5.2, the FluxAggregatorMessageHandler component has been introduced. It is based on the Project Reactor Flux.groupBy() and Flux.window() operators. The incoming messages are emitted into the FluxSink initiated by the Flux.create() in the constructor of this component. If the outputChannel is not provided or it is not an instance of ReactiveStreamsSubscribableChannel, the subscription to the main Flux is done from the Lifecycle.start() implementation. Otherwise it is postponed to the subscription done by the ReactiveStreamsSubscribableChannel implementation. The messages are grouped by the Flux.groupBy() using a CorrelationStrategy for the group key. By default, the IntegrationMessageHeaderAccessor.CORRELATION_ID header of the message is consulted.

By default every closed window is released as a Flux in payload of a message to produce. This message contains all the headers from the first message in the window. This Flux in the output message payload must be subscribed and processed downstream. Such a logic can be customized (or superseded) by the setCombineFunction(Function<Flux<Message<?>>, Mono<Message<?>>>) configuration option of the FluxAggregatorMessageHandler. For example, if we would like to have a List of payloads in the final message, we can configure a Flux.collectList() like this:

There are several options in the FluxAggregatorMessageHandler to select an appropriate window strategy:

Since this component is a MessageHandler implementation it can simply be used as a @Bean definition together with a @ServiceActivator messaging annotation. With Java DSL it can be used from the .handle() EIP-method. The sample below demonstrates how we can register an IntegrationFlow at runtime and how a FluxAggregatorMessageHandler can be correlated with a splitter upstream:

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

With respect to that, you can opt to release all messages at once (after the whole sequence, according to the SEQUENCE_SIZE, and other possibilities) or as soon as a valid sequence is available. (We cover what we mean by "a valid sequence" later in this chapter.)

See Aggregators and Resequencers for configuring a resequencer in Java DSL.

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

The following example shows a resequencer configuration:

The <chain> element provides an input-channel attribute. If the last element in the chain is capable of producing reply messages (optional), it also supports an output-channel attribute. The sub-elements are then filters, transformers, splitters, and service-activators. The last element may also be a router or an outbound channel adapter. The following example shows a chain definition:

The <header-enricher> element used in the preceding example sets a message header named thing1 with a value of thing2 on the message. A header enricher is a specialization of Transformer that touches only header values. You could obtain the same result by implementing a MessageHandler that did the header modifications and wiring that as a bean, but the header-enricher is a simpler option.

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

Beginning with Spring Integration 3.0, if a chain element is given an id attribute, the bean name for the element is a combination of the chain’s id and the id of the element itself. Elements without id attributes are not registered as beans, but each one is given a componentName that includes the chain id. Consider the following example:

In the preceding example:

The id attribute for <chain> elements lets them be eligible for JMX export, and they are trackable in the message history. You can access them from the BeanFactory by using the appropriate bean name, as discussed earlier.

Sometimes, you need to make a nested call to another chain from within a chain and then come back and continue execution within the original chain. To accomplish this, you can use a messaging gateway by including a <gateway> element, as the following example shows:

In the preceding example, nested-chain-a is called at the end of main-chain processing by the 'gateway' element configured there. While in nested-chain-a, a call to a nested-chain-b is made after header enrichment. Then the flow comes back to finish execution in nested-chain-b. Finally, the flow returns to main-chain. When the nested version of a <gateway> element is defined in the chain, it does not require the service-interface attribute. Instead, it takes the message in its current state and places it on the channel defined in the request-channel attribute. When the downstream flow initiated by that gateway completes, a Message is returned to the gateway and continues its journey within the current chain.

The Scatter-Gather pattern suggests two scenarios: “auction” and “distribution”. In both cases, the aggregation function is the same and provides all the options available for the AggregatingMessageHandler. (Actually, the ScatterGatherHandler requires only an AggregatingMessageHandler as a constructor argument.) See Aggregator for more information.