TechiQuest by Matt Vickery

Spring Integration Gateways - Null Handling & Timeouts

2012-02-08T06:50:00.000-08:00

Spring Integration (SI) Gateways

Spring Integration Gateways (<int:gateway>) provide a semantically rich interface to message sub-systems. Gateways are specified using namespace constructs, these reference a specific Java interface (<service-interface>) that is backed by an object dynamically implemented at run-time by the Spring Integration framework. Furthermore, these Java interfaces can, if you so wish, be defined entirely independent of any Spring artefacts - that's both code and configuration.

One of the primary advantages of using the SI gateway as an interface to message sub-systems is that it's possible to automatically adopt the benefit of rich, default and customisable, gateway configuration. One such configuration attribute deserves further scrutiny and discussion primarily because it's easy to misunderstand and misconfigure around - default-reply-timeout.

Primary Motivator for Gateway Analysis

During recent consulting engagements, I've encountered a number of deployments that use Spring Integration Gateway specifications that may, in some circumstances, lead to production operational instability. This has often been in high-pressure environments or those where technology support is not backed by adequate training, testing, review or technology mentoring.

How do gateways behave in Spring Integration (R2.0.5)

One of the key sections, regarding gateways, in the Spring Integration manual clearly explains gateway semantics. Below is a 2-dimensional table of possible non-standard gateway returns for each of the scenarios that the SI Manual (r2.0.5) refers to.

Gateway Non-standard Responses

Runtime Events	default-reply-timeout=x Single-threaded	default-reply-timeout=x Multi-threaded	default-reply-timeout=null Single-threaded	default-reply-timeout=null Multi-threaded
1. Long Running Process	Thread Parked	null returned	Thread Parked	Thread Parked
2. Null Returned Downstream	null returned	null returned	Thread Parked	Thread Parked
3. void method Downstream	null returned	null returned	Thread Parked	Thread Parked
4. Runtime Exception	Error handler invoked or exception thrown.	Error handler invoked or exception thrown.	Error handler invoked or exception thrown.	Error handler invoked or exception thrown.

The key parts of this table are the conditions that lead to invoking threads being parked (noted in red), nulls returned (noted in orange) and exceptions (noted in green). Each contributor consists of configuration that is under the developers control, deployed code that is under developers control and conditions that are usually not under developers control.

Clearly, the column headings in the table above are divided into two sections; two gateway configuration attributes. The default-reply-timeout is set by the SI configured and is the amount of time that a client call is wiling to wait for a response from the gateway. Secondly, synchronous flows are represented by Single-threaded flows, asynchronous by Multi-threaded flows.

A synchronous, or single-threaded flow, is one such as the following:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/integration
       http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">

    <import resource="common-beans.xml"/>

    <int:gateway id="enrollmentServiceGateway"
            service-interface="com.l8mdv.sample.EnrollmentServiceGateway"
            default-request-channel="gateway-request-channel"
            default-reply-channel="gateway-response-channel"
            default-reply-timeout="6000"/>

    <int:service-activator id="sleeperService" 
                           input-channel="gateway-request-channel"
                           output-channel="gateway-response-channel" 
                           ref="sleeper"/>

</beans>

The implicit input channel (gateway-request-channel) has no associated dispatcher configured. An asynchronous, or multi-threaded flow, is one such as the following:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/integration
       http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">

    <import resource="common-beans.xml"/>

    <int:gateway id="enrollmentServiceGateway"
                 service-interface="com.l8mdv.sample.EnrollmentServiceGateway"
                 default-request-channel="gateway-request-channel"
                 default-reply-channel="gateway-response-channel"
                 default-reply-timeout="6000"/>

    <int:channel id="gateway-request-channel">
        <int:dispatcher task-executor="taskExecutor"/>
    </int:channel>
    <int:service-activator id="sleeperService" 
                           input-channel="gateway-request-channel"
                           output-channel="gateway-response-channel" 
                           ref="sleeper"/>

</beans>

The explicit input channel has a dispatcher configured ("taskExecutor"). This task executor specifies a thread pool that supplies threads for execution and whose configuration as above marks a thread boundary. Note: This is not the only way of making channels asynchronous

