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:

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

Configuration of the PayloadTypeRouter is also supported by the namespace provided by Spring Integration (see <<configuration-namespace>>), 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:

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

The following example shows the equivalent router configured in Java:

@ServiceActivator(inputChannel = "routingChannel")
@Bean
public PayloadTypeRouter router() {
    PayloadTypeRouter router = new PayloadTypeRouter();
    router.setChannelMapping(String.class.getName(), "stringChannel");
    router.setChannelMapping(Integer.class.getName(), "integerChannel");
    return router;
}

When using the Java DSL, there are two options.

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

@Bean
public IntegrationFlow routerFlow1() {
    return IntegrationFlows.from("routingChannel")
            .route(router())
            .get();
}

public PayloadTypeRouter router() {
    PayloadTypeRouter router = new PayloadTypeRouter();
    router.setChannelMapping(String.class.getName(), "stringChannel");
    router.setChannelMapping(Integer.class.getName(), "integerChannel");
    return router;
}

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:

@Bean
public IntegrationFlow routerFlow2() {
    return IntegrationFlows.from("routingChannel")
            .<Object, Class<?>>route(Object::getClass, m -> m
                    .channelMapping(String.class, "stringChannel")
                    .channelMapping(Integer.class, "integerChannel"))
            .get();
}

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:

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

During the resolution process, 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.

	Important
	As of Spring Integration 2.1, the attribute was changed from ignore-channel-name-resolution-failures to resolution-required. Attribute resolution-required defaults to true.

The following example shows the equivalent router configured in Java:

@ServiceActivator(inputChannel = "routingChannel")
@Bean
public HeaderValueRouter router() {
    HeaderValueRouter router = new HeaderValueRouter("testHeader");
    router.setChannelMapping("someHeaderValue", "channelA");
    router.setChannelMapping("someOtherHeaderValue", "channelB");
    return router;
}

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

@Bean
public IntegrationFlow routerFlow1() {
    return IntegrationFlows.from("routingChannel")
            .route(router())
            .get();
}

public HeaderValueRouter router() {
    HeaderValueRouter router = new HeaderValueRouter("testHeader");
    router.setChannelMapping("someHeaderValue", "channelA");
    router.setChannelMapping("someOtherHeaderValue", "channelB");
    return router;
}

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:

@Bean
public IntegrationFlow routerFlow2() {
    return IntegrationFlows.from("routingChannel")
            .<Message<?>, String>route(m -> m.getHeaders().get("testHeader", String.class), m -> m
                    .channelMapping("someHeaderValue", "channelA")
                    .channelMapping("someOtherHeaderValue", "channelB"),
                e -> e.id("headerValueRouter"))
            .get();
}

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:

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

Note

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

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

<bean id="recipientListRouter"
      class="org.springframework.integration.router.RecipientListRouter">
    <property name="channels">
        <list>
            <ref bean="channel1"/>
            <ref bean="channel2"/>
            <ref bean="channel3"/>
        </list>
    </property>
</bean>

Spring Integration also provides namespace support for the RecipientListRouter configuration (see Section E.1, “Namespace Support”) as the following example shows:

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

The following example shows the equivalent router configured in Java:

@ServiceActivator(inputChannel = "routingChannel")
@Bean
public RecipientListRouter router() {
    RecipientListRouter router = new RecipientListRouter();
    router.setSendTimeout(1_234L);
    router.setIgnoreSendFailures(true);
    router.setApplySequence(true);
    router.addRecipient("channel1");
    router.addRecipient("channel2");
    router.addRecipient("channel3");
    return router;
}

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

@Bean
public IntegrationFlow routerFlow() {
    return IntegrationFlows.from("routingChannel")
            .routeToRecipients(r -> r
                    .applySequence(true)
                    .ignoreSendFailures(true)
                    .recipient("channel1")
                    .recipient("channel2")
                    .recipient("channel3")
                    .sendTimeout(1_234L))
            .get();
}

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

Another convenient option when configuring a RecipientListRouter is to use Spring Expression Language (SpEL) support as selectors for individual recipient channels. 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:

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

In the 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 Section 12.6, “Control Bus” as well as by using JMX, as the following example shows:

<control-bus input-channel="controlBus"/>

<recipient-list-router id="simpleRouter" input-channel="routingChannelA">
   <recipient channel="channel1"/>
</recipient-list-router>

<channel id="channel2"/>

