You can configure Spring Integration components with XML elements that map directly to the terminology and concepts of enterprise integration. In many cases, the element names match those of the Enterprise Integration Patterns book.

To enable Spring Integration’s core namespace support within your Spring configuration files, add the following namespace reference and schema mapping in your top-level beans element:

<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.xsd">

(We have emphasized the lines that are particular to Spring Integration.)

You can choose any name after "xmlns:". We use int (short for Integration) for clarity, but you might prefer another abbreviation. On the other hand, if you use an XML editor or IDE support, the availability of auto-completion may convince you to keep the longer name for clarity. Alternatively, you can create configuration files that use the Spring Integration schema as the primary namespace, as the following example shows:

<beans:beans xmlns="http://www.springframework.org/schema/integration"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:beans="http://www.springframework.org/schema/beans"
       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.xsd">

(We have emphasized the lines that are particular to Spring Integration.)

When using this alternative, no prefix is necessary for the Spring Integration elements. On the other hand, if you define a generic Spring bean within the same configuration file, the bean element requires a prefix (<beans:bean .../>). Since it is generally a good idea to modularize the configuration files themselves (based on responsibility or architectural layer), you may find it appropriate to use the latter approach in the integration-focused configuration files, since generic beans are seldom necessary within those files. For the purposes of this documentation, we assume the integration namespace is the primary.

Spring Integration provides many other namespaces. In fact, each adapter type (JMS, file, and so on) that provides namespace support defines its elements within a separate schema. In order to use these elements, add the necessary namespaces with an xmlns entry and the corresponding schemaLocation mapping. For example, the following root element shows several of these namespace declarations:

<?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:int-file="http://www.springframework.org/schema/integration/file"
  xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
  xmlns:int-mail="http://www.springframework.org/schema/integration/mail"
  xmlns:int-rmi="http://www.springframework.org/schema/integration/rmi"
  xmlns:int-ws="http://www.springframework.org/schema/integration/ws"
  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.xsd
    http://www.springframework.org/schema/integration/file
    http://www.springframework.org/schema/integration/file/spring-integration-file.xsd
    http://www.springframework.org/schema/integration/jms
    http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
    http://www.springframework.org/schema/integration/mail
    http://www.springframework.org/schema/integration/mail/spring-integration-mail.xsd
    http://www.springframework.org/schema/integration/rmi
    http://www.springframework.org/schema/integration/rmi/spring-integration-rmi.xsd
    http://www.springframework.org/schema/integration/ws
    http://www.springframework.org/schema/integration/ws/spring-integration-ws.xsd">
 ...
</beans>

This reference manual provides specific examples of the various elements in their corresponding chapters. Here, the main thing to recognize is the consistency of the naming for each namespace URI and schema location.

In Spring Integration, the ApplicationContext plays the central role of a message bus, and you need to consider only a couple of configuration options. First, you may want to control the central TaskScheduler instance. You can do so by providing a single bean named taskScheduler. This is also defined as a constant, as follows:

IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME

By default, Spring Integration relies on an instance of ThreadPoolTaskScheduler, as described in the Task Execution and Scheduling section of the Spring Framework reference manual. That default TaskScheduler starts up automatically with a pool of ten threads, but see Section E.4, “Global Properties”. If you provide your own TaskScheduler instance instead, you can set the autoStartup property to false or provide your own pool size value.

When polling consumers provide an explicit task executor reference in their configuration, the invocation of the handler methods happens within that executor’s thread pool and not the main scheduler pool. However, when no task executor is provided for an endpoint’s poller, it is invoked by one of the main scheduler’s threads.

Caution