The other configuration attribute referenced is default-reply-timeout, this is set on the gateway namespace configuration such as the example above. Note that both of these runtime aspects are set by the configurer during SI flow design and implementation. They are entirely under developer control.

The 'Runtime Events' column indicates gateway relevant runtime events that have to be considered during gateway configuration - these are obviously not under developer control. Trigger conditions for these events are not as unusual as one may hope.

1. Long Running Processes

It's not uncommon for thread pools to become exhausted because all pooled threads are waiting for an external resource accessed through a socket, this may be a long running database query, a firewall keeping a connection open despite the server terminating etc. There is significant potential for these types of trigger. Some long-running processes terminate naturally, sometimes they never completed - an application restart is required.

2. Null returned downstream

A null may be returned from a downstream SI construct such as a Transformer, Service Activator or Gateway. A Gateway may return null in some circumstances such as following a gateway timeout event.

3. Void method downstream

Any custom code invoked during an SI flow may use a void method signature. This can also be caused by configuration in circumstances where flows are determined dynamically at runtime.

4. Runtime Exception

RuntimeException's can be triggered during normal operation and are generally handled by catching them at the gateway or allowing them to propagate through. The reason that they are coloured green in the table above is that they are generally much easier to handle than timeouts.

Gateway Timeout Handling Strategies

There are four possible outcomes from invoking a gateway with a request message, all of these as a result of specific runtime events: a) an ordinary message response, b) an exception message, c) a null or d) no-response.

Ordinary business responses and exceptions are straight forward to understand and will not be covered further in this article. The two significant outcomes that will be explored further are strategies for dealing with nulls and no-response.

Generally speaking, long running processes either terminate or not. Long running processes that terminate may eventually return a message through the invoked gateway or timeout depending on timeout configuration, in which case a null may be returned. The severity of this as a problem depends on throughput volume, length of long running process and system resources (thread-pool size).

Configuration exists for default-reply-timeout

In the case where a long running process event is underway and a default-reply-timeout has been set, as long as the long running process completes before the default-reply-timeout expires, there is no problem to deal with. However, if the long running process does not complete before that timeout expires one of three outcomes will apply.

Firstly, if the long running process terminates subsequent to the reply timeout expiry, the gateway will have already returned null to the invoker so the null response needs handling by the invoker. The thread handling the long-running process will be returned to the pool.

Secondly, if the long running process does not terminate and a reply timeout has been set, the gateway will return null to the gateway invoker but the thread executing the long-running process will not get returned to the pool.

Thirdly, and most significantly, if a default-reply-timeout has been configured but the long running process is running on the same thread as the invoker, i.e. synchronous channels supply messages to that process, the thread will not return, the default-reply-timeout has no affect.

Assuming the most common processing scenario, a long running process completes either before or after the reply timeout expiry. When a null is returned by the gateway, the invoker is forced to deal with a null response. It's often unacceptable to force gateway consumers to deal with null responses and is not necessary as with a little additional configuration, this can be avoided.

Absent Configuration for default-reply-timeout

The most significant danger exists around gateways that have no default-reply-timeout configuration set.

A long running process or a null returned from downstream will mean that the invoking thread is parked. This is true for both synchronous and asynchronous flows and may ultimately force an application to be restarted because the invoker thread pool is likely to start on a depletion course if this continues to occur.

Spring Integration Timeout Handling Design Strategies

For those Spring Integration configuration designers that are comfortable with gateway invokers dealing with null responses, exceptions and set default-reply-timeouts on gateways, there's no need to read further. However, if you wish to provide clients of your gateway a more predictable response, a couple of strategies exist for handling null responses from gateways in order that invokers are protected from having to deal with them.

Firstly, the simpliest solution is to wrap the gateway with a service activator. The gateway must have the default-reply-timeout attribute value set in order to avoid unnecessary parking of threads. In order to avoid the consequence of long-running threads it's also very prudent to use a dispatcher soon after entry to the gateway - this breaks the thread boundary.

Whilst this is a valid technical approach, the impact is that we have forced a different entry point to our message sub-system. Entry is now via a Service Activator rather than a Gateway. A side affect of this change is that the testing entry point changes. Integration tests that would normally reference a gateway to send a message now have to locate the backing implementation for the Service Activator, not ideal.

