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 ourside of chains.

A PayloadTypeRouter will send Messages to the channel as defined by payload-type mappings.

<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 via the namespace provided by Spring Integration (see Section F.2, “Namespace Support”), which essentially simplifies configuration by combining the <router/> configuration and its corresponding implementation defined using a <bean/> element into a single and more concise configuration element. The example below demonstrates a PayloadTypeRouter configuration which is equivalent to the one above using 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>

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

1. Arbitrary value

2. Channel name

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

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

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

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

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

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

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

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

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

Note

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

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

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

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

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

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

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

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

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

Starting with version 4.1, the RecipientListRouter provides several operation to manipulate with recipients dynamically at runtime. These management operations are presented by RecipientListRouterManagement @ManagedResource. They are available using Section 8.5, “Control Bus” as well as via JMX:

<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 will have only one channel1 recipient. But after the addRecipient command above the new channel2 recipient will be 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 in some point decide to unsubscribe our interest.

Having the runtime management operation for the <recipient-list-router>, it can be configured without any <recipient> from the start. In this case the behaviour of RecipientListRouter is the same, when there is no one matching recipient for the message: if defaultOutputChannel is configured, the message will be sent there, otherwise the MessageDeliveryException is thrown.

The XPath Router is part of the XML Module. As such, please read chapter Routing XML Messages Using XPath

Spring Integration also provides a special type-based router called ErrorMessageExceptionTypeRouter for routing Error Messages (Messages whose payload is a Throwable instance). ErrorMessageExceptionTypeRouter is very 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 (e.g., payload.getClass().getSuperclass()) to find the most specific type/channel mappings, the ErrorMessageExceptionTypeRouter navigates the hierarchy of 'exception causes' (e.g., payload.getCause()) to find the most specific Throwable type/channel mappings.

Below is 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 simple 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 (extending AbstractMessageRouter):

<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 simple POJO that contains the @Router annotation (see below), or the ref may be combined with an explicit method name. Specifying a method applies the same behavior described in the @Router annotation section below.

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

Using a ref attribute is generally recommended 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 may provide an inner bean definition:

<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, as it creates an ambiguous condition, and an Exception will be thrown.

Routers and the Spring Expression Language (SpEL)

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 where you can now use SpEL to implement simple computations that previously required a custom POJO router.

	Note
	For more information about the Spring Expression Language, please refer to the respective chapter in the Spring Framework Reference Documentation at: http://static.springsource.org/spring/docs/current/spring-framework-reference/html/expressions.html

Generally a SpEL expression is evaluated and the result is mapped to a channel:

<int:router input-channel="inChannel" expression="payload.paymentType">
    <int:mapping value="CASH" channel="cashPaymentChannel"/>
    <int:mapping value="CREDIT" channel="authorizePaymentChannel"/>
    <int:mapping value="DEBIT" channel="authorizePaymentChannel"/>
</int:router>

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

<int:router input-channel="inChannel" expression="payload + 'Channel'"/>

In the above configuration the result channel will be computed by the SpEL expression which simply concatenates the value of the payload with the literal String 'Channel'.

Another value of SpEL for configuring routers is that an expression can actually return a Collection, effectively making every <router> a Recipient List Router. Whenever the expression returns multiple channel values the Message will be forwarded to each channel.

<int:router input-channel="inChannel" expression="headers.channels"/>

In the above configuration, if the Message includes a header with the name 'channels' the value of which is a List of channel names then the Message will be sent to each channel in the list. You may also find Collection Projection and Collection Selection expressions useful to select multiple channels. For further information, please see:

When using @Router to annotate a method, the method may return either a MessageChannel or String type. In the latter case, the endpoint will resolve 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 will be 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 attribute. In this case, a method annotated with @Router may include a parameter annotated with @Header which is mapped to a header value as illustrated below and documented in Section F.5, “Annotation Support”.

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

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

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

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

Note that these methods can be used for simple changes (updating a single route or adding/removing a route). However, if you want to remove one route and add another, the updates are not atomic. This means 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.

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

	Note
	For more information about Spring Integration's JMX support, please see chapter 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 AbstractMessageProducingHandlers, when an outputChannel isn't 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 will be 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 path entries:

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

In this sample we have:

Routing Slip path entries can contain MessageChannel bean names, RoutingSlipRouteStrategy bean names and also Spring expressions (SpEL). The RoutingSlipHeaderValueMessageProcessor checks each Routing Slip path entry against the BeanFactory on the first processMessage invocation. It converts entries, which aren't 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 using the requestMessage, as well as 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 using SpEL (@routingSlipRoutingPojo.get(request, reply), request.headers[myRoutingSlipChannel]) avoiding a bean definition for the RoutingSlipRouteStrategy.

Important

If a Routing Slip is involved in a distributed environment - cross-JVM application, request-reply through a Message Broker (e.g. Chapter 10, AMQP Support, Chapter 19, JMS Support), or persistence MessageStore (Section 8.3, “Message Store”) is used in the integration flow, etc., - it is recommended to not use inline expressions for the Routing Slip path. The framework (RoutingSlipHeaderValueMessageProcessor) converts them to ExpressionEvaluatingRoutingSlipRouteStrategy objects and they are used in the routingSlip message header. Since this class isn't Serializable (and it can't be, because it depends on the BeanFactory) the entire Message becomes non-serializable and in any distributed operation we end up with 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, simply add a RoutingSlipHeaderValueMessageProcessor instance to the HeaderEnricher bean definition:

@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 there is no outputChannel defined:

The EIP also defines the Process Manager pattern. This pattern can now easily be implemented 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 is a bean in the application context. This way, we can provide powerful dynamic routing logic, when there is no prediction which channel should be used; a MessageChannel can be created within the RoutingSlipRouteStrategy and returned. A FixedSubscriberChannel with an associated MessageHandler implementation is good combination for such cases. For example we can route to a Reactor Stream:

@Bean
public PollableChannel resultsChannel() {
	return new QueueChannel();
}
@Bean
public RoutingSlipRouteStrategy routeStrategy() {
	return (requestMessage, reply) -> requestMessage.getPayload() instanceof String
			? new FixedSubscriberChannel(m ->
			Streams.defer((String) m.getPayload())
					.env(this.reactorEnv)
					.get()
					.map(String::toUpperCase)
					.consume(v -> messagingTemplate().convertAndSend(resultsChannel(), v))
					.flush())
			: new FixedSubscriberChannel(m ->
			Streams.defer((Integer) m.getPayload())
					.env(this.reactorEnv)
					.get()
					.map(v -> v * 2)
					.consume(v -> messagingTemplate().convertAndSend(resultsChannel(), v))
					.flush());
}

The <filter> element is used to create a Message-selecting endpoint. In addition to "input-channel and output-channel attributes, it requires a ref. The ref may point to a MessageSelector implementation:

<int:filter input-channel="input" ref="selector" output-channel="output"/>

<bean id="selector" class="example.MessageSelectorImpl"/>

Alternatively, the method attribute can be added at which point the ref 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 will be sent to the output-channel.

<int:filter input-channel="input" output-channel="output"
    ref="exampleObject" method="someBooleanReturningMethod"/>

<bean id="exampleObject" class="example.SomeObject"/>

If the selector or adapted POJO method returns false, there are a few settings that control the handling of the rejected Message. By default (if configured like the example above), rejected Messages will be silently dropped. If rejection should instead result in an error condition, then set the throw-exception-on-rejection attribute to true:

<int:filter input-channel="input" ref="selector"
    output-channel="output" throw-exception-on-rejection="true"/> 

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

<int:filter input-channel="input" ref="selector"
    output-channel="output" discard-channel="rejectedMessages"/> 

Note

Message Filters are commonly used in conjunction with a Publish Subscribe Channel. Many filter endpoints may be subscribed to the same channel, and they decide whether or not to pass the Message to the next endpoint which could be any of the supported types (e.g. Service Activator). This provides a reactive alternative to the more proactive approach of using a Message Router with a single Point-to-Point input channel and multiple output channels.

Using a ref attribute is generally recommended if the custom filter implementation is referenced in other <filter> definitions. However if the custom filter implementation is scoped to a single <filter> element, provide an inner bean definition:

<int:filter method="someMethod" input-channel="inChannel" output-channel="outChannel">
  <beans:bean class="org.foo.MyCustomFilter"/>
</filter>

	Note
	Using both the ref attribute and an inner handler definition in the same <filter> configuration is not allowed, as it creates an ambiguous condition, and an Exception will be thrown.

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.

<int:filter input-channel="input" expression="payload.equals('nonsense')"/>

The string passed as the expression attribute will be evaluated as a SpEL expression with the Message available in the evaluation context. If it is necessary to 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 .

<int:filter input-channel="input"
            expression="payload.matches(#{filterPatterns.nonsensePattern})"/>