Do not run long-running tasks on poller threads. Use a task executor instead. If you have a lot of polling endpoints, you can cause thread starvation, unless you increase the pool size. Also, polling consumers have a default receiveTimeout of one second. Since the poller thread blocks for this time, we recommend that you use a task executor when many such endpoints exist, again to avoid starvation. Alternatively, you can reduce the receiveTimeout.

	Note
	An endpoint is a Polling Consumer if its input channel is one of the queue-based (that is, pollable) channels. Event-driven consumers are those having input channels that have dispatchers instead of queues (in other words, they are subscribable). Such endpoints have no poller configuration, since their handlers are invoked directly.

Important

When running in a JEE container, you may need to use Spring’s TimerManagerTaskScheduler, as described here, instead of the default taskScheduler. To do so, define a bean with the appropriate JNDI name for your environment, as the following example shows: <bean id="taskScheduler" class="o.s.scheduling.commonj.TimerManagerTaskScheduler"> <property name="timerManagerName" value="tm/MyTimerManager" /> <property name="resourceRef" value="true" /> </bean>

The next section describes what happens if exceptions occur within the asynchronous invocations.

As described in the overview at the very beginning of this manual, one of the main motivations behind a message-oriented framework such as Spring Integration is to promote loose coupling between components. The message channel plays an important role, in that producers and consumers do not have to know about each other. However, the advantages also have some drawbacks. Some things become more complicated in a loosely coupled environment, and one example is error handling.

When sending a message to a channel, the component that ultimately handles that message may or may not be operating within the same thread as the sender. If using a simple default DirectChannel (when the <channel> element that has no <queue> child element and no task-executor attribute), the message handling occurs in the same thread that sends the initial message. In that case, if an Exception is thrown, it can be caught by the sender (or it may propagate past the sender if it is an uncaught RuntimeException). So far, everything is fine. This is the same behavior as an exception-throwing operation in a normal call stack.

A message flow that runs on a caller thread might be invoked through a messaging gateway (see Section 10.4, “Messaging Gateways”) or a MessagingTemplate (see Section 6.1.4, “MessagingTemplate”). In either case, the default behavior is to throw any exceptions to the caller. For the messaging gateway, see Section 10.4.8, “Error Handling” for details about how the exception is thrown and how to configure the gateway to route the errors to an error channel instead. When using a MessagingTemplate or sending to a MessageChannel directly, exceptions are always thrown to the caller.

When adding asynchronous processing, things become rather more complicated. For instance, if the channel element does provide a queue child element, the component that handles the message operates in a different thread than the sender. The same is true when an ExecutorChannel is used. The sender may have dropped the Message into the channel and moved on to other things. There is no way for the Exception to be thrown directly back to that sender by using standard Exception throwing techniques. Instead, handling errors for asynchronous processes requires that the error-handling mechanism also be asynchronous.

Spring Integration supports error handling for its components by publishing errors to a message channel. Specifically, the Exception becomes the payload of a Spring Integration ErrorMessage. That Message is then sent to a message channel that is resolved in a way that is similar to the replyChannel resolution. First, if the request Message being handled at the time the Exception occurred contains an errorChannel header (the header name is defined in the MessageHeaders.ERROR_CHANNEL constant), the ErrorMessage is sent to that channel. Otherwise, the error handler sends to a "global" channel whose bean name is errorChannel (this is also defined as a constant: IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME).

A default errorChannel bean is created internally by the Framework. However, you can define your own if you want to control the settings. The following example shows how to define an error channel backed by a queue with a capacity of 500:

<int:channel id="errorChannel">
    <int:queue capacity="500"/>
</int:channel>

	Note
	The default error channel is a PublishSubscribeChannel.

The most important thing to understand here is that the messaging-based error handling applies only to exceptions that are thrown by a Spring Integration task that is executing within a TaskExecutor. This does not apply to exceptions thrown by a handler that operates within the same thread as the sender (for example, through a DirectChannel as described earlier in this section).

	Note
	When exceptions occur in a scheduled poller task’s execution, those exceptions are wrapped in ErrorMessage instances and sent to the errorChannel as well.