An alternative approach toward solving this problem would be to configure two gateways with a Service Activator between them. Only one of the gateways would be exposed to invokers, the outer one. Both Gateways would reference the same service interface. The outer gateway specification would not specify the default-reply-timeout but would specify the input and output channels in the same way that a single gateway would. The Service Activator between the Gateways would handle null gateway responses and possibly any exceptions if preferred to the gateway error handler approach.

An example is as follows:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/integration
       http://www...org/schema/integration/spring-integration-2.1.xsd">

    <import resource="common-beans.xml"/>

    <int:gateway id="enrollmentServiceGateway"
                 service-interface="com.l8mdv.sample.EnrollmentServiceGateway"
                 default-request-channel="gateway-request-channel"/>

    <!-- This service activator has the adapted gateway injected as a bean and
    calls the adapted gateway from within the service method directly. -->
    <int:service-activator id="enrollmentServiceGatewayHandler"
                           input-channel="gateway-request-channel">
        <bean class="com.l8mdv.sample.EnrollmentServiceGatewayHandler">
            <constructor-arg name="enrollmentServiceAdaptedGateway"
                             ref="enrollmentServiceAdaptedGateway"/>
        </bean>
    </int:service-activator>

    <int:gateway id="enrollmentServiceAdaptedGateway"
                 service-interface="com.l8mdv.sample.EnrollmentServiceGateway"
                 default-request-channel="adapted-gateway-request-channel"
                 default-reply-timeout="6000"/>

    <int:channel id="adapted-gateway-request-channel"

                 datatype="java.lang.String">
        <int:dispatcher task-executor="taskExecutor"/>
    </int:channel>

</beans>

The Service Activator bean (enrollmentServiceGatewayHandler) deals with both null and exception responses from the adapted gateway (enrollmentServiceAdaptedGateway), in the situation where these are generated a business response detailing the error is generated.

Spring Integration R2.1 Changes

async-executor on gateway spec

Spring Integration - Splitter-Aggregator

2012-01-30T11:12:00.000-08:00

Within Spring Integration, one form of EIP scatter-gather is provided by the splitter and aggregator constructs. Semantics for both of these are pretty straight forward to understand, the splitter receives an input message and returns a list of objects that are each turned into first-class messages. Those messages get routed to message handlers following which they are aggregated as a list of input messages by an aggregator. This pattern can be used successfully with fairly simple configuration.

The following example works with trivial message processing taking place. It uses simple one-fail-all-fail error handling semantics. This means that any errors occurring in one or both services will result in an exception being thrown back to the invoker.

Having worked on projects that require a far more robust solution following partial failure, I've written another post that includes details regarding a design strategy to support that - A Robust Splitter Aggregator Strategy <URL>.