If the Expression itself needs to be dynamic, then an 'expression' sub-element may be used. 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 sample where the Expression could be reloaded within one minute if the underlying file had been modified. If the ExpressionSource bean is named "expressionSource", then it is not necessary to provide the source attribute on the <expression> element, but in this case it's shown for completeness.

<int:filter input-channel="input" output-channel="output">
    <int:expression key="filterPatterns.example" source="myExpressions"/>
</int:filter>

<beans:bean id="myExpressions" id="myExpressions"
    class="o.s.i.expression.ReloadableResourceBundleExpressionSource">
    <beans:property name="basename" value="config/integration/expressions"/>
    <beans:property name="cacheSeconds" value="60"/>
</beans:bean>

Then, 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) would contain a key/value pair:

filterPatterns.example=payload > 100

Note

All of these examples that use expression as an attribute or sub-element can also be applied within transformer, router, splitter, service-activator, and header-enricher elements. Of course, the semantics/role of the given component type would affect the interpretation of the evaluation result in the same way that the return value of a method-invocation would be interpreted. For example, an expression can return Strings that are to be treated as Message Channel names by a router component. However, the underlying functionality of evaluating the expression against the Message as the root object, and resolving bean names if prefixed with '@' is consistent across all of the core EIP components within Spring Integration.

A filter configured using annotations would look like this.

public class PetFilter {
    ...
    @Filter 
    public boolean dogsOnly(String input) {
        ...
    }
}

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.

Also see Section 7.7.5, “Advising Endpoints Using Annotations”.

A splitter can be configured through XML as follows:

<int:channel id="inputChannel"/>

<int:splitter id="splitter" 
  ref="splitterBean" 
  method="split" 
  input-channel="inputChannel" 
  output-channel="outputChannel" />

<int:channel id="outputChannel"/>

<beans:bean id="splitterBean" class="sample.PojoSplitter"/>

Using a ref attribute is generally recommended if the custom splitter implementation may be referenced in other <splitter> definitions. However if the custom splitter handler implementation should be scoped to a single definition of the <splitter>, configure an inner bean definition:

<int:splitter id="testSplitter" input-channel="inChannel" method="split"
                output-channel="outChannel">
  <beans:bean class="org.foo.TestSplitter"/>
</int:spliter>

	Note
	Using both a ref attribute and an inner handler definition in the same <int:splitter> configuration is not allowed, as it creates an ambiguous condition and will result in an Exception being thrown.

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, then each item will be wrapped in a Message as its payload. Each message will be sent to the designated output channel for the endpoint on which the @Splitter is defined.

@Splitter
List<LineItem> extractItems(Order order) {
    return order.getItems()
}

Also see Section 7.7.5, “Advising Endpoints Using Annotations”.

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

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

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

public abstract class AbstractAggregatingMessageGroupProcessor
              implements MessageGroupProcessor {

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

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

}

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 SequenceSizeReleaseStrategy() : releaseStrategy;
    ...
}

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

Note

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

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

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

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

The ReleaseStrategy interface is defined as follows:

public interface ReleaseStrategy {

  boolean canRelease(MessageGroup group);

}

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

public class MyReleaseStrategy {

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

public class MyReleaseStrategy {

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

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

Caution

When handling potentially large groups, it is important to 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<Foo> 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 Foo) every time the release strategy is called. Collection<?> avoids the conversion but still requires creating the new Collection. For these reasons, for large groups, it is 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 (i.e. if all messages from a sequence have arrived or if there is no sequence defined), then the group is marked as complete. Any new messages for this group will be sent to the discard channel (if defined). Setting expire-groups-upon-completion to true (default is false) removes the entire group and any new messages, with the same correlation id as the removed group, will form a new group. Partial sequences can be released by using a MessageGroupStoreReaper together with send-partial-result-on-expiry being set to true.

Important

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

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

The CorrelationStrategy interface is defined as follows:

public interface CorrelationStrategy {

  Object getCorrelationKey(Message<?> message);

}

The method returns an Object which 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 out-of-the box implementation for CorrelationStrategy, the 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 returning the value of the CORRELATION_ID header attribute. If you have a custom header name you would like to use for correlation, then simply configure that on an instance of HeaderAttributeCorrelationStrategy and provide that as a reference for the Aggregator's correlation-strategy.

Spring Integration supports the configuration of an aggregator via XML through the <aggregator/> element. Below you can see 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.JdbcMessageStore">
	<constructor-arg ref="dataSource"/>
</bean>

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

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

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

