Routers are a crucial element in many messaging architectures. They consume Messages from a Message Channel and forward each consumed message to one or more different Message Channel depending on a set of conditions.
Spring Integration provides the following routers out-of-the-box:
Router implementations share many configuration parameters. Yet, certain differences exist between routers. Furthermore, the availability of configuration parameters depends on whether Routers are used inside or outside of a chain. In order to provide a quick overview, all available attributes are listed in the 2 tables below.
Table 5.1. Routers Outside of a Chain
Attribute | router | header value router | xpath router | payload type router | recipient list router | exception type router |
---|---|---|---|---|---|---|
apply-sequence | ||||||
default-output-channel | ||||||
resolution-required | ||||||
ignore-send-failures | ||||||
timeout | ||||||
id | ||||||
auto-startup | ||||||
input-channel | ||||||
order | ||||||
method | ||||||
ref | ||||||
expression | ||||||
header-name | ||||||
evaluate-as-string | ||||||
xpath-expression-ref | ||||||
converter |
Table 5.2. Routers Inside of a Chain
Attribute | router | header value router | xpath router | payload type router | recipient list router | exception type router |
---|---|---|---|---|---|---|
apply-sequence | ||||||
default-output-channel | ||||||
resolution-required | ||||||
ignore-send-failures | ||||||
timeout | ||||||
id | ||||||
auto-startup | ||||||
input-channel | ||||||
order | ||||||
method | ||||||
ref | ||||||
expression | ||||||
header-name | ||||||
evaluate-as-string | ||||||
xpath-expression-ref | ||||||
converter |
Important | |
---|---|
Router parameters have been more standardized across all router implementations with Spring Integration 2.1. Consequently, there are a few minor changes that leave the possibility of breaking older Spring Integration based applications.
Since Spring Integration 2.1 the
Prior to these changes, the
If you do desire to drop messages silently simply set |
The following parameters are valid for all routers inside and outside of chains.
This attribute specifies whether sequence number and size headers should be added to each Message. This optional attribute defaults to false.
If set, this attribute provides a reference to the
channel, where Messages should be sent, if channel
resolution fails to return any channels. If no default
output channel is provided, the router will throw an Exception.
If you would like to silently drop those messages instead,
add the nullChannel
as the default output
channel attribute value.
Note | |
---|---|
A Message will only be sent to the default-output-channel if
resolution-required is false and the channel is not resolved.
|
If true this attribute specifies
that channel names must always be successfully resolved
to channel instances that exist. If set to
true, a MessagingException
will be raised, in case the channel cannot be resolved.
Setting this attribute to false,
will cause any unresovable channels to be ignored. This
optional attribute will, if not
explicitly set, default to true.
Note | |
---|---|
A Message will only be sent to the default-output-channel , if
specified, when resolution-required is false and the channel
is not resolved.
|
If set to true, failures to send to a
message channel will be ignored. If set to false,
a MessageDeliveryException
will
be thrown instead, and if the router resolves more than
one channel, any subsequent channels will not receive
the message.
The exact behavior of this attribute depends on the type
of the Channel
messages are sent to. For
example, when using direct channels (single threaded),
send-failures can be caused by exceptions thrown by components
much further down-stream. However, when sending
messages to a simple queue channel (asynchronous) the
likelihood of an exception to be thrown is rather remote.
Note | |
---|---|
While most routers will route to a single channel, they
are allowed to return more than one channel name. The
recipient-list-router , for instance, does
exactly that.
If you set this attribute to true
on a router that only routes to a single channel, any
caused exception is simply swallowed, which usually makes
little sense to do. In that case it would be better to
catch the exception in an error flow at the flow entry
point.
Therefore, setting the ignore-send-failures
attribute to true usually makes
more sense when the router implementation returns more
than one channel name, because the other channel(s)
following the one that fails would still receive the Message.
|
This attribute defaults to false.
The timeout
attribute specifies the maximum
amount of time in milliseconds to wait, when
sending Messages to the target Message Channels. By default
the send operation will block indefinitely.
The following parameters are valid only across all top-level routers that are ourside of chains.
Identifies the underlying Spring bean definition which in case of Routers is an instance of EventDrivenConsumer or PollingConsumer depending on whether the Router's input-channel is a SubscribableChannel or PollableChannel, respectively. This is an optional attribute.
This Lifecycle
attribute signaled if this
component should be started during startup of the Application
Context. This optional
attribute defaults to true.
The receiving Message channel of this endpoint.
This attribute defines the order for invocation when this endpoint is connected as a subscriber to a channel. This is particularly relevant when that channel is using a failover dispatching strategy. It has no effect when this endpoint itself is a Polling Consumer for a channel with a queue.
Since content-based routing often requires some domain-specific logic, most use-cases will require Spring Integration's options for delegating to POJOs using the XML namespace support and/or Annotations. Both of these are discussed below, but first we present a couple implementations that are available out-of-the-box since they fulfill common requirements.
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="channelIdentifierMap"> <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
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 |
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.
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 |
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)
Note | |
---|---|
For routing of XML-based Messages, including XPath support, see Chapter 32, XML Support - Dealing with XML Payloads. |
So as you can see, Spring Integration provides quite a few different router configurations for common
content-based routing use cases as well as the option of implementing custom routers as POJOs.
For example PayloadTypeRouter
provides a simple way to configure a router which computes channels
based on the payload type
of the incoming Message while HeaderValueRouter
provides the
same convenience in configuring a router which computes channels
by evaluating the value
of a particular Message Header. There are also expression-based (SpEL) routers where the channel
is determined based on evaluating an expression. Thus, these type of routers exhibit some dynamic characteristics.
However these routers all require static configuration. Even in the case of expression-based routers, the expression itself is defined as part of the router configuration which means that the same expression operating on the same value will always result in the computation of the same channel. This is acceptable in most cases since such routes are well defined and therefore predictable. But there are times when we need to change router configurations dynamically so message flows may be routed to a different channel.
Example:
You might want to bring down some part of your system for maintenance and temporarily re-reroute
messages to a different message flow. Or you may want to introduce more granularity to your message flow by adding another
route to handle a more concrete type of java.lang.Number
(in the case of PayloadTypeRouter
).
Unfortunately with static router configuration to accomplish this, you would have to bring down your entire application, change the configuration of the router (change routes) and bring it back up. This is obviously not the solution.
The Dynamic Router pattern describes the mechanisms by which one can change/configure routers dynamically without bringing down the system or individual routers.
Before we get into the specifics of how this is accomplished in Spring Integration, let's quickly summarize the typical flow of the router, which consists of 3 simple steps:
Step 1 - Compute channel identifier
which is a value calculated by the
router once it receives the Message. Typically it is a String
or and instance of the actual
MessageChannel
.
Step 2 - Resolve channel identifier
to channel name
. We'll describe
specifics of this process in a moment.
Step 3 - Resolve channel name
to the actual MessageChannel
There is not much that can be done with regard to dynamic routing if Step 1 results in the actual instance of the
MessageChannel
, simply because the MessageChannel
is the final product of any
router's job. However, if Step 1 results in a channel identifier
that is not an instance of MessageChannel
,
then there are quite a few possibilities to influence the process of deriving the Message Channel
.
Lets look at couple of the examples in the context of the 3 steps mentioned above:
Payload Type Router
<int:payload-type-router input-channel="routingChannel"> <int:mapping type="java.lang.String" channel="channel1" /> <int:mapping type="java.lang.Integer" channel="channel2" /> </int:payload-type-router>
Within the context of the Payload Type Router the 3 steps mentioned above would be realized as:
Step 1 - Compute channel identifier
which is the fully qualified name of the payload type
(e.g., java.lang.String).
Step 2 - Resolve channel identifier
to channel name
where
the result of the previous step is used to select the appropriate value from the payload type mapping
defined via mapping
element.
Step 3 - Resolve channel name
to the actual instance of the
MessageChannel
as a reference to a bean within the Application Context
(which is hopefully a MessageChannel
) identified by the result of the
previous step.
In other words, each step feeds the next step until the process completes.
Header Value Router
<int:header-value-router input-channel="inputChannel" header-name="testHeader"> <int:mapping value="foo" channel="fooChannel" /> <int:mapping value="bar" channel="barChannel" /> </int:header-value-router>
Similar to the PayloadTypeRouter:
Step 1 - Compute channel identifier
which is the value of the header identified by the
header-name
attribute.
Step 2 - Resolve channel identifier
to channel name
where
the result of the previous step is used to select the appropriate value from the general mapping
defined via mapping
element.
Step 3 - Resolve channel name
to the actual instance of the
MessageChannel
as a reference to a bean within the Application Context
(which is hopefully a MessageChannel
) identified by the result of the
previous step.
The above two configurations of two different router types look almost identical.
However if we look at the alternate configuration of the HeaderValueRouter
we clearly see that
there is no mapping
sub element:
<int:header-value-router input-channel="inputChannel" header-name="testHeader">
But the configuration is still perfectly valid. So the natural question is what about the mapping in the Step 2?
What this means is that Step 2 is now an optional step. If mapping
is not defined then the channel identifier
value computed in Step 1 will automatically be treated as the channel name
, which will now be resolved to the
actual MessageChannel
as in Step 3. What it also means is that Step 2 is one of the key steps to
provide dynamic characteristics to the routers, since it introduces a process which
allows you to change the way 'channel identifier' resolves to 'channel name',
thus influencing the process of determining the final instance of the MessageChannel
from the initial
channel identifier
.
For Example:
In the above configuration let's assume that the testHeader
value is 'kermit' which is now a channel identifier
(Step 1). Since there is no mapping in this router, resolving this channel identifier
to a channel name
(Step 2) is impossible and this channel identifier
is now treated as channel name
. However what if
there was a mapping but for a different value? The end result would still be the same and that is:
if a new value cannot be determined through the process of resolving the 'channel identifier' to a 'channel name',
such 'channel identifier' becomes 'channel name'.
So all that is left is for Step 3 to resolve the channel name
('kermit') to an actual instance of the
MessageChannel
identified by this name. That basically involves a bean lookup
for the name provided. So now all messages which contain the header/value pair as testHeader=kermit
are going to be routed to a MessageChannel
whose bean name (id) is 'kermit'.
But what if you want to route these messages to the 'simpson' channel? Obviously changing a static configuration will work,
but will also require bringing your system down. However if you had access to the channel identifier
map, then you
could just introduce a new mapping where the header/value pair is now kermit=simpson
, thus allowing Step 2 to treat
'kermit' as a channel identifier
while resolving it to 'simpson' as the channel name
.
The same obviously applies for PayloadTypeRouter
, where you can now remap or remove a particular payload type
mapping. In fact, it applies to every other router, including expression-based routers, since their computed values
will now have a chance to go through Step 2 to be additionally resolved to the actual channel name
.
In Spring Integration 2.0 the router hierarchy underwent significant refactoring, so that now any router that is a subclass of the
AbstractMessageRouter
(which includes all framework defined routers) is a Dynamic Router simply because the
channelIdentiferMap
is defined at the AbstractMessageRouter
level. That map's setter method is
exposed as a public method along with 'setChannelMapping' and 'removeChannelMapping' methods. These allow you to change/add/remove
router mappings at runtime as long as you have a reference to the router itself. It also means that you could expose these same
configuration options via JMX (see Section 8.1, “JMX Support”) or the Spring Integration ControlBus (see Section 8.5, “Control Bus”) functionality.
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:
public void setChannelMapping(String channelIdentifier, String channelName)
-
will allow you to add a new or modify an existing mapping between channel identifier
and channel name
public void removeChannelMapping(String channelIdentifier)
-
will allow you to remove a particular channel mapping, thus disconnecting the relationship between
channel identifier
and channel name
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 betweent the updates. Starting with version 4.0, you can now use the control bus to update the entire routing table atomically.
public Map<String, String>getChannelMappings()
returns the current
mappings.
public void replaceChannelMappings(Properties channelMappings)
updates the mappings.
Notice that the parameter is a properties object; this allows the use of the inbuilt
StringToPropertiesConverter
by a control bus command, for example:
"@'router.handler'.replaceChannelMappings('foo=qux \n baz=bar')"
- note that each mapping is separted by a newline character (\n
). For programmatic
changes to the map, it is recommended that the setChannelMappings
method
is used instead, for type-safety. Any non-String keys or values passed into
replaceChannelMappings
are ingnored.
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 suppor, please see chapter JMX Support. |
Message Filters are used to decide whether a Message should be passed along or dropped based on some criteria such as a Message Header value or Message content itself. Therefore, a Message Filter is similar to a router, except that for each Message received from the filter's input channel, that same Message may or may not be sent to the filter's output channel. Unlike the router, it makes no decision regarding which Message Channel to send the Message to but only decides whether to send.
Note | |
---|---|
As you will see momentarily, the Filter also supports a discard channel, so in certain cases it can play the role of a very simple router (or "switch") based on a boolean condition. |
In Spring Integration, a Message Filter may be configured as a Message Endpoint that delegates to an
implementation of the MessageSelector
interface. That interface is itself quite
simple:
public interface MessageSelector { boolean accept(Message<?> message); }
The MessageFilter
constructor accepts a selector instance:
MessageFilter filter = new MessageFilter(someSelector);
In combination with the namespace and SpEL, very powerful filters can be configured with very little java code.
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 |
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) { ... } }
An annotation indicating that this method shall be used as a filter. Must be specified if this class will be used as a filter. |
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”.
The Splitter is a component whose role is to partition a message in several parts, and send the resulting messages to be processed independently. Very often, they are upstream producers in a pipeline that includes an Aggregator.
The API for performing splitting consists of one base class,
AbstractMessageSplitter
, which is a
MessageHandler
implementation,
encapsulating features which are common to splitters, such as filling in
the appropriate message headers CORRELATION_ID, SEQUENCE_SIZE, and
SEQUENCE_NUMBER on the messages that are produced. This enables tracking
down the messages and the results of their processing (in a typical
scenario, these headers would be copied over to the messages that are
produced by the various transforming endpoints), and use them, for
example, in a
Composed Message Processor scenario.
An excerpt from AbstractMessageSplitter
can be seen below:
public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageConsumer { ... protected abstract Object splitMessage(Message<?> message); }
To implement a specific Splitter in an application,
extend AbstractMessageSplitter
and implement the splitMessage
method,
which contains logic for splitting the messages. The return
value may be one of the following:
a Collection
(or subclass thereof) or an array of
Message
objects -
in this case the messages will be sent as such (after the
CORRELATION_ID, SEQUENCE_SIZE and SEQUENCE_NUMBER are populated).
Using this approach gives more control to the developer, for example
for populating custom message headers as part of the splitting
process.
a Collection
(or subclass thereof) or an array of
non-Message objects - works like the prior case, except that each collection
element will be used as a Message payload. Using this approach allows
developers to focus on the domain objects without having to consider
the Messaging system and produces code that is easier to test.
a Message
or non-Message object
(but not a Collection or an Array) - it works like the previous cases,
except a single message will be sent out.
In Spring Integration, any POJO can implement the splitting
algorithm, provided that it defines a method that accepts a single
argument and has a return value. In this case, the return value of the
method will be interpreted as described above. The input argument might
either be a Message
or a simple POJO.
In the latter case, the splitter will receive the payload of the incoming message.
Since this decouples the code from the Spring Integration API and will typically be easier
to test, it is the recommended approach.
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"/>
The id of the splitter is optional. | |
A reference to a bean defined in the application context. The
bean must implement the splitting logic as described in the section
above .Optional.
If reference to a bean is not provided, then it is assumed that the payload
of the Message that arrived on the | |
The method (defined on the bean specified above) that implements the splitting logic. Optional. | |
The input channel of the splitter. Required. | |
The channel to which the splitter will send the results of splitting the incoming message. Optional (because incoming messages can specify a reply channel themselves). |
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 |
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”.
Basically a mirror-image of the Splitter, the Aggregator is a type of Message Handler that receives multiple Messages and combines them into a single Message. In fact, an Aggregator is often a downstream consumer in a pipeline that includes a Splitter.
Technically, the Aggregator is more complex than a Splitter, because
it is stateful as it must hold the Messages to be aggregated and determine
when the complete group of Messages is ready to be aggregated. In order to do
this it requires a MessageStore
.
The Aggregator combines a group of related messages, by correlating and storing them, until the group is deemed complete. At that point, the Aggregator will create a single message by processing the whole group, and will send the aggregated message as output.
Implementing an Aggregator requires providing the logic to perform the aggregation (i.e., the creation of a single message from many). Two related concepts are correlation and release.
Correlation determines how messages are grouped for aggregation.
In Spring Integration correlation is done by default based on the
MessageHeaders.CORRELATION_ID
message
header. Messages with the same MessageHeaders.CORRELATION_ID
will be grouped
together. However, the correlation strategy may be customized to allow
other ways of specifying how the messages should be grouped together by
implementing a CorrelationStrategy
(see below).
To determine the point at which a group of messages is ready to be processed, a
ReleaseStrategy
is consulted.
The default release strategy for the Aggregator will release a group when all
messages included in a sequence are present, based on the
MessageHeaders.SEQUENCE_SIZE
header.
This default strategy may be overridden by providing a reference to a
custom ReleaseStrategy
implementation.
The Aggregation API consists of a number of classes:
The interface MessageGroupProcessor
, and
its subclasses:
MethodInvokingAggregatingMessageGroupProcessor
and
ExpressionEvaluatingMessageGroupProcessor
The ReleaseStrategy
interface and its default
implementation SequenceSizeReleaseStrategy
The CorrelationStrategy
interface and its default
implementation HeaderAttributeCorrelationStrategy
The AggregatingMessageHandler
(subclass of
AbstractCorrelatingMessageHandler
) is a
MessageHandler
implementation, encapsulating the common
functionalities of an Aggregator (and other correlating use cases),
which are:
correlating messages into a group to be aggregated
maintaining those messages in a MessageStore
until the group
can be released
deciding when the group can be released
aggregating the released group into a single message
recognizing and responding to an expired group
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); }The
CorrelationStrategy
is owned by the
AbstractCorrelatingMessageHandler
and it has a default value based on the MessageHeaders.CORRELATION_ID
message header:
public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { ... this.correlationStrategy = correlationStrategy == null ? new HeaderAttributeCorrelationStrategy(MessageHeaders.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
|
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:
if the argument is a java.util.List<T>
, and the
parameter type T is assignable to Message
, then the whole list of
messages accumulated for aggregation will be sent to the aggregator
if the argument is a non-parameterized java.util.List
or the
parameter type is not assignable to Message
, then the method will
receive the payloads of the accumulated messages
if the return type is not assignable to Message
, then it will
be treated as the payload for a Message that will be created
automatically by the framework.
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:
if the argument is a java.util.List<T>
, and the
parameter type T is assignable to Message
, then the whole list of
messages accumulated in the group will be sent to the method
if the argument is a non-parametrized java.util.List
or the
parameter type is not assignable to Message
, then the method will
receive the payloads of the accumulated messages
the method must return true if the message group is ready for aggregation, and false otherwise.
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
For these reasons, for large groups, it is recommended that you implement
|
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) scheduler="taskScheduler" /> (23) <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 0ptional. | |
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 | |
A reference to a | |
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 | |
The timeout interval for sending the aggregated messages to the output or reply channel. Optional. | |
A reference to a bean that implements the message correlation (grouping)
algorithm. The bean can be an implementation of the | |
A method defined on the bean referenced by
| |
A SpEL expression representing the correlation strategy.
Example: | |
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 | |
A reference to a bean that implements the release strategy.
The bean can be an implementation of the | |
A method defined on the bean referenced by
| |
A SpEL expression representing the release strategy; the root object for the
expression is a | |
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. | |
Only applies if a | |
A reference to a | |
A timeout in milliseconds to force the | |
The SpEL expression that evaluates to a | |
A |
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 |
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.
Starting with version 4.0, two new mutually exclusive attributes have been introduced:
group-timeout
and group-timeout-expression
(see the description above). There are some
cases where it is needed to emit the aggregator result (or discard the group) after a timeout
if the ReleaseStrategy
doesn't release when the current Message arrives.
For this purpose the groupTimeout
option allows scheduling the MessageGroup
to
be forced complete:
<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"/>
With this example, the normal release will be possible if the aggregator receives the last message
in sequence as defined by the release-strategy-expression
. If that specific message does not arrive,
the groupTimeout
will force the group complete after 10 seconds as long as the group contains at least 2 Messages.
The results of forcing the group complete depends on the ReleaseStrategy
and
the send-partial-result-on-expiry
.
First, the release strategy is again consulted to see if a normal release is to be made - while the
group won't have changed, the ReleaseStrategy
can decide to release the group at this time. If the
release strategy still does not release the group, it will be expired.
If send-partial-result-on-expiry
is true
,
existing messages in the (partial) MessageGroup
will be released as a normal aggregator reply Message to the
output-channel
, otherwise it will be discarded.
There is a difference between groupTimeout
behavior and MessageGroupStoreReaper
(see Section 5.4.4, “Configuring an Aggregator”). The reaper initiates forced completion for all MessageGroup
s
in the MessageGroupStore
periodically. The groupTimeout
does it for each MessageGroup
individually, if a new Message doesn't arrive during the groupTimeout
. Also, the reaper can be used to
remove empty groups (empty groups are retained in order to discard late messages,
if expire-groups-upon-completion
is false).
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.
Aggregator (and some other patterns in Spring Integration) is a
stateful pattern that requires decisions to be made based on a group of
messages that have arrived over a period of time, all with the same
correlation key. The design of the interfaces in the stateful patterns
(e.g. ReleaseStrategy
) is driven by the principle
that the components (whether defined by the framework or a user) should be able to remain stateless.
All state is carried by the MessageGroup
and its
management is delegated to the
MessageGroupStore
.
public interface MessageGroupStore { int getMessageCountForAllMessageGroups(); int getMarkedMessageCountForAllMessageGroups(); int getMessageGroupCount(); MessageGroup getMessageGroup(Object groupId); MessageGroup addMessageToGroup(Object groupId, Message<?> message); MessageGroup markMessageGroup(MessageGroup group); MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove); MessageGroup markMessageFromGroup(Object key, Message<?> messageToMark); void removeMessageGroup(Object groupId); void registerMessageGroupExpiryCallback(MessageGroupCallback callback); int expireMessageGroups(long timeout); }
For more information please refer to the JavaDoc.
The MessageGroupStore
accumulates state
information in MessageGroups
while waiting for
a release strategy to be triggered, and that event might not ever happen.
So to prevent stale messages from lingering, and for volatile stores to
provide a hook for cleaning up when the application shuts down, the
MessageGroupStore
allows the user to register
callbacks to apply to its MessageGroups
when they
expire. The interface is very straightforward:
public interface MessageGroupCallback { void execute(MessageGroupStore messageGroupStore, MessageGroup group); }
The callback has direct access to the store and the message group so it can manage the persistent state (e.g. by removing the group from the store entirely).
The MessageGroupStore
maintains a list of these callbacks
which it applies, on demand, to all messages whose timestamp is earlier than a time
supplied as a parameter (see the registerMessageGroupExpiryCallback(..)
and expireMessageGroups(..)
methods above).
The expireMessageGroups
method can be called with a timeout value.
Any message older than the current time minus this value will be expired,
and have the callbacks applied. Thus it is the user of the store that
defines what is meant by message group "expiry".
As a convenience for users, Spring Integration provides a wrapper
for the message expiry in the form of a
MessageGroupStoreReaper
:
<bean id="reaper" class="org...MessageGroupStoreReaper"> <property name="messageGroupStore" ref="messageStore"/> <property name="timeout" value="30000"/> </bean> <task:scheduled-tasks scheduler="scheduler"> <task:scheduled ref="reaper" method="run" fixed-rate="10000"/> </task:scheduled-tasks>
The reaper is a Runnable
, and all that is happening
in the example above is that the message group store's expire method is being called
once every 10 seconds. The timeout itself is 30 seconds.
Note | |
---|---|
It is important to understand that the 'timeout' property of the MessageGroupStoreReaper is an
approximate value and is impacted by the the rate of the task scheduler since this property will
only be checked on the next scheduled execution of the MessageGroupStoreReaper task. For example if
the timeout is set for 10 min, but the MessageGroupStoreReaper task is scheduled to run every 60 min
and the last execution of the MessageGroupStoreReaper task happened 1 min before the timeout, the
MessageGroup will not expire for the next 59 min. So it is recommended to set the rate at least equal to the value of the timeout or shorter.
|
In addition to the reaper, the expiry callbacks are invoked when the application
shuts down via a lifecycle callback in the AbstractCorrelatingMessageHandler
.
The AbstractCorrelatingMessageHandler
registers its
own expiry callback, and this is the link with the boolean flag
send-partial-result-on-expiry
in the XML configuration of the
aggregator. If the flag is set to true, then when the expiry callback is
invoked, any unmarked messages in groups that are not yet released can
be sent on to the output channel.
Important | |
---|---|
When using a Some For more information about |
Related to the Aggregator, albeit different from a functional standpoint, is the Resequencer.
The Resequencer works in a similar way to the Aggregator, in the sense that it uses the CORRELATION_ID to store messages in groups, the difference being that the Resequencer does not process the messages in any way. It simply releases them in the order of their SEQUENCE_NUMBER header values.
With respect to that, the user might opt to release all messages at once (after the whole sequence, according to the SEQUENCE_SIZE, has been released), or as soon as a valid sequence is available.
Configuring a resequencer requires only including the appropriate element in XML.
A sample resequencer configuration is shown below.
<int:channel id="inputChannel"/> <int:channel id="outputChannel"/> <int:resequencer id="completelyDefinedResequencer" input-channel="inputChannel" output-channel="outputChannel" discard-channel="discardChannel" release-partial-sequences="true" message-store="messageStore" send-partial-result-on-expiry="true" send-timeout="86420000" correlation-strategy="correlationStrategyBean" correlation-strategy-method="correlate" correlation-strategy-expression="headers['foo']" release-strategy="releaseStrategyBean" release-strategy-method="release" release-strategy-expression="size() == 10" empty-group-min-timeout="60000" lock-registry="lockRegistry" (16) group-timeout="60000" (17) group-timeout-expression="size() ge 2 ? 100 : -1" (18) scheduler="taskScheduler" /> (19)
The id of the resequencer is optional. | |
The input channel of the resequencer. Required. | |
The channel to which the resequencer will send the reordered messages. Optional. | |
The channel to which the resequencer will send the messages that
timed out (if | |
Whether to send out ordered sequences as soon as they are available, or only after the whole message group arrives. Optional (false by default). | |
A reference to a | |
Whether, upon the expiration of the group, the ordered group should be sent out (even if some of the messages are missing). Optional (false by default). See Section 5.4.5, “Managing State in an Aggregator: MessageGroupStore”. | |
The timeout for sending out messages. Optional. | |
A reference to a bean that implements the message correlation (grouping)
algorithm. The bean can be an implementation of the | |
A method defined on the bean referenced by
| |
A SpEL expression representing the correlation strategy.
Example: | |
A reference to a bean that implements the release strategy.
The bean can be an implementation of the | |
A method defined on the bean referenced by
| |
A SpEL expression representing the release strategy; the root object for the
expression is a | |
Only applies if a | |
See the section called “Configuring an Aggregator with XML”. | |
See the section called “Configuring an Aggregator with XML”. | |
See the section called “Configuring an Aggregator with XML”. | |
See the section called “Configuring an Aggregator with XML”. |
Note | |
---|---|
Since there is no custom behavior to be implemented in Java classes for resequencers, there is no annotation support for it. |
The MessageHandlerChain
is an implementation of
MessageHandler
that can be configured as a single Message Endpoint while
actually delegating to a chain of other handlers, such as Filters, Transformers, Splitters, and so on.
This can lead to a much simpler configuration when several handlers need to be connected in a fixed, linear
progression. For example, it is fairly common to provide a Transformer before other components. Similarly, when
providing a Filter before some other component in a chain, you are essentially creating a
Selective Consumer. In either case, the
chain only requires a single input-channel
and a single output-channel
eliminating
the need to define channels for each individual component.
Tip | |
---|---|
Spring Integration's Filter provides a boolean property throwExceptionOnRejection .
When providing multiple Selective Consumers on the same point-to-point channel with different acceptance criteria,
this value should be set to 'true' (the default is false) so that the dispatcher will know that the Message was
rejected and as a result will attempt to pass the Message on to other subscribers. If the Exception were not
thrown, then it would appear to the dispatcher as if the Message had been passed on successfully even though
the Filter had dropped the Message to prevent further processing. If you do indeed want
to "drop" the Messages, then the Filter's 'discard-channel' might be useful since it does give you a chance
to perform some operation with the dropped message (e.g. send to a JMS queue or simply write to a log).
|
The handler chain simplifies configuration while internally maintaining the same degree of loose coupling between components, and it is trivial to modify the configuration if at some point a non-linear arrangement is required.
Internally, the chain will be expanded into a linear setup of the listed endpoints, separated by anonymous channels. The reply channel header will not be taken into account within the chain: only after the last handler is invoked will the resulting message be forwarded on to the reply channel or the chain's output channel. Because of this setup all handlers except the last required to implement the MessageProducer interface (which provides a 'setOutputChannel()' method). The last handler only needs an output channel if the outputChannel on the MessageHandlerChain is set.
Note | |
---|---|
As with other endpoints, the |
In most cases there is no need to implement MessageHandlers yourself. The next section will focus on namespace
support for the chain element. Most Spring Integration endpoints, like Service Activators and Transformers, are
suitable for use within a MessageHandlerChain
.
The <chain> element provides an input-channel
attribute, and if the last element in the chain is capable
of producing reply messages (optional), it also supports an output-channel
attribute. The sub-elements are then
filters, transformers, splitters, and service-activators. The last element may also be a router or an outbound-channel-adapter.
<int:chain input-channel="input" output-channel="output"> <int:filter ref="someSelector" throw-exception-on-rejection="true"/> <int:header-enricher> <int:header name="foo" value="bar"/> </int:header-enricher> <int:service-activator ref="someService" method="someMethod"/> </int:chain>
The <header-enricher> element used in the above example will set a message header named "foo" with a value
of "bar" on the message. A header enricher is a specialization of Transformer
that touches only header values. You could obtain the same result by implementing a MessageHandler that did the
header modifications and wiring that as a bean, but the header-enricher is obviously a simpler option.
The <chain> can be configured as the last 'black-box' consumer of the message flow. For this solution it is enough to put at the end of the <chain> some <outbound-channel-adapter>:
<int:chain input-channel="input"> <si-xml:marshalling-transformer marshaller="marshaller" result-type="StringResult" /> <int:service-activator ref="someService" method="someMethod"/> <int:header-enricher> <int:header name="foo" value="bar"/> </int:header-enricher> <int:logging-channel-adapter level="INFO" log-full-message="true"/> </int:chain>
Disallowed Attributes and Elements
It is important to note that certain attributes, such as order and input-channel are not allowed to be specified on components used within a chain. The same is true for the poller sub-element.
Important | |
---|---|
For the Spring Integration core components, the XML Schema itself will enforce some of these constraints. However, for non-core components or your own custom components, these constraints are enforced by the XML namespace parser, not by the XML Schema.
These XML namespace parser constraints were added with
Spring Integration 2.2. The XML namespace parser
will throw an |
'id' Attribute
Beginning with Spring Integration 3.0, if a chain element is given an id, the
bean name for the element is a combination of the chain's id and the id
of the element itself. Elements without an id are not registered
as beans, but they are given componentName
s that include the chain id. For example:
<int:chain id="fooChain" input-channel="input"> <int:service-activator id="fooService" ref="someService" method="someMethod"/> <int:object-to-json-transformer/> </int:chain>
<chain>
root element has an id 'fooChain'. So, the
AbstractEndpoint
implementation (PollingConsumer
or
EventDrivenConsumer
, depending on the input-channel type)
bean takes this value as it's bean name.
MessageHandlerChain
bean acquires a bean alias 'fooChain.handler', which allows
direct access to this bean from the BeanFactory
.
<service-activator>
is not a fully-fledged Messaging Endpoint (PollingConsumer
or EventDrivenConsumer
) - it is simply a
MessageHandler
within the <chain>
. In this case,
the bean name registered with the BeanFactory
is 'fooChain$child.fooService.handler'.
ServiceActivatingHandler
takes the
same value, but without the '.handler' suffix - 'fooChain$child.fooService'.
<chain>
sub-component, <object-to-json-transformer>
, doesn't have
an id attribute. Its componentName is based on its
position in the <chain>
. In this case, it is 'fooChain$child#1'.
(The final element of the name is the order within the chain, beginning with '#0').
Note, this transformer isn't registered as a bean within the application context,
so, it doesn't get a beanName, however its componentName has
a value which is useful for logging etc.
The id attribute for <chain>
elements allows them to be eligible for
JMX export and they are trackable via Message History.
They can also be accessed from the BeanFactory
using the appropriate bean name
as discussed above.
Tip | |
---|---|
It is useful to provide an explicit id attribute on |
Calling a Chain from within a Chain
Sometimes you need to make a nested call to another chain from within a chain and then come back and continue execution within the original chain. To accomplish this you can utilize a Messaging Gateway by including a <gateway> element. For example:
<int:chain id="main-chain" input-channel="in" output-channel="out"> <int:header-enricher> <int:header name="name" value="Many" /> </int:header-enricher> <int:service-activator> <bean class="org.foo.SampleService" /> </int:service-activator> <int:gateway request-channel="inputA"/> </int:chain> <int:chain id="nested-chain-a" input-channel="inputA"> <int:header-enricher> <int:header name="name" value="Moe" /> </int:header-enricher> <int:gateway request-channel="inputB"/> <int:service-activator> <bean class="org.foo.SampleService" /> </int:service-activator> </int:chain> <int:chain id="nested-chain-b" input-channel="inputB"> <int:header-enricher> <int:header name="name" value="Jack" /> </int:header-enricher> <int:service-activator> <bean class="org.foo.SampleService" /> </int:service-activator> </int:chain>
In the above example the nested-chain-a will be called at the end of
main-chain processing by the 'gateway' element configured there. While in
nested-chain-a a call to a nested-chain-b will be made
after header enrichment and then it will come back to finish execution in nested-chain-b.
Finally the flow returns to the main-chain. When the nested version of a <gateway>
element is defined in the chain, it does not require the service-interface
attribute.
Instead, it simple takes the message in its current state and places it on the channel defined via
the request-channel
attribute. When the downstream flow initiated by that gateway completes,
a Message
will be returned to the gateway and continue its journey within
the current chain.