Let's explore how this context configuration is working and what happens in circumstances where and exception is thrown by one of the services.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/integration
       http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">

    <!--##############################################-->
    <!-- Bean specifications.  -->
    <!--##############################################-->
    <import resource="bean-refs.xml"/>

    <!--##############################################-->
    <!-- Gateway specification. -->
    <!--##############################################-->
    <int:gateway service-interface="com.l8mdv.sample.gateway.BrokerRequestGateway"/>
    <int:channel id="broker-request-channel"
                 datatype="com.l8mdv.sa.BrokerRequestMessage"/>

    <!--##############################################-->
    <!-- Request message splitter. -->
    <!--##############################################-->
    <int:splitter input-channel="broker-request-channel"
                  output-channel="broker-router-channel"
                  
                  ref="brokerQuoteRequestSplitter"/>

    <!--##############################################-->
    <!-- Request message routing. -->
    <!--##############################################-->
    <int:channel id="broker-router-channel"
                 datatype="com.l8mdv.sa.BrokerQuoteRequestMessage"/>
    <int:recipient-list-router input-channel="broker-router-channel">
        <int:recipient channel="openex-broker-channel"
                       selector-expression="payload.BrokerName.equals('openex')"/>
        <int:recipient channel="yahoo-broker-channel"
                       selector-expression="payload.BrokerName.equals('yahoo')"/>
    </int:recipient-list-router>

    <!--##############################################-->
    <!-- Request message routing to OpenEx. -->
    <!--##############################################-->
    <int:channel id="openex-broker-channel" datatype="com.l8mdv.sa.BrokerQuoteRequestMessage"/>
    <int:chain input-channel="openex-broker-channel"
               output-channel="aggregator-channel">
        <int:service-activator>
            <bean id="OpenExServiceFaker" class="com.l8mdv.sample.ServiceFaker">
                <constructor-arg name="response" ref="OpenExFakeResponseData"/>
            </bean>
        </int:service-activator>
    </int:chain>

    <!--##############################################-->
    <!-- Request message routing to Yahoo. -->
    <!--##############################################-->
    <int:channel id="yahoo-broker-channel"
                 datatype="com.l8mdv.sa.BrokerQuoteRequestMessage"/>
    <int:chain input-channel="yahoo-broker-channel"
               output-channel="aggregator-channel">
        <int:service-activator>
            <bean id="YahooServiceFaker" class="com.l8mdv.sample.ServiceFaker">
                <constructor-arg name="response" ref="YahooFakeResponseData"/>
            </bean>
        </int:service-activator>
    </int:chain>

    <!--##############################################-->
    <!-- Response message handling, return the best -->
    <!-- quote to the invoker. -->
    <!--##############################################-->
    <int:channel id="aggregator-channel" datatype="com.l8mdv.sa.BrokerQuoteResponseMessage"/>
    <int:chain input-channel="aggregator-channel">
        <int:aggregator/>
        <int:transformer ref="aggregationToBrokerQuoteResponseTransformer"/>
    </int:chain>

</beans>

Regarding the splitter and aggregator, the interesting configuration starts at line 25 with the splitter specification. The construct has been customised with input-channel, output-channel and ref attributes. The channel configuration for this construct is obvious, the other attribute allows association of a bean reference that is able to perform the split function. It's generally true that a splitter bean with appropriate arguments and return types will be invoked if it's unambiguous.

The next stop for messages is the router that's defined on line 35, the recipient list router. This message endpoint is able to forward route messages into appropriate channels given an expression to invoke on the payload of the message. This router will examine the payload and route each message to one of two services - these are located on lines 48 and 62. The service(s) that are invoked are entirely dependent on what is returned from the splitter. Either or both of these services may be invoked for a given splitter input message.

Finally, results of one or two service invocations are routed towards the aggregator, the configuration starts for this at the chain defined at line 74. Finally the result of the aggregation is input to a transformer where some further processing takes place. Notice that there is no output channel on this chain, the implicitly created default output channel is relied upon here - it goes back through the gateway to the gateway invoker.

There are some interesting aspects to this service:

If the splitter at line 25 receives a message but does not generate a list of one or more response messages, then an empty list will result on the router not getting called. This can be overridden by using the requires-reply attribute in which case an empty list will result in a message handling exception being thrown.
Strong typing has been used on the data channels in an attempt to enforce strict processing rules and make the configuration easier to follow and understand.
The chain construct has been used in an attempt to keep configuration compact where useful. It should be noticed that chain definition and strong typing are often two sides of the same coin. By grouping the aggregator and transformer in a chain I have been unable to control and hence document message type input to the transformer endpoint.
Spring beans referenced from within this context have been loaded from an external file. Whilst they could have been component-scanned or defined in the same file I have chosen to keep them distinct in order that they are not loaded if not necessary for operation - I'd usually create mock and spy objects around these message endpoints.
Any exception in Service Activator invocation at lines 48 and 62 would result in aggregation not completing. In this example, I have not created an error handler on the gateway in line 18 and so any exceptions thrown by SAs would result in an exception being thrown to the invoker of the gateway. A more robust solution, in the face of exception handling, would require a different design approach.
I have documented, albeit briefly, intent for each section of the configuration specifically in order to help readers understand intentions of my design.