To enable global error handling, register a handler on that channel. For example, you can configure Spring Integration’s ErrorMessageExceptionTypeRouter as the handler of an endpoint that is subscribed to the errorChannel. That router can then spread the error messages across multiple channels, based on the Exception type.

Starting with version 4.3.10, Spring Integration provides the ErrorMessagePublisher and the ErrorMessageStrategy. You can use them as a general mechanism for publishing ErrorMessage instances. You can call or extend them in any error handling scenarios. The ErrorMessageSendingRecoverer extends this class as a RecoveryCallback implementation that can be used with retry, such as the RequestHandlerRetryAdvice. The ErrorMessageStrategy is used to build an ErrorMessage based on the provided exception and an AttributeAccessor context. It can be injected into any MessageProducerSupport or MessagingGatewaySupport. The requestMessage is stored under ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY in the AttributeAccessor context. The ErrorMessageStrategy can use that requestMessage as the originalMessage property of the ErrorMessage it creates. The DefaultErrorMessageStrategy does exactly that.

Certain global framework properties can be overridden by providing a properties file on the classpath.

The default properties can be found in /META-INF/spring.integration.default.properties in the spring-integration-core jar. You can see them on GitHub here. The following listing shows the default values:

spring.integration.channels.autoCreate=true 
spring.integration.channels.maxUnicastSubscribers=0x7fffffff 
spring.integration.channels.maxBroadcastSubscribers=0x7fffffff 
spring.integration.taskScheduler.poolSize=10 
spring.integration.messagingTemplate.throwExceptionOnLateReply=false 
spring.integration.readOnly.headers= 
spring.integration.endpoints.noAutoStartup= 
spring.integration.postProcessDynamicBeans=false 

These properties can be overridden by adding a /META-INF/spring.integration.properties file to the classpath. You need not provide all the properties — only those that you want to override.

In addition to the XML namespace support for configuring message endpoints, you can also use annotations. First, Spring Integration provides the class-level @MessageEndpoint as a stereotype annotation, meaning that it is itself annotated with Spring’s @Component annotation and is therefore automatically recognized as a bean definition by Spring’s component scanning.

Even more important are the various method-level annotations. They indicate that the annotated method is capable of handling a message. The following example demonstrates both class-level and method-level annotations:

@MessageEndpoint
public class FooService {

    @ServiceActivator
    public void processMessage(Message message) {
        ...
    }
}

Exactly what it means for the method to "handle" the Message depends on the particular annotation. Annotations available in Spring Integration include:

	Note
	If you use XML configuration in combination with annotations, the @MessageEndpoint annotation is not required. If you want to configure a POJO reference from the ref attribute of a <service-activator/> element, you can provide only the method-level annotations. In that case, the annotation prevents ambiguity even when no method-level attribute exists on the <service-activator/> element.

In most cases, the annotated handler method should not require the Message type as its parameter. Instead, the method parameter type can match the message’s payload type, as the following example shows:

public class ThingService {

    @ServiceActivator
    public void bar(Thing thing) {
        ...
    }

}

When the method parameter should be mapped from a value in the MessageHeaders, another option is to use the parameter-level @Header annotation. In general, methods annotated with the Spring Integration annotations can accept the Message itself, the message payload, or a header value (with @Header) as the parameter. In fact, the method can accept a combination, as the following example shows:

public class ThingService {

    @ServiceActivator
    public void otherThing(String payload, @Header("x") int valueX, @Header("y") int valueY) {
        ...
    }

}

You can also use the @Headers annotation to provide all of the message headers as a Map, as the following example shows:

public class ThingService {

    @ServiceActivator
    public void otherThing(String payload, @Headers Map<String, Object> headerMap) {
        ...
    }

}

	Note
	The value of the annotation can also be a SpEL expression (for example, someHeader.toUpperCase()), which is useful when you wish to manipulate the header value before injecting it. It also provides an optional required property, which specifies whether the attribute value must be available within the headers. The default value for the required property is true.

