The following parameters are valid for all routers inside and outside of chains.

The following parameters are valid only across all top-level routers that are outside of chains.

A PayloadTypeRouter sends messages to the channel defined by payload-type mappings, as the following example shows:

Configuration of the PayloadTypeRouter is also supported by the namespace provided by Spring Integration (see Namespace Support), which essentially simplifies configuration by combining the <router/> configuration and its corresponding implementation (defined by using a <bean/> element) into a single and more concise configuration element. The following example shows a PayloadTypeRouter configuration that is equivalent to the one above but uses the namespace support:

The following example shows the equivalent router configured in Java:

When using the Java DSL, there are two options.

First, you can define the router object as shown in the preceding example:

Note that the router can be, but does not have to be, a @Bean. The flow registers it if it is not a @Bean.

Second, you can define the routing function within the DSL flow itself, as the following example shows:

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

If it is an arbitrary value, additional mappings for these header values to channel names are required. Otherwise, no additional configuration is needed.

Spring Integration provides a simple namespace-based XML configuration to configure a HeaderValueRouter. The following example demonstrates configuration for the HeaderValueRouter when mapping of header values to channels is required:

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

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

The following example shows the equivalent router configured in Java:

When using the Java DSL, there are two options. First, you can define the router object as shown in the preceding example:

Note that the router can be, but does not have to be, a @Bean. The flow registers it if it is not a @Bean.

Second, you can define the routing function within the DSL flow itself, as the following example shows:

Configuration where mapping of header values to channel names is not required, because header values themselves represent channel names. The following example shows a router that does not require mapping of header values to channel names:

A RecipientListRouter sends each received message to a statically defined list of message channels. The following example creates a RecipientListRouter:

Spring Integration also provides namespace support for the RecipientListRouter configuration (see Namespace Support) as the following example shows:

The following example shows the equivalent router configured in Java:

The following example shows the equivalent router configured by using the Java DSL:

Another convenient option when configuring a RecipientListRouter is to use Spring Expression Language (SpEL) support as selectors for individual recipient channels. Doing so is similar to using a filter at the beginning of a 'chain' to act as a “selective consumer”. However, in this case, it is all combined rather concisely into the router’s configuration, as the following example shows:

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

Starting with version 4.1, the RecipientListRouter provides several operations to manipulate recipients dynamically at runtime. These management operations are presented by RecipientListRouterManagement through the @ManagedResource annotation. They are available by using Control Bus as well as by using JMX, as the following example shows:

From the application start up the simpleRouter, has only one channel1 recipient. But after the addRecipient command, channel2 recipient is added. It is a “registering an interest in something that is part of the message” use case, when we may be interested in messages from the router at some time period, so we are subscribing to the recipient-list-router and, at some point, decide to unsubscribe.

Because of the runtime management operation for the <recipient-list-router>, it can be configured without any <recipient> from the start. In this case, the behavior of RecipientListRouter is the same when there is no one matching recipient for the message. If defaultOutputChannel is configured, the message is sent there. Otherwise, the MessageDeliveryException is thrown.

The XPath Router is part of the XML Module. See Routing XML Messages with XPath.

Spring Integration also provides a special type-based router called ErrorMessageExceptionTypeRouter for routing error messages (defined as messages whose payload is a Throwable instance). ErrorMessageExceptionTypeRouter is similar to the PayloadTypeRouter. In fact, they are almost identical. The only difference is that, while PayloadTypeRouter navigates the instance hierarchy of a payload instance (for example, payload.getClass().getSuperclass()) to find the most specific type and channel mappings, the ErrorMessageExceptionTypeRouter navigates the hierarchy of 'exception causes' (for example, payload.getCause()) to find the most specific Throwable type or channel mappings and uses mappingClass.isInstance(cause) to match the cause to the class or any super class.

The following example shows a sample configuration for ErrorMessageExceptionTypeRouter:

The router element provides a way to connect a router to an input channel and also accepts the optional default-output-channel attribute. The ref attribute references the bean name of a custom router implementation (which must extend AbstractMessageRouter). The following example shows three generic routers:

Alternatively, ref may point to a POJO that contains the @Router annotation (shown later), or you can combine the ref with an explicit method name. Specifying a method applies the same behavior described in the @Router annotation section, later in this document. The following example defines a router that points to a POJO in its ref attribute:

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

The following example shows the equivalent router configured in Java:

The following example shows the equivalent router configured by using the Java DSL:

Alternately, you can route on data from the message payload, as the following example shows:

When using @Router to annotate a method, the method may return either a MessageChannel or a String type. In the latter case, the endpoint resolves the channel name as it does for the default output channel. Additionally, the method may return either a single value or a collection. If a collection is returned, the reply message is sent to multiple channels. To summarize, the following method signatures are all valid:

In addition to payload-based routing, a message may be routed based on metadata available within the message header as either a property or an attribute. In this case, a method annotated with @Router may include a parameter annotated with @Header, which is mapped to a header value as the following example shows and documented in Annotation Support:

See also Message Routers in the Java DSL chapter for more information about router configuration.

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

Typically, you would send a control message asking to invoke a particular operation on a particular managed component (such as a router). The following managed operations (methods) are specific to changing the router resolution process:

Note that these methods can be used for simple changes (such as updating a single route or adding or removing a route). However, if you want to remove one route and add another, the updates are not atomic. This means that the routing table may be in an indeterminate state between the updates. Starting with version 4.0, you can now use the control bus to update the entire routing table atomically. The following methods let you do so:

Note that each mapping is separated by a newline character (\n). For programmatic changes to the map, we recommend that you use the setChannelMappings method, due to type-safety concerns. replaceChannelMappings ignores keys or values that are not String objects.

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

Starting with version 4.1, Spring Integration provides an implementation of the routing slip enterprise integration pattern. It is implemented as a routingSlip message header, which is used to determine the next channel in AbstractMessageProducingHandler instances, when an outputChannel is not specified for the endpoint. This pattern is useful in complex, dynamic cases, when it can become difficult to configure multiple routers to determine message flow. When a message arrives at an endpoint that has no output-channel, the routingSlip is consulted to determine the next channel to which the message is sent. When the routing slip is exhausted, normal replyChannel processing resumes.

Configuration for the routing slip is presented as a HeaderEnricher option — a semicolon-separated routing slip that contains path entries, as the following example shows:

The preceding example has:

Routing Slip path entries can contain MessageChannel bean names, RoutingSlipRouteStrategy bean names, and Spring expressions (SpEL). The RoutingSlipHeaderValueMessageProcessor checks each routing slip path entry against the BeanFactory on the first processMessage invocation. It converts entries (which are not bean names in the application context) to ExpressionEvaluatingRoutingSlipRouteStrategy instances. RoutingSlipRouteStrategy entries are invoked multiple times, until they return null or an empty String.

Since the routing slip is involved in the getOutputChannel process, we have a request-reply context. The RoutingSlipRouteStrategy has been introduced to determine the next outputChannel that uses the requestMessage and the reply object. An implementation of this strategy should be registered as a bean in the application context, and its bean name is used in the routing slip path. The ExpressionEvaluatingRoutingSlipRouteStrategy implementation is provided. It accepts a SpEL expression and an internal ExpressionEvaluatingRoutingSlipRouteStrategy.RequestAndReply object is used as the root object of the evaluation context. This is to avoid the overhead of EvaluationContext creation for each ExpressionEvaluatingRoutingSlipRouteStrategy.getNextPath() invocation. It is a simple Java bean with two properties: Message<?> request and Object reply. With this expression implementation, we can specify routing slip path entries by using SpEL (for example, @routingSlipRoutingPojo.get(request, reply) and request.headers[myRoutingSlipChannel]) and avoid defining a bean for the RoutingSlipRouteStrategy.