In the case that they're useful, Spring bean definitions are as follows:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xmlns:util="http://www.springframework.org/schema/util"
       xsi:schemaLocation="http://www.springframework.org/schema/beans 
       http://www.springframework.org/schema/beans/spring-beans.xsd">

    <bean id="brokerQuoteRequestSplitter"
          class="com.l8mdv.sample.service.BrokerQuoteRequestSplitter"/>

    <bean id="brokerRawResponseTransformer"
          class="com.l8mdv.sample.service.impl.BrokerRawResponseTransformer"/>

    <bean id="aggregationToBrokerQuoteResponseTransformer"
          class="com.l8mdv.sample.service.impl.AggregationToBrokerQuoteResponseTransformer"/>

    <bean id="YahooFakeResponseData" class="com.l8mdv.sa.BrokerResponseMessage">
        <property name="brokerName" value="Yahoo"/>
        <property name="value" value="2"/>
        <property name="brokerRequestType" value="#{T(com.l8mdv.sa.BrokerRequestType).QUOTE}"/>
    </bean>

    <bean id="OpenExFakeResponseData" class="com.l8mdv.sa.BrokerResponseMessage">
        <property name="brokerName" value="OpenEx"/>
        <property name="value" value="5"/>
        <property name="brokerRequestType" value="#{T(com.l8mdv.sa.BrokerRequestType).QUOTE}"/>
    </bean>

</beans>

The splitter code:

package com.l8mdv.sample.service;

import com.l8mdv.sa.BrokerQuoteRequestMessage;
import com.l8mdv.sa.BrokerRequest;
import com.l8mdv.sa.BrokerRequestMessage;
import com.l8mdv.sa.QuoteRequestSortPolicy;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.List;

public class BrokerQuoteRequestSplitter {

    public List<BrokerQuoteRequestMessage> split(BrokerRequestMessage message) {

        Assert.notNull(message, "Mandatory argument missing.");
        List requests = new ArrayList();

        for (BrokerRequest brokerRequest:
                message.getBrokerRequest()) {
            BrokerQuoteRequestMessage brokerQuoteRequestMessage
                    = new BrokerQuoteRequestMessage();
            brokerQuoteRequestMessage
                    .setBrokerName(brokerRequest.getBrokerName());
            brokerQuoteRequestMessage
                    .setQuoteRequestSortPolicy(QuoteRequestSortPolicy.BUY_LOWEST);
            requests.add(brokerQuoteRequestMessage);
        }

        return requests;
    }
}

The transformer:

package com.l8mdv.sample.service.impl;

import com.l8mdv.sa.BrokerQuoteResponseMessage;
import org.springframework.util.Assert;

import java.util.List;

public class AggregationToBrokerQuoteResponseTransformer {

    public BrokerQuoteResponseMessage
    transform(List<BrokerQuoteResponseMessage> serviceResponses) {

        Assert.notNull(serviceResponses, "Mandatory argument missing.");

        BrokerQuoteResponseMessage bestQuote = null;
        for (BrokerQuoteResponseMessage 
                brokerQuoteResponseMessage: serviceResponses) {
            if (bestQuote == null)
                bestQuote = brokerQuoteResponseMessage;
            else {
                if (brokerQuoteResponseMessage.getSellPrice()
                        .compareTo(bestQuote.getSellPrice()) > 0)
                    bestQuote = brokerQuoteResponseMessage;
            }
        }

        return bestQuote;
    }
}

and the integration test:

package com.l8mdv.sample;

import com.l8mdv.sa.BrokerQuoteResponseMessage;
import com.l8mdv.sa.BrokerRequest;
import com.l8mdv.sa.BrokerRequestMessage;
import com.l8mdv.sa.BrokerRequestType;
import com.l8mdv.sample.gateway.BrokerRequestGateway;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(
    locations = {"classpath:META-INF/spring/simple-splitter-aggregator.xml"}
)
public class SimpleSplitterAggregatorIntegrationTest {

    @Autowired
    public BrokerRequestGateway brokerRequestGateway;

    @Test
    public void run() throws Exception {
        BrokerRequestMessage requestMessage = new BrokerRequestMessage();

        BrokerRequest yahooRequest = new BrokerRequest();
        yahooRequest.setBrokerName("yahoo");
        yahooRequest.setBrokerRequestType(BrokerRequestType.QUOTE);
        requestMessage.getBrokerRequest().add(yahooRequest);

        BrokerRequest oxRequest = new BrokerRequest();
        oxRequest.setBrokerName("openex");
        oxRequest.setBrokerRequestType(BrokerRequestType.QUOTE);
        requestMessage.getBrokerRequest().add(oxRequest);

        BrokerQuoteResponseMessage response = 
           brokerRequestGateway.send(requestMessage);
    }

}