messagingTemplate.convertAndSend(controlBus, "@'simpleRouter.handler'.addRecipient('channel2')");

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 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 Section 38.5, “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.

	Note
	Since version 4.3 the ErrorMessageExceptionTypeRouter loads all mapping classes during the initialization phase to fail-fast for a ClassNotFoundException.

The following example shows a sample configuration for ErrorMessageExceptionTypeRouter:

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

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

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:

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

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

<int:router ref="customRouter" input-channel="input3"
            default-output-channel="defaultOutput3"/>

<beans:bean id="customRouterBean" class="org.foo.MyCustomRouter"/>

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:

<int:router input-channel="input" ref="somePojo" method="someMethod"/>

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:

<int:router method="someMethod" input-channel="input3"
            default-output-channel="defaultOutput3">
    <beans:bean class="org.foo.MyCustomRouter"/>
</int:router>

	Note
	Using both the ref attribute and an inner handler definition in the same <router> configuration is not allowed. Doing so creates an ambiguous condition and throws an exception.

Important

If the ref attribute references a bean that extends AbstractMessageProducingHandler (such as routers provided by the framework itself), the configuration is optimized to reference the router directly. In this case, each ref attribute must refer to a separate bean instance (or a prototype-scoped bean) or use the inner <bean/> configuration type. However, this optimization applies only if you do not provide any router-specific attributes in the router XML definition. If you inadvertently reference the same message handler from multiple beans, you get a configuration exception.

The following example shows the equivalent router configured in Java:

@Bean
@Router(inputChannel = "routingChannel")
public AbstractMessageRouter myCustomRouter() {
    return new AbstractMessageRouter() {

        @Override
        protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
            return // determine channel(s) for message
        }

    };
}

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

@Bean
public IntegrationFlow routerFlow() {
    return IntegrationFlows.from("routingChannel")
            .route(myCustomRouter())
            .get();
}

public AbstractMessageRouter myCustomRouter() {
    return new AbstractMessageRouter() {

        @Override
        protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
            return // determine channel(s) for message
        }

    };
}

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

@Bean
public IntegrationFlow routerFlow() {
    return IntegrationFlows.from("routingChannel")
            .route(String.class, p -> p.contains("foo") ? "fooChannel" : "barChannel")
            .get();
}

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:

@Router
public MessageChannel route(Message message) {...}

@Router
public List<MessageChannel> route(Message message) {...}

@Router
public String route(Foo payload) {...}

@Router
public List<String> route(Foo payload) {...}

In addition to payload-based routing, a 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 Section E.5, “Annotation Support”:

@Router
public List<String> route(@Header("orderStatus") OrderStatus status)

	Note
	For routing of XML-based Messages, including XPath support, see Chapter 38, XML Support - Dealing with XML Payloads.

See also Section 11.7, “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.

	Note
	For more information about the control bus, see Section 12.6, “Control Bus”.

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:

"@'router.handler'.replaceChannelMappings('foo=qux \n baz=bar')"

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.

	Note
	For more information about Spring Integration’s JMX support, see Section 12.2, “JMX Support”.

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:

<util:properties id="properties">
    <beans:prop key="myRoutePath1">channel1</beans:prop>
    <beans:prop key="myRoutePath2">request.headers[myRoutingSlipChannel]</beans:prop>
</util:properties>

<context:property-placeholder properties-ref="properties"/>

<header-enricher input-channel="input" output-channel="process">
    <routing-slip
        value="${myRoutePath1}; @routingSlipRoutingPojo.get(request, reply);
               routingSlipRoutingStrategy; ${myRoutePath2}; finishChannel"/>
</header-enricher>

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.

Note

The requestMessage argument is always a Message<?>. Depending on context, the reply object may be a Message<?>, an AbstractIntegrationMessageBuilder, or an arbitrary application domain object (when, for example, it is returned by a POJO method invoked by a service activator). In the first two cases, the usual Message properties (payload and headers) are available when using SpEL (or a Java implementation). For an arbitrary domain object, these properties are not available. For this reason, be careful when you use routing slips in conjunction with POJO methods if the result is used to determine the next path.

Important

If a routing slip is involved in a distributed environment, we recommend not using inline expressions for the Routing Slip path. This recommendation applies to distributed environments such as cross-JVM applications, using a request-reply through a message broker (such asChapter 14, AMQP Support or Chapter 23, JMS Support), or using a persistent MessageStore (Section 12.4, “Message Store”) in the integration flow. The framework uses RoutingSlipHeaderValueMessageProcessor to convert them to ExpressionEvaluatingRoutingSlipRouteStrategy objects, and they are used in the routingSlip message header. Since this class is not Serializable (it cannot be, because it depends on the BeanFactory), the entire Message becomes non-serializable and, in any distributed operation, we end up with a NotSerializableException. To overcome this limitation, register an ExpressionEvaluatingRoutingSlipRouteStrategy bean with the desired SpEL and use its bean name in the routing slip path configuration.

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