For several of these annotations, when a message-handling method returns a non-null value, the endpoint tries to send a reply. This is consistent across both configuration options (namespace and annotations) in that such an endpoint’s output channel is used (if available), and the REPLY_CHANNEL message header value is used as a fallback.

Tip

The combination of output channels on endpoints and the reply channel message header enables a pipeline approach, where multiple components have an output channel and the final component allows the reply message to be forwarded to the reply channel (as specified in the original request message). In other words, the final component depends on the information provided by the original sender and can dynamically support any number of clients as a result. This is an example of the return address pattern.

In addition to the examples shown here, these annotations also support the inputChannel and outputChannel properties, as the following example shows:

@Service
public class ThingService {

    @ServiceActivator(inputChannel="input", outputChannel="output")
    public void otherThing(String payload, @Headers Map<String, Object> headerMap) {
        ...
    }

}

The processing of these annotations creates the same beans as the corresponding XML components — AbstractEndpoint instances and MessageHandler instances (or MessageSource instances for the inbound channel adapter). See Section E.6.1, “Annotations on @Bean Methods”. The bean names are generated from the following pattern: [componentName].[methodName].[decapitalizedAnnotationClassShortName] (for example, for the preceding example the bean name is thingService.otherThing.serviceActivator) for the AbstractEndpoint and the same name with an additional .handler (.source) suffix for the MessageHandler (MessageSource) bean. The MessageHandler instances (MessageSource instances) are also eligible to be tracked by the message history.

Starting with version 4.0, all messaging annotations provide SmartLifecycle options (autoStartup and phase) to allow endpoint lifecycle control on application context initialization. They default to true and 0, respectively. To change the state of an endpoint (such as ` start()` or stop()), you can obtain a reference to the endpoint bean by using the BeanFactory (or autowiring) and invoke the methods. Alternatively, you can send a command message to the Control Bus (see Section 12.6, “Control Bus”). For these purposes, you should use the beanName mentioned earlier in the preceding paragraph.

public class AnnotationService {

    @Transformer(inputChannel = "input", outputChannel = "output",
        poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}", fixedDelay = "${poller.fixedDelay}"))
    public String handle(String payload) {
        ...
    }
}

@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata defaultPoller() {
    PollerMetadata pollerMetadata = new PollerMetadata();
    pollerMetadata.setTrigger(new PeriodicTrigger(10));
    return pollerMetadata;
}

public class AnnotationService {

    @Transformer(inputChannel = "aPollableChannel", outputChannel = "output")
    public String handle(String payload) {
        ...
    }
}

@Bean
public PollerMetadata myPoller() {
    PollerMetadata pollerMetadata = new PollerMetadata();
    pollerMetadata.setTrigger(new PeriodicTrigger(1000));
    return pollerMetadata;
}

public class AnnotationService {

    @Transformer(inputChannel = "aPollableChannel", outputChannel = "output"
                           poller = @Poller("myPoller"))
    public String handle(String payload) {
         ...
    }
}

@InboundChannelAdapter("counterChannel")
public Integer count() {
    return this.counter.incrementAndGet();
}

@InboundChannelAdapter(value = "fooChannel", poller = @Poller(fixed-rate = "5000"))
public String foo() {
    return "foo";
}

Starting with version 4.0, all messaging annotations can be configured as meta-annotations and all user-defined messaging annotations can define the same attributes to override their default values. In addition, meta-annotations can be configured hierarchically, as the following example shows:

@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@ServiceActivator(inputChannel = "annInput", outputChannel = "annOutput")
public @interface MyServiceActivator {

    String[] adviceChain = { "annAdvice" };
}

@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@MyServiceActivator
public @interface MyServiceActivator1 {

    String inputChannel();

    String outputChannel();
}
...

@MyServiceActivator1(inputChannel = "inputChannel", outputChannel = "outputChannel")
public Object service(Object payload) {
   ...
}