Resources:

Spring Integration Reference Manual - Splitter Overview

Spring Integration - Input Channel Definition

2012-01-23T21:50:00.000-08:00

In a pipes and filters architecture, pipes are connectors or channels. Although at first sight trivial, channels are fairly rich semantically - they allow typing, synchronous and asynchronous input, direct and multicast notifications, send and wait (rendezvous) as well as queued input and wrapping by adapters.

To define or not to define?

Explicit specification or definition of the Spring Integration input-channel is not mandatory for SI constructs to operate. The framework will create input channels automatically if they are not explicitly defined in configuration, but what are the pros and cons of this aproach?

Take the following chain as an example:

<int:chain input-channel="processing-channel"
           output-channel="claim-check-out-channel">
        <int:service-activator expression="new String('different string')"/>
</int:chain>

The input-channel named "processing-channel" will get created automatically as it's not explicitly defined here. In a trivial configuration such as this one there's very little difference between including the channel explicitly or using the frameworks implicit creation. In larger configurations the extra lines taken by a dozen or more channel definitions may start to make the context appear a little cluttered. In this case it's then possible to consider decomposing the configuration but sometimes you just can't decompose those flows into distinct contexts, all of the constructs naturally belong together in a single context.

Channel Typing

One of the features of Spring Integration channels is that they can be strongly type matched with payload object types. It's worth considering adopting a convention for specifying the payload type on the channel because not only does this provide strong payload typing but improved better confidence that whomever is reading the flow after its built can readily see which type of objects are traversing the flows.

Of course channels can't always be strongly typed but in many cases they can. Here's an example of one that can:

<int:channel id="processing-channel" datatype="java.util.UUID"/>
<int:chain input-channel="processing-channel"
           output-channel="claim-check-out-channel">
    <int:service-activator expression="new String('different string')"/>
</int:chain>

You can see that on the first line of this Spring Integration configuration the contract for operation specifies that a java.util.UUID type object is expected on the input channel. In this case the payload contained a claim-check identifier and is to be replaced with an arbitrary string for the example.

In the case where a channel is strongly typed and the contract broken an exception will be thrown, here's an example of what you'll see:

org.springframework.integration.MessageDeliveryException: 
Channel 'processing-channel' expected one of the following datataypes 
[interface java.util.Map], 
but received [class java.util.UUID]
 at org.springframework.integration.channel.AbstractMessageChannel.convertPayloadIfNecessary(AbstractMessageChannel.java:187)

In this example I changed the datatype to java.util.Map and ran a test with a payload that's actually a java.util.UUID. That was the start of the stack trace that was generated when the exception was thrown.

It's possible to provide a collection of message types for channel specification. Any messages entering the channel must conform to at least one of the specified types otherwise a MessageDeliveryException is generated. The following example shows a change that will prevent the exception above. The configuration now allows for two types of message payload, java.util.Map and java.util.UUID to enter the channel.

<int:channel id="processing-channel" datatype="java.util.UUID, java.util.Map"/>
    <int:chain input-channel="processing-channel"
               output-channel="claim-check-out-channel">
        <int:service-activator expression="new String('different string')"/>
</int:chain>

Sources
Spring Integration Reference Manual - Message Channels

Spring Integration - Transforming a Header Object to a Payload

2012-01-23T20:35:00.000-08:00

Moving objects from the Spring Integration header can be done using several mechanisms. Firstly, a transformer construct can be used:

<int:chain input-channel="request-message-retrieval-channel">
  <int:transformer expression="headers.get(T(com.l8mdv.sample.HeaderManagementGateway)

      .REQUEST_PAYLOAD_HEADER_KEY)"/>
</int:chain>

Secondly, the same action can be performed using a Service Activator:

<int:chain input-channel="request-message-retrieval-channel">
  <int:service-activator expression="headers.get(T(com.l8mdv.sample.HeaderManagementGateway)

       .REQUEST_PAYLOAD_HEADER_KEY)"/>