@Bean
@Transformer(inputChannel = "routingSlipHeaderChannel")
public HeaderEnricher headerEnricher() {
    return new HeaderEnricher(Collections.singletonMap(IntegrationMessageHeaderAccessor.ROUTING_SLIP,
            new RoutingSlipHeaderValueMessageProcessor("myRoutePath1",
                                                       "@routingSlipRoutingPojo.get(request, reply)",
                                                       "routingSlipRoutingStrategy",
                                                       "request.headers[myRoutingSlipChannel]",
                                                       "finishChannel")));
}

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

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 reactor stream, as the following example shows:

@Bean
public PollableChannel resultsChannel() {
    return new QueueChannel();
}
@Bean
public RoutingSlipRouteStrategy routeStrategy() {
    return (requestMessage, reply) -> requestMessage.getPayload() instanceof String
            ? new FixedSubscriberChannel(m ->
            Mono.just((String) m.getPayload())
                    .map(String::toUpperCase)
                    .subscribe(v -> messagingTemplate().convertAndSend(resultsChannel(), v)))
            : new FixedSubscriberChannel(m ->
            Mono.just((Integer) m.getPayload())
                    .map(v -> v * 2)
                    .subscribe(v -> messagingTemplate().convertAndSend(resultsChannel(), v)));
}

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.

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

public abstract class AbstractAggregatingMessageGroupProcessor
              implements MessageGroupProcessor {

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

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

}

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

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

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.

Note

When using a publish-subscribe channel or a recipient list router in this type of scenario, be sure to enable the apply-sequence flag. Doing so adds the necessary headers: CORRELATION_ID, SEQUENCE_NUMBER, and SEQUENCE_SIZE. That behavior is enabled by default for splitters in Spring Integration, but it is not enabled for publish-subscribe channels or for recipient list routers because those components may be used in a variety of contexts in which these headers are not necessary.

When implementing a specific aggregator strategy for an application, 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:

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

Starting with version 5.1, 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 message or collection of messages. In that case a target MessageGroupProcessor is responsible for the MessageBuilder.popSequenceDetails() call while building those messages. 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 Section 8.3, “Splitter” for more information.

Important

The SimpleMessageGroup.getMessages() method returns an unmodifiableCollection. Therefore, if your aggregating POJO method has a Collection<Message> parameter, the argument passed in is exactly that Collection instance and, when you use a SimpleMessageStore for the aggregator, that original Collection<Message> is cleared after releasing the group. Consequently, the Collection<Message> variable in the POJO is cleared too, if it is passed out of the aggregator. If you wish to simply release that collection as-is for further processing, you must build a new Collection (for example, new ArrayList<Message>(messages)). Starting with version 4.3, the framework no longer copies the messages to a new collection, to avoid undesired extra object creation.

If the processMessageGroup method of the MessageGroupProcessor returns a collection, it must be a collection of Message<?> objects. In this case, the messages are individually released. 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.

The ReleaseStrategy interface is defined as follows:

public interface ReleaseStrategy {

  boolean canRelease(MessageGroup group);

}

In general, any POJO can implement the completion decision logic if it provides a method that accepts a single java.util.List as an argument (parameterized lists are supported as well) and returns a boolean value. This method 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:

public class MyReleaseStrategy {

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

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

public class MyReleaseStrategy {