For Java configuration, you can add a RoutingSlipHeaderValueMessageProcessor instance to the HeaderEnricher bean definition, as the following example shows:

The routing slip algorithm works as follows when an endpoint produces a reply and no outputChannel has been defined:

Starting with version 4.1, the AbstractMessageSplitter supports the Iterator type for the value to split. Note, in the case of an Iterator (or Iterable), we don’t have access to the number of underlying items and the SEQUENCE_SIZE header is set to 0. This means that the default SequenceSizeReleaseStrategy of an <aggregator> won’t work and the group for the CORRELATION_ID from the splitter won’t be released; it will remain as incomplete. In this case you should use an appropriate custom ReleaseStrategy or rely on send-partial-result-on-expiry together with group-timeout or a MessageGroupStoreReaper.

Starting with version 5.0, the AbstractMessageSplitter provides protected obtainSizeIfPossible() methods to allow the determination of the size of the Iterable and Iterator objects if that is possible. For example XPathMessageSplitter can determine the size of the underlying NodeList object. And starting with version 5.0.9, this method also properly returns a size of the com.fasterxml.jackson.core.TreeNode.

An Iterator object is useful to avoid the need for building an entire collection in the memory before splitting. For example, when underlying items are populated from some external system (e.g. DataBase or FTP MGET) using iterations or streams.

Starting with version 5.0, the AbstractMessageSplitter supports the Java Stream and Reactive Streams Publisher types for the value to split. In this case, the target Iterator is built on their iteration functionality.

In addition, if the splitter’s output channel is an instance of a ReactiveStreamsSubscribableChannel, the AbstractMessageSplitter produces a Flux result instead of an Iterator, and the output channel is subscribed to this Flux for back-pressure-based splitting on downstream flow demand.

Starting with version 5.2, the splitter supports a discardChannel option for sending those request messages for which a split function has returned an empty container (collection, array, stream, Flux etc.). In this case there is just no item to iterate for sending to the outputChannel. The null splitting result remains as an end of flow indicator.

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

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

The following listing shows a brief highlight of the base AbstractAggregatingMessageGroupProcessor (the responsibility for implementing the aggregatePayloads method is left to the developer):

See DefaultAggregatingMessageGroupProcessor, ExpressionEvaluatingMessageGroupProcessor and MethodInvokingMessageGroupProcessor as out-of-the-box implementations of the AbstractAggregatingMessageGroupProcessor.

Starting with version 5.2, a Function<MessageGroup, Map<String, Object>> strategy is available for the AbstractAggregatingMessageGroupProcessor to merge and compute (aggregate) headers for an output message. The DefaultAggregateHeadersFunction implementation is available with logic that returns all headers that have no conflicts among the group; an absent header on one or more messages within the group is not considered a conflict. Conflicting headers are omitted. Along with the newly introduced DelegatingMessageGroupProcessor, this function is used for any arbitrary (non-AbstractAggregatingMessageGroupProcessor) MessageGroupProcessor implementation. Essentially, the framework injects a provided function into an AbstractAggregatingMessageGroupProcessor instance and wraps all other implementations into a DelegatingMessageGroupProcessor. The difference in logic between the AbstractAggregatingMessageGroupProcessor and the DelegatingMessageGroupProcessor that the latter doesn’t compute headers in advance, before calling the delegate strategy, and doesn’t invoke the function if the delegate returns a Message or AbstractIntegrationMessageBuilder. In that case, the framework assumes that the target implementation has taken care of producing a proper set of headers populated into the returned result. The Function<MessageGroup, Map<String, Object>> strategy is available as the headers-function reference attribute for XML configuration, as the AggregatorSpec.headersFunction() option for the Java DSL and as AggregatorFactoryBean.setHeadersFunction() for plain Java configuration.

The CorrelationStrategy is owned by the AbstractCorrelatingMessageHandler and has a default value based on the IntegrationMessageHeaderAccessor.CORRELATION_ID message header, as the following example shows:

As for the actual processing of the message group, the default implementation is the DefaultAggregatingMessageGroupProcessor. It creates a single Message whose payload is a List of the payloads received for a given group. This works well for simple scatter-gather implementations with a splitter, a publish-subscribe channel, or a recipient list router upstream.

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

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

Starting with version 5.3, after processing message group, an AbstractCorrelatingMessageHandler performs a MessageBuilder.popSequenceDetails() message headers modification for the proper splitter-aggregator scenario with several nested levels. It is done only if the message group release result is not a collection of messages. In that case a target MessageGroupProcessor is responsible for the MessageBuilder.popSequenceDetails() call while building those messages.

If the MessageGroupProcessor returns a Message, a MessageBuilder.popSequenceDetails() will be performed on the output message only if the sequenceDetails matches with first message in group. (Previously this has been done only if a plain payload or an AbstractIntegrationMessageBuilder has been returned from the MessageGroupProcessor.)

This functionality can be controlled by a new popSequence boolean property, so the MessageBuilder.popSequenceDetails() can be disabled in some scenarios when correlation details have not been populated by the standard splitter. This property, essentially, undoes what has been done by the nearest upstream applySequence = true in the AbstractMessageSplitter. See Splitter for more information.

Prior to version 4.2, it was not possible to provide a MessageGroupProcessor by using XML configuration. Only POJO methods could be used for aggregation. Now, if the framework detects that the referenced (or inner) bean implements MessageProcessor, it is used as the aggregator’s output processor.

If you wish to release a collection of objects from a custom MessageGroupProcessor as the payload of a message, your class should extend AbstractAggregatingMessageGroupProcessor and implement aggregatePayloads().

Also, since version 4.2, a SimpleMessageGroupProcessor is provided. It returns the collection of messages from the group, which, as indicated earlier, causes the released messages to be sent individually.

This lets the aggregator work as a message barrier, where arriving messages are held until the release strategy fires and the group is released as a sequence of individual messages.

Starting with version 6.0, the splitting behaviour, described above, works only if the group processor is a SimpleMessageGroupProcessor. Otherwise, with any other MessageGroupProcessor implementation that returns a Collection<Message>, only a single reply message is emitted with the whole collection of messages as its payload. Such logic is dictated by the canonical purpose of an aggregator - collect request messages by some key and produce a single grouped message.

The ReleaseStrategy interface is defined as follows:

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

The following example shows how to use the @ReleaseStrategy annotation for a List of type Message:

The following example shows how to use the @ReleaseStrategy annotation for a List of type String:

Based on the signatures in the preceding two examples, the POJO-based release strategy is passed a Collection of not-yet-released messages (if you need access to the whole Message) or a Collection of payload objects (if the type parameter is anything other than Message). This satisfies the majority of use cases. However, if, for some reason, you need to access the full MessageGroup, you should provide an implementation of the ReleaseStrategy interface.

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

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

If you are aggregating large groups, you don’t need to release partial groups, and you don’t need to detect/reject duplicate sequences, consider using the SimpleSequenceSizeReleaseStrategy instead - it is much more efficient for these use cases, and is the default since version 5.0 when partial group release is not specified.

The 4.3 release changed the default Collection for messages in a SimpleMessageGroup to HashSet (it was previously a BlockingQueue). This was expensive when removing individual messages from large groups (an O(n) linear scan was required). Although the hash set is generally much faster to remove, it can be expensive for large messages, because the hash has to be calculated on both inserts and removes. If you have messages that are expensive to hash, consider using some other collection type. As discussed in Using MessageGroupFactory, a SimpleMessageGroupFactory is provided so that you can select the Collection that best suits your needs. You can also provide your own factory implementation to create some other Collection<Message<?>>.

The following example shows how to configure an aggregator with the previous implementation and a SimpleSequenceSizeReleaseStrategy:

The CorrelationStrategy interface is defined as follows:

The method returns an Object that represents the correlation key used for associating the message with a message group. The key must satisfy the criteria used for a key in a Map with respect to the implementation of equals() and hashCode().

In general, any POJO can implement the correlation logic, and the rules for mapping a message to a method’s argument (or arguments) are the same as for a ServiceActivator (including support for @Header annotations). The method must return a value, and the value must not be null.

Spring Integration provides an implementation for CorrelationStrategy: HeaderAttributeCorrelationStrategy. This implementation returns the value of one of the message headers (whose name is specified by a constructor argument) as the correlation key. By default, the correlation strategy is a HeaderAttributeCorrelationStrategy that returns the value of the CORRELATION_ID header attribute. If you have a custom header name you would like to use for correlation, you can configure it on an instance of HeaderAttributeCorrelationStrategy and provide that as a reference for the aggregator’s correlation strategy.

Changes to groups are thread safe. So, when you send messages for the same correlation ID concurrently, only one of them will be processed in the aggregator, making it effectively as a single-threaded per message group. A LockRegistry is used to obtain a lock for the resolved correlation ID. A DefaultLockRegistry is used by default (in-memory). For synchronizing updates across servers where a shared MessageGroupStore is being used, you must configure a shared lock registry.

As discussed above, when message groups are mutated (messages added or released) a lock is held.

Consider the following flow:

If there are multiple threads, and the aggregators share a common lock registry, it is possible to get a deadlock. This will cause hung threads and jstack <pid> might present a result such as:

There are several ways to avoid this problem:

Spring Integration supports the configuration of an aggregator with XML through the <aggregator/> element. The following example shows an example of an aggregator:

We generally recommend using a ref attribute if a custom aggregator handler implementation may be referenced in other <aggregator> definitions. However, if a custom aggregator implementation is only being used by a single definition of the <aggregator>, you can use an inner bean definition (starting with version 1.0.3) to configure the aggregation POJO within the <aggregator> element, as the following example shows:

The following example shows an implementation of the aggregator bean:

An implementation of the completion strategy bean for the preceding example might be as follows:

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

The aggregator in the preceding example would group numbers by some criterion (in this case, the remainder after dividing by ten) and hold the group until the sum of the numbers provided by the payloads exceeds a certain value.

The following example shows an aggregator configured with annotations:

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

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

Annotation configuration (@Aggregator and others) for the Aggregator component covers only simple use cases, where most default options are sufficient. If you need more control over those options when using annotation configuration, consider using a @Bean definition for the AggregatingMessageHandler and mark its @Bean method with @ServiceActivator, as the following example shows:

See Programming Model and Annotations on @Bean Methods for more information.

The auction Scatter-Gather variant uses “publish-subscribe” logic for the request message, where the “scatter” channel is a PublishSubscribeChannel with apply-sequence="true". However, this channel can be any MessageChannel implementation (as is the case with the request-channel in the ContentEnricher — see Content Enricher). However, in this case, you should create your own custom correlationStrategy for the aggregation function.

The distribution Scatter-Gather variant is based on the RecipientListRouter (see RecipientListRouter) with all available options for the RecipientListRouter. This is the second ScatterGatherHandler constructor argument. If you want to rely on only the default correlationStrategy for the recipient-list-router and the aggregator, you should specify apply-sequence="true". Otherwise, you should supply a custom correlationStrategy for the aggregator. Unlike the PublishSubscribeChannel variant (the auction variant), having a recipient-list-router selector option lets filter target suppliers based on the message. With apply-sequence="true", the default sequenceSize is supplied, and the aggregator can release the group correctly. The distribution option is mutually exclusive with the auction option.

For both the auction and the distribution variants, the request (scatter) message is enriched with the gatherResultChannel header to wait for a reply message from the aggregator.

By default, all suppliers should send their result to the replyChannel header (usually by omitting the output-channel from the ultimate endpoint). However, the gatherChannel option is also provided, letting suppliers send their reply to that channel for the aggregation.