</int:chain>

Here's a claim-check retrieval - notice the configuration option that triggers the Spring Integration framework to remove the object from the message store once it has been retrieved. If a message requires access several times in a flow, the last claim-check-out can be configured in this way in order to clean-up.

<int:chain input-channel="claim-check-out-channel">
  <int:transformer
     expression="headers.get('#{T(com.l8mdv.sample.ClaimCheckGateway).CLAIM_CHECK_ID}')"/>

  <int:claim-check-out message-store="simpleMessageStore"

                       remove-message="true"/>
</int:chain>

Several interesting points arise:

As can be seen here, it's advisable to reference constants for header key values rather than have the header name appearing as literals in several places throughout the code and configuration. On first sight the SpEL expression language may seem hard to fathom but it's worth investing the effort in order to understand it.
Don't forget that when header objects are moved back into Spring Integration payloads, the object remains in the Spring Integration message header. It may be worth clearing the header value away as part of transformation from the header back to the payload in order to maintain a cleaner and clearer Spring Integration message.
The intent should be clear from the configuration, what looks better a Service Activator expression or a Transformer? In my view the transformer conveys a slightly more accurate intent.
Consider using a claim-check integration pattern instead of loading the header. The Spring Integration claim-check uses a referenced message store (and provides a default implementation) that can be house-cleaned automatically by the framework.

Spring Integration - Payload Storage via Claim-check

2012-01-21T19:36:00.000-08:00

Continuing on the theme of temporary storage for transient messages used within Spring Integration flows, the claim-check model offers configurable storage for message payloads. The advantage in using this Enterprise Integration pattern, compared against header enrichment, is that objects don't have to be packed into the header using a Header Enrichment technique. They can be stored in a local Java Map, an IMDB, cache or anything else that be used to hold data.

Several advantages using this approach are evident. Firstly, performance and efficiency. When using header enrichment, if message payloads need to be managed outside of the JVM that generates the enriched message header, the object will not be available unless it's serialised and transported around the distributed application. This could be costly in terms of performance and transport efficiency. The key factor here is the frequency of remote dispatch and the size of the header object. In specific circumstances the claim-check pattern may offer an advantage here, objects can be serialised and/or transformed into a storage specific format and stored internally in memory or externally in a data store.

Secondly, accessibility. It's conceivable that message payloads undergoing claim-check processing may need to be accessed by third party applications that are unable to receive Spring Integration messages. The claim-check pattern allows this type of processing to take place.

Thirdly, resiliency is offered. A data store can be chosen that guarantees persistence for messages in order that they can be recovered following failure.

The following code details how the claim-check pattern can be used:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/integration
       http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">

    <bean id="simpleMessageStore"
          class="org.springframework.integration.store.SimpleMessageStore"/>
    <int:gateway id="claimCheckGateway"
                 service-interface="com.l8mdv.sample.ClaimCheckGateway"/>

    <int:chain input-channel="claim-check-in-channel"
               output-channel="processing-channel">
        <int:claim-check-in message-store="simpleMessageStore"/>
        <int:header-enricher>
            <int:header

                name="#{T(com.l8mdv.sample.ClaimCheckGateway).CLAIM_CHECK_ID}"
                expression="payload"/>
        </int:header-enricher>
    </int:chain>

    <int:chain input-channel="processing-channel"
               output-channel="claim-check-out-channel">
        <int:service-activator expression="new String('different string')"/>
    </int:chain>

    <int:chain input-channel="claim-check-out-channel">
        <int:transformer
                expression="headers.get('#{T(com.l8mdv.sample.ClaimCheckGateway)
                .CLAIM_CHECK_ID}')"/>
        <int:claim-check-out message-store="simpleMessageStore"
                             remove-message="true"/>
    </int:chain>

</beans>

The gateway used is specified as the following Java class:

package com.l8mdv.sample;

import org.springframework.integration.Message;
import org.springframework.integration.annotation.Gateway;

public interface ClaimCheckGateway {

    public static final String CLAIM_CHECK_ID = "CLAIM_CHECK_ID";