	The id of the aggregator is Optional.
	Lifecycle attribute signaling if aggregator should be started during Application Context startup. Optional (default is 'true').
	The channel from which where aggregator will receive messages. Required.
	The channel to which the aggregator will send the aggregation results. Optional (because incoming messages can specify a reply channel themselves via 'replyChannel' Message Header).
	The channel to which the aggregator will send the messages that timed out (if send-partial-result-on-expiry is false). Optional.
	A reference to a MessageGroupStore used to store groups of messages under their correlation key until they are complete. Optional, by default a volatile in-memory store.
	Order of this aggregator when more than one handle is subscribed to the same DirectChannel (use for load balancing purposes). Optional.
	Indicates that expired messages should be aggregated and sent to the 'output-channel' or 'replyChannel' once their containing MessageGroup is expired (see MessageGroupStore.expireMessageGroups(long)). One way of expiring MessageGroups is by configuring a MessageGroupStoreReaper. However MessageGroups can alternatively be expired by simply calling MessageGroupStore.expireMessageGroup(groupId). That could be accomplished via a Control Bus operation or by simply invoking that method if you have a reference to the MessageGroupStore instance. Otherwise by itself this attribute has no behavior. It only serves as an indicator of what to do (discard or send to the output/reply channel) with Messages that are still in the MessageGroup that is about to be expired. Optional. Default - 'false'. Note This attribute is more properly 'send-partial-result-on-timeout' because the group may not actually expire if expire-groups-upon-timeout is set to false.
	The timeout interval to wait when sending a reply Message to the output-channel or discard-channel. By default the send will block for one second. It is applied only if the output channel has some 'sending' limitations, e.g. QueueChannel with a fixed 'capacity'. In this case a MessageDeliveryException is thrown. The send-timeout is ignored in case of AbstractSubscribableChannel implementations. In case of group-timeout(-expression) the MessageDeliveryException from the scheduled expire task leads this task to be rescheduled. Optional.
	A reference to a bean that implements the message correlation (grouping) algorithm. The bean can be an implementation of the CorrelationStrategy interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined as well. Optional (by default, the aggregator will use the IntegrationMessageHeaderAccessor.CORRELATION_ID header) .
	A method defined on the bean referenced by correlation-strategy, that implements the correlation decision algorithm. Optional, with restrictions (requires correlation-strategy to be present).
	A SpEL expression representing the correlation strategy. Example: "headers['foo']". Only one of correlation-strategy or correlation-strategy-expression is allowed.
	A reference to a bean defined in the application context. The bean must implement the aggregation logic as described above. Optional (by default the list of aggregated Messages will become a payload of the output message).
	A method defined on the bean referenced by ref, that implements the message aggregation algorithm. Optional, depends on ref attribute being defined.
	A reference to a bean that implements the release strategy. The bean can be an implementation of the ReleaseStrategy interface or a POJO. In the latter case the release-strategy-method attribute must be defined as well. Optional (by default, the aggregator will use the IntegrationMessageHeaderAccessor.SEQUENCE_SIZE header attribute).
(16)	A method defined on the bean referenced by release-strategy, that implements the completion decision algorithm. Optional, with restrictions (requires release-strategy to be present).
(17)	A SpEL expression representing the release strategy; the root object for the expression is a Collection of Messages. Example: "size() == 5". Only one of release-strategy or release-strategy-expression is allowed.
(18)	When set to true (default false), completed groups are removed from the message store, allowing subsequent messages with the same correlation to form a new group. The default behavior is to send messages with the same correlation as a completed group to the discard-channel.
(19)	Only applies if a MessageGroupStoreReaper is configured for the <aggregator>'s MessageStore. By default, when a MessageGroupStoreReaper is configured to expire partial groups, empty groups are also removed. Empty groups exist after a group is released normally. This is to enable the detection and discarding of late-arriving messages. If you wish to expire empty groups on a longer schedule than expiring partial groups, set this property. Empty groups will then not be removed from the MessageStore until they have not been modified for at least this number of milliseconds. Note that the actual time to expire an empty group will also be affected by the reaper's timeout property and it could be as much as this value plus the timeout.
(20)	A reference to a org.springframework.integration.util.LockRegistry bean; used to obtain a Lock based on the groupId for concurrent operations on the MessageGroup. By default, an internal DefaultLockRegistry is used. Use of a distributed LockRegistry, such as the RedisLockRegistry, ensures only one instance of the aggregator will operate on a group concurrently. See Section 23.11, “Redis Lock Registry” for more information.
(21)	A timeout in milliseconds to force the MessageGroup complete, when the ReleaseStrategy doesn't release the group when the current Message arrives. This attribute provides a built-in Time-base Release Strategy for the aggregator, when there is a need to emit a partial result (or discard the group), if a new Message does not arrive for the MessageGroup within the timeout. When a new Message arrives at the aggregator, any existing ScheduledFuture<?> for its MessageGroup is canceled. If the ReleaseStrategy returns false (don't release) and the groupTimeout > 0 a new task will be scheduled to expire the group. Setting this attribute to zero is not advised because it will effectively disable the aggregator because every message group will be immediately completed. It is possible, however to conditionally set it to zero using an expression; see group-timeout-expression for information. The action taken during the completion depends on the ReleaseStrategy and the send-partial-group-on-expiry attribute. See the section called “Aggregator and Group Timeout” for more information. Mutually exclusive with 'group-timeout-expression' attribute.
(22)	The SpEL expression that evaluates to a groupTimeout with the MessageGroup as the #root evaluation context object. Used for scheduling the MessageGroup to be forced complete. If the expression evaluates to null or < 0, the completion is not scheduled. If it evaluates to zero, the group is completed immediately on the current thread. In effect, this provides a dynamic group-timeout property. See group-timeout for more information. Mutually exclusive with 'group-timeout' attribute.
(23)	When a group is completed due to a timeout (or by a MessageGroupStoreReaper), the group is expired (completely removed) by default. Late arriving messages will start a new group. Set this to false to complete the group but have its metadata remain so that late arriving messages will be discarded. Empty groups can be expired later using a MessageGroupStoreReaper together with the empty-group-min-timeout attribute. Default: 'true'.
(24)	A TaskScheduler bean reference to schedule the MessageGroup to be forced complete if no new message arrives for the MessageGroup within the groupTimeout. If not provided, the default scheduler taskScheduler, registered in the ApplicationContext (ThreadPoolTaskScheduler) will be used. This attribute does not apply if group-timeout or group-timeout-expression is not specified.
(25)	Since version 4.1. Allows a transaction to be started for the forceComplete operation. It is initiated from a group-timeout(-expression) or by a MessageGroupStoreReaper and is not applied to the normal add/release/discard operations. Only this sub-element or <expire-advice-chain/> is allowed.
(26)	Since version 4.1. Allows the configuration of any Advice for the forceComplete operation. It is initiated from a group-timeout(-expression) or by a MessageGroupStoreReaper and is not applied to the normal add/release/discard operations. Only this sub-element or <expire-transactional/> is allowed. A transaction Advice can also be configured here using the Spring tx namespace.

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 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 now control this behavior using expire-groups-upon-timeout; this 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, then 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).