    @ReleaseStrategy
    public boolean canMessagesBeReleased(List<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.

Warning

When handling potentially large groups, you should understand how these methods are invoked, because the release strategy may be invoked multiple times before the group is released. The most efficient is an implementation of ReleaseStrategy, because the aggregator can invoke it directly. The second most efficient is a POJO method with a Collection<Message<?>> parameter type. The least efficient is a POJO method with a Collection<Something> type. The framework has to copy the payloads from the messages in the group into a new collection (and possibly attempt conversion on the payloads to Something) every time the release strategy is called. Using Collection<?> avoids the conversion but still requires creating the new Collection. For these reasons, for large groups, we recommended that you implement ReleaseStrategy.

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.

Important

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

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.

	Note
	Before version 5.0, the default release strategy was SequenceSizeReleaseStrategy, which does not perform well with large groups. With that strategy, duplicate sequence numbers are detected and rejected. This operation can be expensive.

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 Section 12.4.1, “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:

<int:aggregator input-channel="aggregate"
    output-channel="out" message-store="store" release-strategy="releaser" />

<bean id="store" class="org.springframework.integration.store.SimpleMessageStore">
    <property name="messageGroupFactory">
        <bean class="org.springframework.integration.store.SimpleMessageGroupFactory">
            <constructor-arg value="BLOCKING_QUEUE"/>
        </bean>
    </property>
</bean>

<bean id="releaser" class="SimpleSequenceSizeReleaseStrategy" />

The CorrelationStrategy interface is defined as follows:

public interface CorrelationStrategy {

  Object getCorrelationKey(Message<?> message);

}

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:

...->aggregator1-> ... ->aggregator2-> ...

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:

Found one Java-level deadlock:
=============================
"t2":
  waiting for ownable synchronizer 0x000000076c1cbfa0, (a java.util.concurrent.locks.ReentrantLock$NonfairSync),
  which is held by "t1"
"t1":
  waiting for ownable synchronizer 0x000000076c1ccc00, (a java.util.concurrent.locks.ReentrantLock$NonfairSync),
  which is held by "t2"

There are several ways to avoid this problem:

	Note
	This problem can also be caused if, for some reason, the output of a single aggregator is eventually routed back to the same aggregator. Of course, the first solution above does not apply in this case.

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

<channel id="inputChannel"/>

<int:aggregator id="myAggregator"                          
        auto-startup="true"                                
        input-channel="inputChannel"                       
        output-channel="outputChannel"                     
        discard-channel="throwAwayChannel"                 
        message-store="persistentMessageStore"             
        order="1"                                          
        send-partial-result-on-expiry="false"              
        send-timeout="1000"                                

        correlation-strategy="correlationStrategyBean"     
        correlation-strategy-method="correlate"            
        correlation-strategy-expression="headers['foo']"   

        ref="aggregatorBean"                               
        method="aggregate"                                 

        release-strategy="releaseStrategyBean"             
        release-strategy-method="release"                  (16)
        release-strategy-expression="size() == 5"          (17)

        expire-groups-upon-completion="false"              (18)
        empty-group-min-timeout="60000"                    (19)

        lock-registry="lockRegistry"                       (20)

        group-timeout="60000"                              (21)
        group-timeout-expression="size() ge 2 ? 100 : -1"  (22)
        expire-groups-upon-timeout="true"                  (23)

        scheduler="taskScheduler" >                        (24)
            <expire-transactional/>                        (25)
            <expire-advice-chain/>                         (26)
</aggregator>

<int:channel id="outputChannel"/>

<int:channel id="throwAwayChannel"/>

<bean id="persistentMessageStore" class="org.springframework.integration.jdbc.store.JdbcMessageStore">
    <constructor-arg ref="dataSource"/>
</bean>

<bean id="aggregatorBean" class="sample.PojoAggregator"/>

<bean id="releaseStrategyBean" class="sample.PojoReleaseStrategy"/>

<bean id="correlationStrategyBean" class="sample.PojoCorrelationStrategy"/>

Expiring Groups

There are two attributes related to expiring (completely removing) groups. When a group is expired, there is no record of it, and, if a new message arrives with the same correlation, a new group is started. When a group is completed (without expiry), the empty group remains and late-arriving messages are discarded. Empty groups can be removed later by using a MessageGroupStoreReaper in combination with the empty-group-min-timeout attribute. expire-groups-upon-completion relates to "normal" completion when the ReleaseStrategy releases the group. This defaults to false. If a group is not completed normally but is released or discarded because of a timeout, the group is normally expired. Since version 4.1, you can control this behavior by using expire-groups-upon-timeout. It defaults to true for backwards compatibility. Note When a group is timed out, the ReleaseStrategy is given one more opportunity to release the group. If it does so and expire-groups-upon-timeout is false, expiration is controlled by expire-groups-upon-completion. If the group is not released by the release strategy during timeout, then the expiration is controlled by the expire-groups-upon-timeout. Timed-out groups are either discarded or a partial release occurs (based on send-partial-result-on-expiry). Since version 5.0, empty groups are also scheduled for removal after empty-group-min-timeout. If expireGroupsUponCompletion == false and minimumTimeoutForEmptyGroups > 0, the task to remove the group is scheduled when normal or partial sequences release happens.

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:

<aggregator input-channel="input" method="sum" output-channel="output">
    <beans:bean class="org.foo.PojoAggregator"/>
</aggregator>

	Note
	Using both a ref attribute and an inner bean definition in the same <aggregator> configuration is not allowed, as it creates an ambiguous condition. In such cases, an Exception is thrown.

The following example shows an implementation of the aggregator bean:

public class PojoAggregator {

  public Long add(List<Long> results) {
    long total = 0l;
    for (long partialResult: results) {
      total += partialResult;
    }
    return total;
  }
}

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

public class PojoReleaseStrategy {
...
  public boolean canRelease(List<Long> numbers) {
    int sum = 0;
    for (long number: numbers) {
      sum += number;
    }
    return sum >= maxValue;
  }
}

	Note
	Wherever it makes sense to do so, the release strategy method and the aggregator method can be combined into a single bean.

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

public class PojoCorrelationStrategy {
...
  public Long groupNumbersByLastDigit(Long number) {
    return number % 10;
  }
}

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.

	Note
	Wherever it makes sense to do so, the release strategy method, the correlation strategy method, and the aggregator method can be combined in a single bean. (Actually, all of them or any two of them can be combined.)

Aggregators and Spring Expression Language (SpEL)

Since Spring Integration 2.0, you can handle the various strategies (correlation, release, and aggregation) with SpEL, which we recommend if the logic behind such a release strategy is relatively simple. Suppose you have a legacy component that was designed to receive an array of objects. We know that the default release strategy assembles all aggregated messages in the List. Now we have two problems. First, we need to extract individual messages from the list. Second, we need to extract the payload of each message and assemble the array of objects. The following example solves both problems:

public String[] processRelease(List<Message<String>> messages){
    List<String> stringList = new ArrayList<String>();
    for (Message<String> message : messages) {
        stringList.add(message.getPayload());
    }
    return stringList.toArray(new String[]{});
}

However, with SpEL, such a requirement could actually be handled relatively easily with a one-line expression, thus sparing you from writing a custom class and configuring it as a bean. The following example shows how to do so:

<int:aggregator input-channel="aggChannel"
    output-channel="replyChannel"
    expression="#this.![payload].toArray()"/>

In the preceding configuration, we use a collection projection expression to assemble a new collection from the payloads of all the messages in the list and then transform it to an array, thus achieving the same result as the earlier Java code.

You can apply the same expression-based approach when dealing with custom release and correlation strategies.

Instead of defining a bean for a custom CorrelationStrategy in the correlation-strategy attribute, you can implement your simple correlation logic as a SpEL expression and configure it in the correlation-strategy-expression attribute, as the following example shows:

correlation-strategy-expression="payload.person.id"

In the preceding example, we assume that the payload has a person attribute with an id, which is going to be used to correlate messages.

Likewise, for the ReleaseStrategy, you can implement your release logic as a SpEL expression and configure it in the release-strategy-expression attribute. The root object for evaluation context is the MessageGroup itself. The List of messages can be referenced by using the message property of the group within the expression.

	Note
	In releases prior to version 5.0, the root object was the collection of Message<?>, as the previous example shows:

release-strategy-expression="!messages.?[payload==5].empty"

In the preceding example, the root object of the SpEL evaluation context is the MessageGroup itself, and you are stating that, as soon as there is a message with payload of 5 in this group, the group should be released.

<aggregator input-channel="input" output-channel="output"
        send-partial-result-on-expiry="true"
        group-timeout-expression="size() ge 2 ? 10000 : -1"
        release-strategy-expression="messages[0].headers.sequenceNumber == messages[0].headers.sequenceSize"/>

The following example shows an aggregator configured with annotations:

public class Waiter {
  ...

  @Aggregator  
  public Delivery aggregatingMethod(List<OrderItem> items) {
    ...
  }

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

  @CorrelationStrategy  
  public String correlateBy(OrderItem item) {
    ...
  }
}

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:

@ServiceActivator(inputChannel = "aggregatorChannel")
@Bean
public MessageHandler aggregator(MessageGroupStore jdbcMessageGroupStore) {
     AggregatingMessageHandler aggregator =
                       new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor(),
                                                 jdbcMessageGroupStore);
     aggregator.setOutputChannel(resultsChannel());
     aggregator.setGroupTimeoutExpression(new ValueExpression<>(500L));
     aggregator.setTaskScheduler(this.taskScheduler);
     return aggregator;
}

See Section 8.4.2, “Programming Model” and Section E.6.1, “Annotations on @Bean Methods” for more information.

	Note
	Starting with version 4.2, the AggregatorFactoryBean is available to simplify Java configuration for the AggregatingMessageHandler.

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 Section 9.2, “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 the section called “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.