    @Gateway (requestChannel = "claim-check-in-channel")
    public Message<String> send(Message<String> message);
}

Lastly, this can all be tested by using the following JUnit test case:

package com.l8mdv.sample;

import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import static com.l8mdv.sample.ClaimCheckGateway.CLAIM_CHECK_ID;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(
    locations = {"classpath:META-INF/spring/claim-check.xml"}
)
public class ClaimCheckIntegrationTest {

    @Autowired ClaimCheckGateway claimCheckGateway;

    @Test
    public void locatePayloadInHeader() {
        String payload = "Sample test message.";
        Message<String> message = MessageBuilder.withPayload(payload).build();
        Message<String> response = claimCheckGateway.send(message);

        Assert.assertTrue(response.getPayload().equals(payload));
        Assert.assertTrue(response.getHeaders().get(CLAIM_CHECK_ID) != null);
    }
}

Spring Integration - Payload Storage via Header Enrichment

2012-01-21T08:27:00.000-08:00

There's often a need to temporarily store transient messages during design of Spring Integration flows - several different mechanisms are available in the toolkit.

It's pretty straight forward to take a message, use an SI header enricher construct and place the message in the header using a SpEL expression - in fact one for the header key name and one for the payload extraction.

The following SI flow demonstrates an example of how to do just that :

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans

       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/integration

       http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">

  <int:gateway id="headerManagementGateway"

                 service-interface="com.l8mdv.sample.HeaderManagementGateway"/>
  <int:chain input-channel="request-message-storage-channel"
             output-channel="request-message-retrieval-channel">
    <int:header-enricher>
      <int:header name="#{T(com.l8mdv.sample.HeaderManagementGateway)

                   .REQUEST_PAYLOAD_HEADER_KEY}" expression="getPayload()"/>
    </int:header-enricher>
  </int:chain>

  <int:chain input-channel="request-message-retrieval-channel">
    <int:transformer expression="headers.get(T(com.l8mdv.sample.HeaderManagementGateway)

               .REQUEST_PAYLOAD_HEADER_KEY)"/>

  </int:chain>
</beans>

This example can be executed by implementing a gateway as follows:

package com.l8mdv.sample;

import org.springframework.integration.Message;
import org.springframework.integration.annotation.Gateway;

public interface HeaderManagementGateway {

    public static final String REQUEST_PAYLOAD_HEADER_KEY = "REQUEST_PAYLOAD_HEADER_KEY";

    @Gateway (requestChannel = "request-message-storage-channel")
    public Message<String> send(Message<String> message);
}

and then running a test such as this one:

package com.l8mdv.sample;

import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import static com.l8mdv.sample.HeaderManagementGateway.REQUEST_PAYLOAD_HEADER_KEY;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {"classpath:META-INF/spring/header-management.xml"})
public class HeaderManagementIntegrationTest {

    @Autowired
    HeaderManagementGateway headerManagementGateway;

    @Test
    public void locatePayloadInHeader() {
        String payload = "Sample test message.";
        Message<String> message = MessageBuilder.withPayload(payload).build();
        Message<String> response = headerManagementGateway.send(message);

        Assert.assertTrue(response.getHeaders().get(REQUEST_PAYLOAD_HEADER_KEY).equals(payload));
    }

    @Test
    public void locateTransformedPayload() {
        String payload = "Sample test message.";
        Message<String> message = MessageBuilder.withPayload(payload).build();
        Message<String> response = headerManagementGateway.send(message);

        Assert.assertTrue(response.getPayload().contains(payload));
    }
}

For full source code and configuration, see the header-management maven module under git://github.com/L8MDV/l8mdv-si-samples.git

Subversion Commit Revert

2012-01-13T04:49:00.000-08:00

Subversion Commit Backout [svn merge -c]

Use:

Step one, perform a local merge with the previous version. Ensure that any work after the commit being reverted-to is saved as it will be overwritten by the merge.

svn merge -c -<commit-version-to-revert> <repo-url>

Step two, commit the reverted code with an appropriate commit message.

svn commit -m "Reverting commit [bad-commit-version] to previous version"

Note:- If any further commits take place between the initial check-in and the merge revert there will be further work to do.