Using a ref attribute is generally recommended 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:

<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 will be thrown.

An example implementation of the aggregator bean looks as follows:

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 example above may 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, the release strategy method and the aggregator method can be combined in a single bean.

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

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

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

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

Aggregators and Spring Expression Language (SpEL)

Since Spring Integration 2.0, the various strategies (correlation, release, and aggregation) may be handled with SpEL which is recommended if the logic behind such release strategy is relatively simple. Let's say you have a legacy component that was designed to receive an array of objects. We know that the default release strategy will assemble all aggregated messages in the List. So now we have two problems. First we need to extract individual messages from the list, and then we need to extract the payload of each message and assemble the array of objects (see code below).

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.

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

In the above configuration we are using a Collection Projection expression to assemble a new collection from the payloads of all messages in the list and then transforming it to an Array, thus achieving the same result as the java code above.

The same expression-based approach can be applied when dealing with custom Release and Correlation strategies.

Instead of defining a bean for a custom CorrelationStrategy via the correlation-strategy attribute, you can implement your simple correlation logic via a SpEL expression and configure it via the correlation-strategy-expression attribute.

For example:

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

In the above example it is assumed that the payload has an attribute person 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 via the release-strategy-expression attribute. The only difference is that since ReleaseStrategy is passed the List of Messages, the root object in the SpEL evaluation context is the List itself. That List can be referenced as #this within the expression.

For example:

release-strategy-expression="#this.size() gt 5"

In this example the root object of the SpEL Evaluation Context is the MessageGroup itself, and you are simply stating that as soon as there are more than 5 messages in this group, it 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="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>

An aggregator configured using annotations would look like this.

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.