Configuring meta-annotations hierarchically lets users set defaults for various attributes and enables isolation of framework Java dependencies to user annotations, avoiding their use in user classes. If the framework finds a method with a user annotation that has a framework meta-annotation, it is treated as if the method were annotated directly with the framework annotation.

E.6.1 Annotations on @Bean Methods

Starting with version 4.0, you can configure messaging annotations on @Bean method definitions in @Configuration classes, to produce message endpoints based on the beans, not the methods. It is useful when @Bean definitions are "out-of-the-box" MessageHandler instances (AggregatingMessageHandler, DefaultMessageSplitter, and others), Transformer instances (JsonToObjectTransformer, ClaimCheckOutTransformer, and others), and MessageSource instances (FileReadingMessageSource, RedisStoreMessageSource, and others). The following example shows how to use messaging annotations with @Bean annotations:

@Configuration
@EnableIntegration
public class MyFlowConfiguration {

    @Bean
    @InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
    public MessageSource<String> consoleSource() {
        return CharacterStreamReadingMessageSource.stdin();
    }

    @Bean
    @Transformer(inputChannel = "inputChannel", outputChannel = "httpChannel")
    public ObjectToMapTransformer toMapTransformer() {
        return new ObjectToMapTransformer();
    }

    @Bean
    @ServiceActivator(inputChannel = "httpChannel")
    public MessageHandler httpHandler() {
    HttpRequestExecutingMessageHandler handler = new HttpRequestExecutingMessageHandler("http://foo/service");
        handler.setExpectedResponseType(String.class);
        handler.setOutputChannelName("outputChannel");
        return handler;
    }

    @Bean
    @ServiceActivator(inputChannel = "outputChannel")
    public LoggingHandler loggingHandler() {
        return new LoggingHandler("info");
    }

}

Version 5.0 introduced support for a @Bean annotated with @InboundChannelAdapter that returns java.util.function.Supplier, which can produce either a POJO or a Message. The followig example shows how to use that combination:

@Configuration
@EnableIntegration
public class MyFlowConfiguration {

    @Bean
    @InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
    public Supplier<String> pojoSupplier() {
        return () -> "foo";
    }

    @Bean
    @InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
    public Supplier<Message<String>> messageSupplier() {
        return () -> new GenericMessage<>("foo");
    }
}

The meta-annotation rules work on @Bean methods as well (the @MyServiceActivator annotation described earlier can be applied to a @Bean definition).

Note

When you use these annotations on consumer @Bean definitions, if the bean definition returns an appropriate MessageHandler (depending on the annotation type), you must set attributes (such as outputChannel, requiresReply, order, and others), on the MessageHandler @Bean definition itself. Only the following annotation attributes are used: adviceChain, autoStartup, inputChannel, phase, and poller. All other attributes are for the handler.

	Note
	The bean names are generated with the following algorithm:

• The MessageHandler (MessageSource) @Bean gets its own standard name from the method name or name attribute on the @Bean. This works as though there were no messaging annotation on the @Bean method.
• The AbstractEndpoint bean name is generated with the following pattern: [configurationComponentName].[methodName].[decapitalizedAnnotationClassShortName]. For example, the SourcePollingChannelAdapter endpoint for the consoleSource() definition shown earlier gets a bean name of myFlowConfiguration.consoleSource.inboundChannelAdapter. See also Section 5.4.8, “Endpoint Bean Names”.

	Important
	When using these annotations on @Bean definitions, the inputChannel must reference a declared bean. Channels are not automatically declared in this case.

Note

With Java configuration, you can use any @Conditional (for example, @Profile) definition on the @Bean method level to skip the bean registration for some conditional reason. The following example shows how to do so: @Bean @ServiceActivator(inputChannel = "skippedChannel") @Profile("thing") public MessageHandler skipped() { return System.out::println; } Together with the existing Spring container logic, the messaging endpoint bean (based on the @ServiceActivator annotation), is also not registered.

@Bean
public PollableChannel bridgeFromInput() {
    return new QueueChannel();
}

@Bean
@BridgeFrom(value = "bridgeFromInput", poller = @Poller(fixedDelay = "1000"))
public MessageChannel bridgeFromOutput() {
    return new DirectChannel();
}
@Bean
public QueueChannel bridgeToOutput() {
    return new QueueChannel();
}

@Bean
@BridgeTo("bridgeToOutput")
public MessageChannel bridgeToInput() {
    return new DirectChannel();
}

Spring Integration implements a flexible facility to map messages to methods and their arguments without providing extra configuration, by relying on some default rules and defining certain conventions. The examples in the following sections articulate the rules.

public String doSomething(Object o);

public Message doSomething(Object o);

public int doSomething(Message  msg);

public Message doSomething(Message msg);

public Message doSomething(Map m);

public Message doSomething(Map h, <T> t);

public String doSomething();

public void soSomething();

public String doSomething(@Payload String s, @Header("someheader") String b) 

public String doSomething(@Payload String s, @RequestParam("something") String b) 

public String foo(String s, @Header("foo") String b) 

public String soSomething(@Headers Map m, @Header("something") Map f, @Header("someotherthing") String bar)

E.7.3 Complex Scenarios

The following example uses multiple parameters:

Multiple parameters can create a lot of ambiguity with regards to determining the appropriate mappings. The general advice is to annotate your method parameters with @Payload, @Header, and @Headers. The examples in this section show ambiguous conditions that result in an exception being raised.

public String doSomething(String s, int i)

The two parameters are equal in weight. Therefore, there is no way to determine which one is a payload.

The following example shows a similar problem, only with three parameters:

public String foo(String s, Map m, String b)

Although the Map could be easily mapped to message headers, there is no way to determine what to do with the two String parameters.

The following example shows another ambiguous method:

public String foo(Map m, Map f)

Although one might argue that one Map could be mapped to the message payload and the other one to the message headers, we cannot rely on the order.

	Tip
	Any method signature with more than one method argument that is not (Map, <T>) and with unannotated parameters results in an ambiguous condition and triggers an exception.

The next set of examples each show mutliple methods that result in ambiguity.

Message handlers with multiple methods are mapped based on the same rules that are described earlier (in the examples). However, some scenarios might still look confusing.

The following example shows multiple methods with legal (mappable and unambiguous) signatures:

public class Something {
    public String doSomething(String str, Map m);

    public String doSomething(Map m);
}

(Whether the methods have the same name or different names makes no difference). The Message could be mapped to either method. The first method would be invoked when the message payload could be mapped to str and the message headers could be mapped to m. The second method could also be a candidate by mapping only the message headers to m. To make matters worse, both methods have the same name. At first, that might look ambiguous because of the following configuration:

<int:service-activator input-channel="input" output-channel="output" method="doSomething">
    <bean class="org.things.Something"/>
</int:service-activator>

It works because mappings are based on the payload first and everything else next. In other words, the method whose first argument can be mapped to a payload takes precedence over all other methods.

Now consider an alternate example, which produces a truly ambiguous condition:

public class Something {
    public String doSomething(String str, Map m);

    public String doSomething(String str);
}

Both methods have signatures that could be mapped to a message payload. They also have the same name. Such handler methods will trigger an exception. However, if the method names were different, you could influence the mapping with a method attribute (shown in the next example). The following example shows the same example with two different method names:

public class Something {
    public String doSomething(String str, Map m);

    public String doSomethingElse(String str);
}

The following example shows how to use the method attribute to dictate the mapping:

<int:service-activator input-channel="input" output-channel="output" method="doSomethingElse">
    <bean class="org.bar.Foo"/>
</int:service-activator>

Because the configuration explicitly maps the doSomethingElse method, we have eliminated the ambiguity.