See Section 14.13, “Strict Message Ordering”.

The java.util.function interfaces now have improved integration support in the Framework components. Also Kotlin lambdas now can be used for handler and source methods.

See Section 10.11, “java.util.function Interfaces Support”.

A JUnit 5 @LongRunningTest conditional annotation is provided to check the environment or system properties for the RUN_LONG_INTEGRATION_TESTS entry with the value of true to determine if test should be run or skipped.

See Section F.1.5, “JUnit Rules and Conditions”.

The IntegrationFlowContext is now an interface and IntegrationFlowRegistration is an inner interface of IntegrationFlowContext.

A new logAndReply() operator has been introduced for convenience when you wish to log at the end of a flow for request-reply configurations. This avoid confusion with log() which is treated as a one-way end flow component.

A generated bean name for any NamedComponent within an integration flow is now based on the component type for better readability from visual tools, logs analyzers and metrics collectors.

The GenericHandler.handle() now excepts a MessageHeaders type for the second argument.

Exceptions caught and re-thrown by AbstractDispatcher are now more consistent:

Previously:

Global channel interceptors now apply to dynamically registered channels, such as through the IntegrationFlowContext when using the Java DSL or beans that are initialized using beanFactory.initializeBean(). Previously, when beans were created after the application context was refreshed, interceptors were not applied.

ChannelInterceptor.postReceive() is no longer called when no message is received; it is no longer necessary to check for a null Message<?>. Previously, the method was called. If you have an interceptor that relies on the previous behavior, implement afterReceiveCompleted() instead, since that method is invoked, regardless of whether a message is received or not. Furthermore, the PolledAmqpChannel and PolledJmsChannel previously did not invoke afterReceiveCompleted() with null; they now do.

A new ResultType.BYTES mode is introduced for the ObjectToJsonTransformer.

See the section called “JSON Transformers” for more information.

Starting with version 5.0.5, generated bean names for the components in an IntegrationFlow include the flow bean name, followed by a dot, as a prefix. For example, if a flow bean were named flowBean, a generated bean might be named flowBean.generatedBean.

See Section 11.13, “Working With Message Flows” for more information.

If the groupTimeout is evaluated to a negative value, an aggregator now expires the group immediately. Only null is considered as a signal to do nothing for the current message.

A new popSequence property has been introduced to allow (by default) to call a MessageBuilder.popSequenceDetails() for the output message. Also an AbstractAggregatingMessageGroupProcessor returns now an AbstractIntegrationMessageBuilder instead of the whole Message for optimization.

See Section 8.4, “Aggregator” for more information.

Starting with version 5.1, you must explicitly turn on the @Publisher AOP functionality by using @EnablePublisher or by using the <int:enable-publisher> child element on <int:annotation-config>.

See Section B.1.1, “Annotation-driven Configuration with the @Publisher Annotation” for more information.

In Spring Integration, a message is a generic wrapper for any Java object combined with metadata used by the framework while handling that object. It consists of a payload and headers. The payload can be of any type, and the headers hold commonly required information such as ID, timestamp, correlation ID, and return address. Headers are also used for passing values to and from connected transports. For example, when creating a message from a received file, the file name may be stored in a header to be accessed by downstream components. Likewise, if a message’s content is ultimately going to be sent by an outbound mail adapter, the various properties (to, from, cc, subject, and others) may be configured as message header values by an upstream component. Developers can also store any arbitrary key-value pairs in the headers.

Figure 5.1. Message

A message channel represents the "pipe" of a pipes-and-filters architecture. Producers send messages to a channel, and consumers receive messages from a channel. The message channel therefore decouples the messaging components and also provides a convenient point for interception and monitoring of messages.

Figure 5.2. Message Channel

A message channel may follow either point-to-point or publish-subscribe semantics. With a point-to-point channel, no more than one consumer can receive each message sent to the channel. Publish-subscribe channels, on the other hand, attempt to broadcast each message to all subscribers on the channel. Spring Integration supports both of these models.

Whereas "point-to-point" and "publish-subscribe" define the two options for how many consumers ultimately receive each message, there is another important consideration: Should the channel buffer messages? In Spring Integration, pollable channels are capable of buffering Messages within a queue. The advantage of buffering is that it allows for throttling the inbound messages and thereby prevents overloading a consumer. However, as the name suggests, this also adds some complexity, since a consumer can only receive the messages from such a channel if a poller is configured. On the other hand, a consumer connected to a subscribable channel is simply message-driven. Section 6.1.2, “Message Channel Implementations” has a detailed discussion of the variety of channel implementations available in Spring Integration.

One of the primary goals of Spring Integration is to simplify the development of enterprise integration solutions through inversion of control. This means that you should not have to implement consumers and producers directly, and you should not even have to build messages and invoke send or receive operations on a message channel. Instead, you should be able to focus on your specific domain model with an implementation based on plain objects. Then, by providing declarative configuration, you can "connect" your domain-specific code to the messaging infrastructure provided by Spring Integration. The components responsible for these connections are message endpoints. This does not mean that you should necessarily connect your existing application code directly. Any real-world enterprise integration solution requires some amount of code focused upon integration concerns such as routing and transformation. The important thing is to achieve separation of concerns between the integration logic and the business logic. In other words, as with the Model-View-Controller (MVC) paradigm for web applications, the goal should be to provide a thin but dedicated layer that translates inbound requests into service layer invocations and then translates service layer return values into outbound replies. The next section provides an overview of the message endpoint types that handle these responsibilities, and, in upcoming chapters, you can see how Spring Integration’s declarative configuration options provide a non-invasive way to use each of these.

A message transformer is responsible for converting a message’s content or structure and returning the modified message. Probably the most common type of transformer is one that converts the payload of the message from one format to another (such as from XML to java.lang.String). Similarly, a transformer can add, remove, or modify the message’s header values.

A message filter determines whether a message should be passed to an output channel at all. This simply requires a boolean test method that may check for a particular payload content type, a property value, the presence of a header, or other conditions. If the message is accepted, it is sent to the output channel. If not, it is dropped (or, for a more severe implementation, an Exception could be thrown). Message filters are often used in conjunction with a publish-subscribe channel, where multiple consumers may receive the same message and use the criteria of the filter to narrow down the set of messages to be processed.

Note

Be careful not to confuse the generic use of "filter" within the pipes-and-filters architectural pattern with this specific endpoint type that selectively narrows down the messages flowing between two channels. The pipes-and-filters concept of a "filter" matches more closely with Spring Integration’s message endpoint: any component that can be connected to a message channel in order to send or receive messages.

A message router is responsible for deciding what channel or channels (if any) should receive the message next. Typically, the decision is based upon the message’s content or the metadata available in the message headers. A message router is often used as a dynamic alternative to a statically configured output channel on a service activator or other endpoint capable of sending reply messages. Likewise, a message router provides a proactive alternative to the reactive message filters used by multiple subscribers, as described earlier.

Figure 5.3. Message Router

A splitter is another type of message endpoint whose responsibility is to accept a message from its input channel, split that message into multiple messages, and send each of those to its output channel. This is typically used for dividing a "composite" payload object into a group of messages containing the subdivided payloads.

Basically a mirror-image of the splitter, the aggregator is a type of message endpoint that receives multiple messages and combines them into a single message. In fact, aggregators are often downstream consumers in a pipeline that includes a splitter. Technically, the aggregator is more complex than a splitter, because it is required to maintain state (the messages to be aggregated), to decide when the complete group of messages is available, and to timeout if necessary. Furthermore, in case of a timeout, the aggregator needs to know whether to send the partial results, discard them, or send them to a separate channel. Spring Integration provides a CorrelationStrategy, a ReleaseStrategy, and configurable settings for timeout, whether to send partial results upon timeout, and a discard channel.

A Service Activator is a generic endpoint for connecting a service instance to the messaging system. The input message channel must be configured, and, if the service method to be invoked is capable of returning a value, an output message Channel may also be provided.

	Note
	The output channel is optional, since each message may also provide its own Return Address header. This same rule applies for all consumer endpoints.

The service activator invokes an operation on some service object to process the request message, extracting the request message’s payload and converting (if the method does not expect a message-typed parameter). Whenever the service object’s method returns a value, that return value is likewise converted to a reply message if necessary (if it is not already a message type). That reply message is sent to the output channel. If no output channel has been configured, the reply is sent to the channel specified in the message’s "return address", if available.

A request-reply service activator endpoint connects a target object’s method to input and output Message Channels.

Figure 5.4. Service Activator

	Note
	As discussed earlier, in Message Channel, channels can be pollable or subscribable. In the preceding diagram, this is depicted by the "clock" symbol and the solid arrow (poll) and the dotted arrow (subscribe).

A channel adapter is an endpoint that connects a message channel to some other system or transport. Channel adapters may be either inbound or outbound. Typically, the channel adapter does some mapping between the message and whatever object or resource is received from or sent to the other system (file, HTTP Request, JMS message, and others). Depending on the transport, the channel adapter may also populate or extract message header values. Spring Integration provides a number of channel adapters, which are described in upcoming chapters.

Figure 5.5. An inbound channel adapter endpoint connects a source system to a MessageChannel.

	Note
	Message sources can be pollable (for example, POP3) or message-driven_ (for example, IMAP Idle). In the preceding diagram, this is depicted by the "clock" symbol and the solid arrow (poll) and the dotted arrow (message-driven).

Figure 5.6. An outbound channel adapter endpoint connects a MessageChannel to a target system.

	Note
	As discussed earlier in Message Channel, channels can be pollable or subscribable. In the preceding diagram, this is depicted by the "clock" symbol and the solid arrow (poll) and the dotted arrow (subscribe).

Consuming endpoints (anything with an inputChannel) consist of two beans, the consumer and the message handler. The consumer has a reference to the message handler and invokes it as messages arrive.

Consider the following XML example:

<int:service-activator id = "someService" ... />

Given the preceding example, the bean names are as follows:

When using Enterprise Integration Pattern (EIP) annotations, the names depend on several factors. Consider the following example of an annotated POJO:

@Component
public class SomeComponent {

    @ServiceActivator(inputChannel = ...)
    public String someMethod(...) {
        ...
    }

}

Given the preceding example, the bean names are as follows:

Starting with version 5.0.4, you can modify these names by using the @EndpointId annotation, as the following example shows:

@Component
public class SomeComponent {

    @EndpointId("someService")
    @ServiceActivator(inputChannel = ...)
    public String someMethod(...) {
        ...
    }

}

Given the preceding example, the bean names are as follows:

The @EndpointId creates names as created by the id attribute with XML configuration. Consider the following example of an annotated bean:

@Configuratiom
public class SomeConfiguration {

    @Bean
    @ServiceActivator(inputChannel = ...)
    public MessageHandler someHandler() {
        ...
    }

}

Given the preceding example, the bean names are as follows:

Starting with version 5.0.4, you can modify these names by using the @EndpointId annotation, as the following example shows:

@Configuratiom
public class SomeConfiguration {

    @Bean("someService.handler")             
    @EndpointId("someService")               
    @ServiceActivator(inputChannel = ...)
    public MessageHandler someHandler() {
        ...
    }

}

The @EndpointId annotation creates names as created by the id attribute with XML configuration, as long as you use the convention of appending .handler to the @Bean name.

There is one special case where a third bean is created: For architectural reasons, if a MessageHandler @Bean does not define an AbstractReplyProducingMessageHandler, the framework wraps the provided bean in a ReplyProducingMessageHandlerWrapper. This wrapper supports request handler advice handling and emits the normal produced no reply debug log messages. Its bean name is the handler bean name plus .wrapper (when there is an @EndpointId — otherwise, it is the normal generated handler name).

Similarly Pollable Message Sources create two beans, a SourcePollingChannelAdapter (SPCA) and a MessageSource.

Consider the following XML configuration:

<int:inbound-channel-adapter id = "someAdapter" ... />

Given the preceding XML configuration, the bean names are as follows:

Consider the following Java configuration of a POJO to define an @EndpointId:

@EndpointId("someAdapter")
@InboundChannelAdapter(channel = "channel3", poller = @Poller(fixedDelay = "5000"))
public String pojoSource() {
    ...
}

Given the preceding Java configuration example, the bean names are as follows:

Consider the following Java configuration of a bean to define an @EndpointID:

@Bean("someAdapter.source")
@EndpointId("someAdapter")
@InboundChannelAdapter(channel = "channel3", poller = @Poller(fixedDelay = "5000"))
public MessageSource<?> source() {
    return () -> {
        ...
    };
}

Given the preceding example, the bean names are as follows:

Spring Integration bootstraps certain features by using Spring Framework’s SpringFactories mechanism to load several IntegrationConfigurationInitializer classes. This includes the -core jar as well as certain others, including -http and -jmx. The information for this process is stored in a META-INF/spring.factories file in each jar.

Some developers prefer to repackage their application and all dependencies into a single jar by using well known tools, such as the Apache Maven Shade Plugin.

By default, the shade plugin does not merge the spring.factories files when producing the shaded jar.

In addition to spring.factories, other META-INF files (spring.handlers and spring.schemas) are used for XML configuration. These files also need to be merged.

	Important
	Spring Boot’s executable jar mechanism takes a different approach, in that it nests the jars, thus retaining each spring.factories file on the class path. So, with a Spring Boot application, nothing more is needed if you use its default executable jar format.

Even if you do not use Spring Boot, you can still use the tooling provided by Boot to enhance the shade plugin by adding transformers for the above mentioned files.

You may wish to consult the current spring-boot-starter-parent pom to see the current settings that boot uses. The following example shows how to configure the plugin:

...
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <configuration>
                <keepDependenciesWithProvidedScope>true</keepDependenciesWithProvidedScope>
                <createDependencyReducedPom>true</createDependencyReducedPom>
            </configuration>
            <dependencies>
                <dependency> 
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                    <version>${spring.boot.version}</version>
                </dependency>
            </dependencies>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <transformers> 
                            <transformer
                                implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                <resource>META-INF/spring.handlers</resource>
                            </transformer>
                            <transformer
                                implementation="org.springframework.boot.maven.PropertiesMergingResourceTransformer">
                                <resource>META-INF/spring.factories</resource>
                            </transformer>
                            <transformer
                                implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                <resource>META-INF/spring.schemas</resource>
                            </transformer>
                            <transformer
                                implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
...

You can add a property for ${spring.boot.version} or use an explicit version.

When using XML configuration, to avoid getting false schema validation errors, you should use a "Spring-aware" IDE, such as the Spring Tool Suite (STS), Eclipse with the Spring IDE plugins, or IntelliJ IDEA. These IDEs know how to resolve the correct XML schema from the classpath (by using the META-INF/spring.schemas file in the jars). When using STS or Eclipse with the plugin, you must enable Spring Project Nature on the project.

The schemas hosted on the internet for certain legacy modules (those that existed in version 1.0) are the 1.0 versions for compatibility reasons. If your IDE uses these schemas, you are likely to see false errors.

Each of these online schemas has a warning similar to the following:

Important

This schema is for the 1.0 version of Spring Integration Core. We cannot update it to the current schema because that will break any applications using 1.0.3 or lower. For subsequent versions, the unversioned schema is resolved from the classpath and obtained from the jar. Please refer to github: https://github.com/spring-projects/spring-integration/tree/master/spring-integration-core/src/main/resources/org/springframework/integration/config

The affected modules are

With XML configuration and Spring Integration Namespace support, the XML parsers hide how target beans are declared and wired together. For Java configuration, it is important to understand the Framework API for target end-user applications.

The first-class citizens for EIP implementation are Message, Channel, and Endpoint (see Section 5.3, “Main Components”, earlier in this chapter). Their implementations (contracts) are:

The first two are simple enough to understand how to implement, configure, and use. The last one deserves more attention

The AbstractEndpoint is widely used throughout the Spring Framework for different component implementations. Its main implementations are:

When you use messaging annotations or the Java DSL, you need ot worry about these components, because the Framework automatically produces them with appropriate annotations and BeanPostProcessor implementations. When building components manually, you should use the ConsumerEndpointFactoryBean to help determine the target AbstractEndpoint consumer implementation to create, based on the provided inputChannel property.

On the other hand, the ConsumerEndpointFactoryBean delegates to an another first class citizen in the Framework: org.springframework.messaging.MessageHandler. The goal of the implementation of this interface is to handle the message consumed by the endpoint from the channel. All EIP components in Spring Integration are MessageHandler implementations (for example, AggregatingMessageHandler, MessageTransformingHandler, AbstractMessageSplitter, and others). The target protocol outbound adapters (FileWritingMessageHandler, HttpRequestExecutingMessageHandler, AbstractMqttMessageHandler, and others) are also MessageHandler implementations. When you develop Spring Integration applications with Java configuration, you should look into the Spring Integration module to find an appropriate MessageHandler implementation to use for the @ServiceActivator\ configuration. For example, to send an XMPP message (see Chapter 39, XMPP Support) you should configure something like the following:

@Bean
@ServiceActivator(inputChannel = "input")
public MessageHandler sendChatMessageHandler(XMPPConnection xmppConnection) {
    ChatMessageSendingMessageHandler handler = new ChatMessageSendingMessageHandler(xmppConnection);

    DefaultXmppHeaderMapper xmppHeaderMapper = new DefaultXmppHeaderMapper();
    xmppHeaderMapper.setRequestHeaderNames("*");
    handler.setHeaderMapper(xmppHeaderMapper);

    return handler;
}

The MessageHandler implementations represent the outbound and processing part of the message flow.

The inbound message flow side has its own components, which are divided into polling and listening behaviors. The listening (message-driven) components are simple and typically require only one target class implementation to be ready to produce messages. Listening components can be one-way MessageProducerSupport implementations, (such as AbstractMqttMessageDrivenChannelAdapter and ImapIdleChannelAdapter) or request-reply MessagingGatewaySupport implementations (such as AmqpInboundGateway and AbstractWebServiceInboundGateway).

Polling inbound endpoints are for those protocols that do not provide a listener API or are not intended for such a behavior, including any file based protocol (such as FTP), any data bases (RDBMS or NoSQL), and others.

These inbound endpoints consist of two components: the poller configuration, to initiate the polling task periodically, and a message source class to read data from the target protocol and produce a message for the downstream integration flow. The first class for the poller configuration is a SourcePollingChannelAdapter. It is one more AbstractEndpoint implementation, but especially for polling to initiate an integration flow. Typically, with the messaging annotations or Java DSL, you should not worry about this class. The Framework produces a bean for it, based on the @InboundChannelAdapter configuration or a Java DSL builder spec.

Message source components are more important for the target application development, and they all implement the MessageSource interface (for example, MongoDbMessageSource and AbstractTwitterMessageSource). With that in mind, our config for reading data from an RDBMS table with JDBC could resemble the following:

@Bean
@InboundChannelAdapter(value = "fooChannel", poller = @Poller(fixedDelay="5000"))
public MessageSource<?> storedProc(DataSource dataSource) {
    return new JdbcPollingChannelAdapter(dataSource, "SELECT * FROM foo where status = 0");
}

You can find all the required inbound and outbound classes for the target protocols in the particular Spring Integration module (in most cases, in the respective package). For example, the spring-integration-websocket adapters are:

If you are familiar with Spring Integration XML configuration, starting with version 4.3, we provide information in the XSD element definitions about which target classes are used to declare beans for the adapter or gateway, as the following example shows:

<xsd:element name="outbound-async-gateway">
    <xsd:annotation>
		<xsd:documentation>
Configures a Consumer Endpoint for the 'o.s.i.amqp.outbound.AsyncAmqpOutboundGateway'
that will publish an AMQP Message to the provided Exchange and expect a reply Message.
The sending thread returns immediately; the reply is sent asynchronously; uses 'AsyncRabbitTemplate.sendAndReceive()'.
       </xsd:documentation>
	</xsd:annotation>

Spring Integration’s top-level MessageChannel interface is defined as follows:

public interface MessageChannel {

    boolean send(Message message);

    boolean send(Message message, long timeout);
}

When sending a message, the return value is true if the message is sent successfully. If the send call times out or is interrupted, it returns false.

public interface PollableChannel extends MessageChannel {

    Message<?> receive();

    Message<?> receive(long timeout);

}

public interface SubscribableChannel extends MessageChannel {

    boolean subscribe(MessageHandler handler);

    boolean unsubscribe(MessageHandler handler);

}

Spring Integration provides several different message channel implementations. The following sections briefly describe each one.

PublishSubscribeChannel

The PublishSubscribeChannel implementation broadcasts any Message sent to it to all of its subscribed handlers. This is most often used for sending event messages, whose primary role is notification (as opposed to document messages, which are generally intended to be processed by a single handler). Note that the PublishSubscribeChannel is intended for sending only. Since it broadcasts to its subscribers directly when its send(Message) method is invoked, consumers cannot poll for messages (it does not implement PollableChannel and therefore has no receive() method). Instead, any subscriber must itself be a MessageHandler, and the subscriber’s handleMessage(Message) method is invoked in turn.

Prior to version 3.0, invoking the send method on a PublishSubscribeChannel that had no subscribers returned false. When used in conjunction with a MessagingTemplate, a MessageDeliveryException was thrown. Starting with version 3.0, the behavior has changed such that a send is always considered successful if at least the minimum subscribers are present (and successfully handle the message). This behavior can be modified by setting the minSubscribers property, which defaults to 0.

	Note
	If you use a TaskExecutor, only the presence of the correct number of subscribers is used for this determination, because the actual handling of the message is performed asynchronously.

public QueueChannel(int capacity)

RendezvousChannel

The RendezvousChannel enables a "direct-handoff" scenario, wherein a sender blocks until another party invokes the channel’s receive() method. The other party blocks until the sender sends the message. Internally, this implementation is quite similar to the QueueChannel, except that it uses a SynchronousQueue (a zero-capacity implementation of BlockingQueue). This works well in situations where the sender and receiver operate in different threads, but asynchronously dropping the message in a queue is not appropriate. In other words, with a RendezvousChannel, the sender knows that some receiver has accepted the message, whereas with a QueueChannel, the message would have been stored to the internal queue and potentially never received.

Tip

Keep in mind that all of these queue-based channels are storing messages in-memory only by default. When persistence is required, you can either provide a message-store attribute within the queue element to reference a persistent MessageStore implementation or you can replace the local channel with one that is backed by a persistent broker, such as a JMS-backed channel or channel adapter. The latter option lets you take advantage of any JMS provider’s implementation for message persistence, as discussed in Chapter 23, JMS Support. However, when buffering in a queue is not necessary, the simplest approach is to rely upon the DirectChannel, discussed in the next section.

The RendezvousChannel is also useful for implementing request-reply operations. The sender can create a temporary, anonymous instance of RendezvousChannel, which it then sets as the replyChannel header when building a Message. After sending that Message, the sender can immediately call receive (optionally providing a timeout value) in order to block while waiting for a reply Message. This is very similar to the implementation used internally by many of Spring Integration’s request-reply components.

DirectChannel

The DirectChannel has point-to-point semantics but otherwise is more similar to the PublishSubscribeChannel than any of the queue-based channel implementations described earlier. It implements the SubscribableChannel interface instead of the PollableChannel interface, so it dispatches messages directly to a subscriber. As a point-to-point channel, however, it differs from the PublishSubscribeChannel in that it sends each Message to a single subscribed MessageHandler.

In addition to being the simplest point-to-point channel option, one of its most important features is that it enables a single thread to perform the operations on "both sides" of the channel. For example, if a handler subscribes to a DirectChannel, then sending a Message to that channel triggers invocation of that handler’s handleMessage(Message) method directly in the sender’s thread, before the send() method invocation can return.

The key motivation for providing a channel implementation with this behavior is to support transactions that must span across the channel while still benefiting from the abstraction and loose coupling that the channel provides. If the send call is invoked within the scope of a transaction, the outcome of the handler’s invocation (for example, updating a database record) plays a role in determining the ultimate result of that transaction (commit or rollback).

Note

Since the DirectChannel is the simplest option and does not add any additional overhead that would be required for scheduling and managing the threads of a poller, it is the default channel type within Spring Integration. The general idea is to define the channels for an application, consider which of those need to provide buffering or to throttle input, and modify those to be queue-based PollableChannels. Likewise, if a channel needs to broadcast messages, it should not be a DirectChannel but rather a PublishSubscribeChannel. Later, we show how each of these channels can be configured.

The DirectChannel internally delegates to a message dispatcher to invoke its subscribed message handlers, and that dispatcher can have a load-balancing strategy exposed by load-balancer or load-balancer-ref attributes (mutually exclusive). The load balancing strategy is used by the message dispatcher to help determine how messages are distributed amongst message handlers when multiple message handlers subscribe to the same channel. As a convenience, the load-balancer attribute exposes an enumeration of values pointing to pre-existing implementations of LoadBalancingStrategy. round-robin (load-balances across the handlers in rotation) and none (for the cases where one wants to explicitly disable load balancing) are the only available values. Other strategy implementations may be added in future versions. However, since version 3.0, you can provide your own implementation of the LoadBalancingStrategy and inject it by using the load-balancer-ref attribute, which should point to a bean that implements LoadBalancingStrategy, as the following example shows:

<int:channel id="lbRefChannel">
  <int:dispatcher load-balancer-ref="lb"/>
</int:channel>

<bean id="lb" class="foo.bar.SampleLoadBalancingStrategy"/>

Note that the load-balancer and load-balancer-ref attributes are mutually exclusive.

The load-balancing also works in conjunction with a boolean failover property. If the "failover" value is true (the default), the dispatcher falls back to any subsequent handlers (as necessary) when preceding handlers throw exceptions. The order is determined by an optional order value defined on the handlers themselves or, if no such value exists, the order in which the handlers subscribed.

If a certain situation requires that the dispatcher always try to invoke the first handler and then fall back in the same fixed order sequence every time an error occurs, no load-balancing strategy should be provided. In other words, the dispatcher still supports the failover boolean property even when no load-balancing is enabled. Without load-balancing, however, the invocation of handlers always begins with the first, according to their order. For example, this approach works well when there is a clear definition of primary, secondary, tertiary, and so on. When using the namespace support, the order attribute on any endpoint determines the order.

	Note
	Keep in mind that load-balancing and failover apply only when a channel has more than one subscribed message handler. When using the namespace support, this means that more than one endpoint shares the same channel reference defined in the input-channel attribute.

ExecutorChannel

The ExecutorChannel is a point-to-point channel that supports the same dispatcher configuration as DirectChannel (load-balancing strategy and the failover boolean property). The key difference between these two dispatching channel types is that the ExecutorChannel delegates to an instance of TaskExecutor to perform the dispatch. This means that the send method typically does not block, but it also means that the handler invocation may not occur in the sender’s thread. It therefore does not support transactions that span the sender and receiving handler.

Caution

The sender can sometimes block. For example, when using a TaskExecutor with a rejection policy that throttles the client (such as the ThreadPoolExecutor.CallerRunsPolicy), the sender’s thread can execute the method any time the thread pool is at its maximum capacity and the executor’s work queue is full. Since that situation would only occur in a non-predictable way, you should not rely upon it for transactions.

<int:channel id="threadScopedChannel" scope="thread">
     <int:queue />
</int:channel>

<bean class="org.springframework.beans.factory.config.CustomScopeConfigurer">
    <property name="scopes">
        <map>
            <entry key="thread" value="org.springframework.context.support.SimpleThreadScope" />
        </map>
    </property>
</bean>

One of the advantages of a messaging architecture is the ability to provide common behavior and capture meaningful information about the messages passing through the system in a non-invasive way. Since the Message instances are sent to and received from MessageChannel instances, those channels provide an opportunity for intercepting the send and receive operations. The ChannelInterceptor strategy interface, shown in the following listing, provides methods for each of those operations:

public interface ChannelInterceptor {

    Message<?> preSend(Message<?> message, MessageChannel channel);

    void postSend(Message<?> message, MessageChannel channel, boolean sent);

    void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex);

    boolean preReceive(MessageChannel channel);

    Message<?> postReceive(Message<?> message, MessageChannel channel);

    void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex);
}

After implementing the interface, registering the interceptor with a channel is just a matter of making the following call:

channel.addInterceptor(someChannelInterceptor);

The methods that return a Message instance can be used for transforming the Message or can return null to prevent further processing (of course, any of the methods can throw a RuntimeException). Also, the preReceive method can return false to prevent the receive operation from proceeding.

Note

Keep in mind that receive() calls are only relevant for PollableChannels. In fact, the SubscribableChannel interface does not even define a receive() method. The reason for this is that when a Message is sent to a SubscribableChannel, it is sent directly to zero or more subscribers, depending on the type of channel (for example, a PublishSubscribeChannel sends to all of its subscribers). Therefore, the preReceive(...), postReceive(...), and afterReceiveCompletion(...) interceptor methods are invoked only when the interceptor is applied to a PollableChannel.

Spring Integration also provides an implementation of the Wire Tap pattern. It is a simple interceptor that sends the Message to another channel without otherwise altering the existing flow. It can be very useful for debugging and monitoring. An example is shown in the section called “Wire Tap”.

Because it is rarely necessary to implement all of the interceptor methods, the interface provides no-op methods (methods returning void method have no code, the Message-returning methods return the Message as-is, and the boolean method returns true).

Tip

The order of invocation for the interceptor methods depends on the type of channel. As described earlier, the queue-based channels are the only ones where the receive method is intercepted in the first place. Additionally, the relationship between send and receive interception depends on the timing of the separate sender and receiver threads. For example, if a receiver is already blocked while waiting for a message, the order could be as follows: preSend, preReceive, postReceive, postSend. However, if a receiver polls after the sender has placed a message on the channel and has already returned, the order would be as follows: preSend, postSend (some-time-elapses), preReceive, postReceive. The time that elapses in such a case depends on a number of factors and is therefore generally unpredictable (in fact, the receive may never happen). The type of queue also plays a role (for example, rendezvous versus priority). In short, you cannot rely on the order beyond the fact that preSend precedes postSend and preReceive precedes postReceive.

Starting with Spring Framework 4.1 and Spring Integration 4.1, the ChannelInterceptor provides new methods: afterSendCompletion() and afterReceiveCompletion(). They are invoked after send()' and 'receive() calls, regardless of any exception that is raised, which allow for resource cleanup. Note that the channel invokes these methods on the ChannelInterceptor list in the reverse order of the initial preSend() and preReceive() calls.

Starting with version 5.1, global channel interceptors now apply to dynamically registered channels - such as through beans that are initialized by using beanFactory.initializeBean() or IntegrationFlowContext when using the Java DSL. Previously, interceptors were not applied when beans were created after the application context was refreshed.

Also, starting with version 5.1, ChannelInterceptor.postReceive() is no longer called when no message is received; it is no longer necessary to check for a null Message<?>. Previously, the method was called. If you have an interceptor that relies on the previous behavior, implement afterReceiveCompleted() instead, since that method is invoked, regardless of whether a message is received or not.

When the endpoints and their various configuration options are introduced, Spring Integration provides a foundation for messaging components that enables non-invasive invocation of your application code from the messaging system. However, it is sometimes necessary to invoke the messaging system from your application code. For convenience when implementing such use cases, Spring Integration provides a MessagingTemplate that supports a variety of operations across the message channels, including request and reply scenarios. For example, it is possible to send a request and wait for a reply, as follows:

MessagingTemplate template = new MessagingTemplate();

Message reply = template.sendAndReceive(someChannel, new GenericMessage("test"));

In the preceding example, a temporary anonymous channel would be created internally by the template. The sendTimeout and receiveTimeout properties may also be set on the template, and other exchange types are also supported. The following listing shows the signatures for such methods:

public boolean send(final MessageChannel channel, final Message<?> message) { ...
}

public Message<?> sendAndReceive(final MessageChannel channel, final Message<?> request) { ...
}

public Message<?> receive(final PollableChannel<?> channel) { ...
}

	Note
	A less invasive approach that lets you invoke simple interfaces with payload or header values instead of Message instances is described in Section 10.4.1, “Enter the GatewayProxyFactoryBean”.

To create a message channel instance, you can use the <channel/> element, as follows:

<int:channel id="exampleChannel"/>

The default channel type is point-to-point. To create a publish-subscribe channel, use the <publish-subscribe-channel/> element, as follows:

<int:publish-subscribe-channel id="exampleChannel"/>

When you use the <channel/> element without any sub-elements, it creates a DirectChannel instance (a SubscribableChannel).

However, you can alternatively provide a variety of <queue/> sub-elements to create any of the pollable channel types (as described in Section 6.1.2, “Message Channel Implementations”). The following sections shows examples of each channel type.

<int:channel id="directChannel"/>

<int:channel id="failFastChannel">
    <int:dispatcher failover="false"/>
</channel>

<int:channel id="channelWithFixedOrderSequenceFailover">
    <int:dispatcher load-balancer="none"/>
</int:channel>

<int:channel id="numberChannel" datatype="java.lang.Number"/>

<int:channel id="stringOrNumberChannel" datatype="java.lang.String,java.lang.Number"/>

MessageChannel inChannel = context.getBean("numberChannel", MessageChannel.class);
inChannel.send(new GenericMessage<String>("5"));

Exception in thread "main" org.springframework.integration.MessageDeliveryException:
Channel 'numberChannel'
expected one of the following datataypes [class java.lang.Number],
but received [class java.lang.String]
…

public static class StringToIntegerConverter implements Converter<String, Integer> {
    public Integer convert(String source) {
        return Integer.parseInt(source);
    }
}

<int:converter ref="strToInt"/>

<bean id="strToInt" class="org.springframework.integration.util.Demo.StringToIntegerConverter"/>

QueueChannel Configuration

To create a QueueChannel, use the <queue/> sub-element. You may specify the channel’s capacity as follows:

<int:channel id="queueChannel">
    <queue capacity="25"/>
</int:channel>

	Note
	If you do not provide a value for the capacity attribute on this <queue/> sub-element, the resulting queue is unbounded. To avoid issues such as running out of memory, we highly recommend that you set an explicit value for a bounded queue.

Persistent QueueChannel Configuration

Since a QueueChannel provides the capability to buffer messages but does so in-memory only by default, it also introduces a possibility that messages could be lost in the event of a system failure. To mitigate this risk, a QueueChannel may be backed by a persistent implementation of the MessageGroupStore strategy interface. For more details on MessageGroupStore and MessageStore, see Section 12.4, “Message Store”.

	Important
	The capacity attribute is not allowed when the message-store attribute is used.

When a QueueChannel receives a Message, it adds the message to the message store. When a Message is polled from a QueueChannel, it is removed from the message store.

By default, a QueueChannel stores its messages in an in-memory queue, which can lead to the lost message scenario mentioned earlier. However, Spring Integration provides persistent stores, such as the JdbcChannelMessageStore.

You can configure a message store for any QueueChannel by adding the message-store attribute, as the following example shows:

<int:channel id="dbBackedChannel">
    <int:queue message-store="channelStore"/>
</int:channel>

<bean id="channelStore" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
    <property name="dataSource" ref="dataSource"/>
    <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
</bean>

The Spring Integration JDBC module also provides a schema Data Definition Language (DDL) for a number of popular databases. These schemas are located in the org.springframework.integration.jdbc.store.channel package of that module (spring-integration-jdbc).

	Important
	One important feature is that, with any transactional persistent store (such as JdbcChannelMessageStore), as long as the poller has a transaction configured, a message removed from the store can be permanently removed only if the transaction completes successfully. Otherwise the transaction rolls back, and the Message is not lost.

Many other implementations of the message store are available as the growing number of Spring projects related to "NoSQL" data stores come to provide underlying support for these stores. You can also provide your own implementation of the MessageGroupStore interface if you cannot find one that meets your particular needs.

Since version 4.0, we recommend that QueueChannel instances be configured to use a ChannelMessageStore, if possible. These are generally optimized for this use, as compared to a general message store. If the ChannelMessageStore is a ChannelPriorityMessageStore, the messages are received in FIFO within priority order. The notion of priority is determined by the message store implementation. For example, the following example shows the Java configuration for the Section 25.2.1, “MongoDB Channel Message Store”:

@Bean
public BasicMessageGroupStore mongoDbChannelMessageStore(MongoDbFactory mongoDbFactory) {
    MongoDbChannelMessageStore store = new MongoDbChannelMessageStore(mongoDbFactory);
    store.setPriorityEnabled(true);
    return store;
}

@Bean
public PollableChannel priorityQueue(BasicMessageGroupStore mongoDbChannelMessageStore) {
    return new PriorityChannel(new MessageGroupQueue(mongoDbChannelMessageStore, "priorityQueue"));
}

	Note
	Pay attention to the MessageGroupQueue class. That is a BlockingQueue implementation to use the MessageGroupStore operations.

The same implementation with Java DSL might look like the following example:

@Bean
public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) {
    return IntegrationFlows.from((Channels c) ->
            c.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup"))
            ....
            .get();
}

Another option to customize the QueueChannel environment is provided by the ref attribute of the <int:queue> sub-element or its particular constructor. This attribute supplies the reference to any java.util.Queue implementation. For example, a Hazelcast distributed IQueue can be configured as follows:

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance(new Config()
                                           .setProperty("hazelcast.logging.type", "log4j"));
}

@Bean
public PollableChannel distributedQueue() {
    return new QueueChannel(hazelcastInstance()
                              .getQueue("springIntegrationQueue"));
}

PublishSubscribeChannel Configuration

To create a PublishSubscribeChannel, use the <publish-subscribe-channel/> element. When using this element, you can also specify the task-executor used for publishing messages (if none is specified, it publishes in the sender’s thread), as follows:

<int:publish-subscribe-channel id="pubsubChannel" task-executor="someExecutor"/>

If you provide a resequencer or aggregator downstream from a PublishSubscribeChannel, you can set the apply-sequence property on the channel to true. Doing so indicates that the channel should set the sequence-size and sequence-number message headers as well as the correlation ID prior to passing along the messages. For example, if there are five subscribers, the sequence-size would be set to 5, and the messages would have sequence-number header values ranging from 1 to 5.

Along with the Executor, you can also configure an ErrorHandler. By default, the PublishSubscribeChannel uses a MessagePublishingErrorHandler implementation to send an error to the MessageChannel from the errorChannel header or into the global errorChannel instance. If an Executor is not configured, the ErrorHandler is ignored and exceptions are thrown directly to the caller’s thread.

If you provide a Resequencer or Aggregator downstream from a PublishSubscribeChannel, you can set the apply-sequence property on the channel to true. Doing so indicates that the channel should set the sequence-size and sequence-number message headers as well as the correlation ID prior to passing along the messages. For example, if there are five subscribers, the sequence-size would be set to 5, and the messages would have sequence-number header values ranging from 1 to 5.

The following example shows how to set the apply-sequence header to true:

<int:publish-subscribe-channel id="pubsubChannel" apply-sequence="true"/>

	Note
	The apply-sequence value is false by default so that a publish-subscribe channel can send the exact same message instances to multiple outbound channels. Since Spring Integration enforces immutability of the payload and header references, when the flag is set to true, the channel creates new Message instances with the same payload reference but different header values.

ExecutorChannel

To create an ExecutorChannel, add the <dispatcher> sub-element with a task-executor attribute. The attribute’s value can reference any TaskExecutor within the context. For example, doing so enables configuration of a thread pool for dispatching messages to subscribed handlers. As mentioned earlier, doing so breaks the single-threaded execution context between sender and receiver so that any active transaction context is not shared by the invocation of the handler (that is, the handler may throw an Exception, but the send invocation has already returned successfully). The following example shows how to use the dispatcher element and specify an executor in the task-executor attribute:

<int:channel id="executorChannel">
    <int:dispatcher task-executor="someExecutor"/>
</int:channel>

Note

The load-balancer and failover options are also both available on the <dispatcher/> sub-element, as described earlier in the section called “DirectChannel Configuration”. The same defaults apply. Consequently, the channel has a round-robin load-balancing strategy with failover enabled unless explicit configuration is provided for one or both of those attributes, as the following example shows: <int:channel id="executorChannelWithoutFailover"> <int:dispatcher task-executor="someExecutor" failover="false"/> </int:channel>

<int:channel id="priorityChannel">
    <int:priority-queue capacity="20"/>
</int:channel>

<int:channel id="priorityChannel" datatype="example.Widget">
    <int:priority-queue comparator="widgetComparator"
                    capacity="10"/>
</int:channel>

<int:channel id="rendezvousChannel"/>
    <int:rendezvous-queue/>
</int:channel>

<int:channel id="threadLocalChannel" scope="thread"/>

<int:channel id="exampleChannel">
    <int:interceptors>
        <ref bean="trafficMonitoringInterceptor"/>
    </int:interceptors>
</int:channel>

Global Channel Interceptor Configuration

Channel interceptors provide a clean and concise way of applying cross-cutting behavior per individual channel. If the same behavior should be applied on multiple channels, configuring the same set of interceptors for each channel would not be the most efficient way. To avoid repeated configuration while also enabling interceptors to apply to multiple channels, Spring Integration provides global interceptors. Consider the following pair of examples:

<int:channel-interceptor pattern="input*, thing2*, thing1, !cat*" order="3">
    <bean class="thing1.thing2SampleInterceptor"/>
</int:channel-interceptor>

<int:channel-interceptor ref="myInterceptor" pattern="input*, thing2*, thing1, !cat*" order="3"/>

<bean id="myInterceptor" class="thing1.thing2SampleInterceptor"/>

Each <channel-interceptor/> element lets you define a global interceptor, which is applied on all channels that match any patterns defined by the pattern attribute. In the preceding case, the global interceptor is applied on the thing1 channel and all other channels that begin with thing2 or input but not to channels starting with thing3 (since version 5.0).

	Warning
	The addition of this syntax to the pattern causes one possible (though perhaps unlikely) problem. If you have a bean named !thing1 and you included a pattern of !thing1 in your channel interceptor’s pattern patterns, it no longer matches. The pattern now matches all beans not named thing1. In this case, you can escape the ! in the pattern with \. The pattern \!thing1 matches a bean named !thing1.

The order attribute lets you manage where this interceptor is injected when there are multiple interceptors on a given channel. For example, channel inputChannel could have individual interceptors configured locally (see below), as the following example shows:

<int:channel id="inputChannel"> 
  <int:interceptors>
    <int:wire-tap channel="logger"/> 
  </int:interceptors>
</int:channel>

A reasonable question is "how is a global interceptor injected in relation to other interceptors configured locally or through other global interceptor definitions?" The current implementation provides a simple mechanism for defining the order of interceptor execution. A positive number in the order attribute ensures interceptor injection after any existing interceptors, while a negative number ensures that the interceptor is injected before existing interceptors. This means that, in the preceding example, the global interceptor is injected after (since its order is greater than 0) the wire-tap interceptor configured locally. If there were another global interceptor with a matching pattern, its order would be determined by comparing the values of both interceptors' order attributes. To inject a global interceptor before the existing interceptors, use a negative value for the order attribute.

	Note
	Note that both the order and pattern attributes are optional. The default value for order will be 0 and for pattern, the default is * (to match all channels).

Starting with version 4.3.15, you can configure the spring.integration.postProcessDynamicBeans = true property to apply any global interceptors to dynamically created MessageChannel beans. See Section E.4, “Global Properties” for more information.

Wire Tap

As mentioned earlier, Spring Integration provides a simple wire tap interceptor. You can configure a wire tap on any channel within an <interceptors/> element. Doing so is especially useful for debugging and can be used in conjunction with Spring Integration’s logging channel adapter as follows:

<int:channel id="in">
    <int:interceptors>
        <int:wire-tap channel="logger"/>
    </int:interceptors>
</int:channel>

<int:logging-channel-adapter id="logger" level="DEBUG"/>

Tip

The logging-channel-adapter also accepts an expression attribute so that you can evaluate a SpEL expression against the payload and headers variables. Alternatively, to log the full message toString() result, provide a value of true for the log-full-message attribute. By default, it is false so that only the payload is logged. Setting it to true enables logging of all headers in addition to the payload. The expression option provides the most flexibility (for example, expression="payload.user.name").

One of the common misconceptions about the wire tap and other similar components (Section B.1, “Message Publishing Configuration”) is that they are automatically asynchronous in nature. By default, wire tap as a component is not invoked asynchronously. Instead, Spring Integration focuses on a single unified approach to configuring asynchronous behavior: the message channel. What makes certain parts of the message flow synchronous or asynchronous is the type of Message Channel that has been configured within that flow. That is one of the primary benefits of the message channel abstraction. From the inception of the framework, we have always emphasized the need and the value of the message channel as a first-class citizen of the framework. It is not just an internal, implicit realization of the EIP pattern. It is fully exposed as a configurable component to the end user. So, the wire tap component is only responsible for performing the following tasks:

• Intercept a message flow by tapping into a channel (for example, channelA)
• Grab each message
• Send the message to another channel (for example, channelB)

It is essentially a variation of the bridge pattern, but it is encapsulated within a channel definition (and hence easier to enable and disable without disrupting a flow). Also, unlike the bridge, it basically forks another message flow. Is that flow synchronous or asynchronous? The answer depends on the type of message channel that channelB is. We have the following options: direct channel, pollable channel, and executor channel. The last two break the thread boundary, making communication over such channels asynchronous, because the dispatching of the message from that channel to its subscribed handlers happens on a different thread than the one used to send the message to that channel. That is what is going to make your wire-tap flow synchronous or asynchronous. It is consistent with other components within the framework (such as message publisher) and adds a level of consistency and simplicity by sparing you from worrying in advance (other than writing thread-safe code) about whether a particular piece of code should be implemented as synchronous or asynchronous. The actual wiring of two pieces of code (say, component A and component B) over a message channel is what makes their collaboration synchronous or asynchronous. You may even want to change from synchronous to asynchronous in the future, and message channel lets you to do it swiftly without ever touching the code.

One final point regarding the wire tap is that, despite the rationale provided above for not being asynchronous by default, you should keep in mind that it is usually desirable to hand off the message as soon as possible. Therefore, it would be quite common to use an asynchronous channel option as the wire tap’s outbound channel. However we doe not enforce asynchronous behavior by default. There are a number of use cases that would break if we did, including that you might not want to break a transactional boundary. Perhaps you use the wire tap pattern for auditing purposes, and you do want the audit messages to be sent within the original transaction. As an example, you might connect the wire tap to a JMS outbound channel adapter. That way, you get the best of both worlds: 1) the sending of a JMS Message can occur within the transaction while 2) it is still a "fire-and-forget" action, thereby preventing any noticeable delay in the main message flow.

Important

Starting with version 4.0, it is important to avoid circular references when an interceptor (such as the WireTap class) references a channel. You need to exclude such channels from those being intercepted by the current interceptor. This can be done with appropriate patterns or programmatically. If you have a custom ChannelInterceptor that references a channel, consider implementing VetoCapableInterceptor. That way, the framework asks the interceptor if it is OK to intercept each channel that is a candidate, based on the supplied pattern. You can also add runtime protection in the interceptor methods to ensure that the channel is not one that is referenced by the interceptor. The WireTap uses both of these techniques.

Starting with version 4.3, the WireTap has additional constructors that take a channelName instead of a MessageChannel instance. This can be convenient for Java configuration and when channel auto-creation logic is being used. The target MessageChannel bean is resolved from the provided channelName later, on the first interaction with the interceptor.

	Important
	Channel resolution requires a BeanFactory, so the wire tap instance must be a Spring-managed bean.

This late-binding approach also allows simplification of typical wire-tapping patterns with Java DSL configuration, as the following example shows:

@Bean
public PollableChannel myChannel() {
    return MessageChannels.queue()
            .wireTap("loggingFlow.input")
            .get();
}

@Bean
public IntegrationFlow loggingFlow() {
    return f -> f.log();
}

Global Wire Tap Configuration

It is possible to configure a global wire tap as a special case of the the section called “Global Channel Interceptor Configuration”. To do so, configure a top level wire-tap element. Now, in addition to the normal wire-tap namespace support, the pattern and order attributes are supported and work in exactly the same way as they do for the channel-interceptor. The following examlpe shows how to configure a global wire tap:

<int:wire-tap pattern="input*, thing2*, thing1" order="3" channel="wiretapChannel"/>

	Tip
	A global wire tap provides a convenient way to configure a single-channel wire tap externally without modifying the existing channel configuration. To do so, set the pattern attribute to the target channel name. For example, you can use this technique to configure a test case to verify messages on a channel.

If namespace support is enabled, two special channels are defined within the application context by default: errorChannel and nullChannel. The nullChannel acts like /dev/null, logging any message sent to it at the DEBUG level and returning immediately. Any time you face channel resolution errors for a reply that you do not care about, you can set the affected component’s output-channel attribute to nullChannel (the name, nullChannel, is reserved within the application context). The errorChannel is used internally for sending error messages and may be overridden with a custom configuration. This is discussed in greater detail in Section E.3, “Error Handling”.

See also Section 11.2, “Message Channels” in the Java DSL chapter for more information about message channel and interceptors.

When Message Endpoints (Channel Adapters) are connected to channels and instantiated, they produce one of the following instances:

The actual implementation depends on the type of channel to which these endpoints connect. A channel adapter connected to a channel that implements the org.springframework.messaging.SubscribableChannel interface produces an instance of EventDrivenConsumer. On the other hand, a channel adapter connected to a channel that implements the org.springframework.messaging.PollableChannel interface (such as a QueueChannel) produces an instance of PollingConsumer.

Polling consumers let Spring Integration components actively poll for Messages rather than process messages in an event-driven manner.

They represent a critical cross-cutting concern in many messaging scenarios. In Spring Integration, polling consumers are based on the pattern with the same name, which is described in the book Enterprise Integration Patterns, by Gregor Hohpe and Bobby Woolf. You can find a description of the pattern on the book’s website.

Spring Integration offers a second variation of the polling consumer pattern. When inbound channel adapters are used, these adapters are often wrapped by a SourcePollingChannelAdapter. For example, when retrieving messages from a remote FTP Server location, the adapter described in Section 18.4, “FTP Inbound Channel Adapter” is configured with a poller to periodically retrieve messages. So, when components are configured with pollers, the resulting instances are of one of the following types:

This means that pollers are used in both inbound and outbound messaging scenarios. Here are some use cases in which pollers are used:

Note

AOP advice classes can be applied to pollers, in an advice-chain, such as a transaction advice to start a transaction. Starting with version 4.1, a PollSkipAdvice is provided. Pollers use triggers to determine the time of the next poll. The PollSkipAdvice can be used to suppress (skip) a poll, perhaps because there is some downstream condition that would prevent the message being processed. To use this advice, you have to provide it with an implementation of a PollSkipStrategy. Starting with version 4.2.5, a SimplePollSkipStrategy is provided. To use it, you can add an instance as a bean to the application context, inject it into a PollSkipAdvice, and add that to the poller’s advice chain. To skip polling, call skipPolls(). To resume polling, call reset(). Version 4.2 added more flexibility in this area. See Section 6.2.4, “Conditional Pollers for Message Sources”.

This chapter is meant to only give a high-level overview of polling consumers and how they fit into the concept of message channels (see Section 6.1, “Message Channels”) and channel adapters (see Section 6.3, “Channel Adapter”). For more information regarding messaging endpoints in general and polling consumers in particular, see Section 10.1, “Message Endpoints”.

Starting with version 5.0.1, certain modules provide MessageSource implementations that support deferring acknowledgment until the downstream flow completes (or hands off the message to another thread). This is currently limited to the AmqpMessageSource and the KafkaMessageSource provided by the spring-kafka-integration extension project, version 3.0.1 or higher.

With these message sources, the IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK header (see Section 7.2.1, “MessageHeaderAccessor API”) is added to the message. The value of the header is an instance of AcknowledgmentCallback, as the following example shows:

@FunctionalInterface
public interface AcknowledgmentCallback {

    void acknlowledge(Status status);

    boolean isAcknowledged();

    void noAutoAck();

    default boolean isAutoAck();

    enum Status {

        /**
         * Mark the message as accepted.
         */
        ACCEPT,

        /**
         * Mark the message as rejected.
         */
        REJECT,

        /**
         * Reject the message and requeue so that it will be redelivered.
         */
        REQUEUE

    }

}

Not all message sources (for example, Kafka) support the REJECT status. It is treated the same as ACCEPT.

Applications can acknowledge a message at any time, as the following example shows:

Message<?> received = source.receive();

...

StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
        .acknowledge(Status.ACCEPT);

If the MessageSource is wired into a SourcePollingChannelAdapter, when the poller thread returns to the adapter after the downstream flow completes, the adapter checks whether the acknowledgment has already been acknowledged and, if not, sets its status to ACCEPT it (or REJECT if the flow throws an exception). The status values are defined in the AcknowledgmentCallback.Status enumeration.

Spring Integration provides MessageSourcePollingTemplate to perform ad-hoc polling of a MessageSource. This, too, takes care of setting ACCEPT or REJECT on the AcknowledgmentCallback when the MessageHandler callback returns (or throws an exception). The following example shows how to poll with the MessageSourcePollingTemplate:

MessageSourcePollingTemplate template =
    new MessageSourcePollingTemplate(this.source);
template.poll(h -> {
    ...
});

In both cases (SourcePollingChannelAdapter and MessageSourcePollingTemplate), you can disable auto ack/nack by calling noAutoAck() on the callback. You might do this if you hand off the message to another thread and wish to acknowledge later. Not all implementations support this (for example, Apache Kafka does not, because the offset commit has to be performed on the same thread).

This section covers how to use conditional pollers.

"Smart" Polling

Version 4.2 introduced the AbstractMessageSourceAdvice. Any Advice objects in the advice-chain that subclass this class are applied only to the receive operation. Such classes implement the following methods:

• beforeReceive(MessageSource<?> source) This method is called before the MessageSource.receive() method. It lets you examine and reconfigure the source. Returning false cancels this poll (similar to the PollSkipAdvice mentioned earlier).
• Message<?> afterReceive(Message<?> result, MessageSource<?> source) This method is called after the receive() method. Again, you can reconfigure the source or take any action (perhaps depending on the result, which can be null if there was no message created by the source). You can even return a different message

	Thread safety
	If an advice mutates the MessageSource, you should not configure the poller with a TaskExecutor. If an advice mutates the source, such mutations are not thread safe and could cause unexpected results, especially with high frequency pollers. If you need to process poll results concurrently, consider using a downstream ExecutorChannel instead of adding an executor to the poller.

Advice Chain Ordering

You should understand how the advice chain is processed during initialization. Advice objects that do not extend AbstractMessageSourceAdvice are applied to the whole poll process and are all invoked first, in order, before any AbstractMessageSourceAdvice. Then AbstractMessageSourceAdvice objects are invoked in order around the MessageSource receive() method. If you have, for example, Advice objects a, b, c, d, where b and d are AbstractMessageSourceAdvice, the objects are applied in the following order: a, c, b, d. Also, if a MessageSource is already a Proxy, the AbstractMessageSourceAdvice is invoked after any existing Advice objects. If you wish to change the order, you must wire up the proxy yourself.

SimpleActiveIdleMessageSourceAdvice

This advice is a simple implementation of AbstractMessageSourceAdvice. When used in conjunction with a DynamicPeriodicTrigger, it adjusts the polling frequency, depending on whether or not the previous poll resulted in a message or not. The poller must also have a reference to the same DynamicPeriodicTrigger.

	Important: Async Handoff
	SimpleActiveIdleMessageSourceAdvice modifies the trigger based on the receive() result. This works only if the advice is called on the poller thread. It does not work if the poller has a task-executor. To use this advice where you wish to use async operations after the result of a poll, do the async handoff later, perhaps by using an ExecutorChannel.

CompoundTriggerAdvice

This advice allows the selection of one of two triggers based on whether a poll returns a message or not. Consider a poller that uses a CronTrigger. CronTrigger instances are immutable, so they cannot be altered once constructed. Consider a use case where we want to use a cron expression to trigger a poll once each hour but, if no message is received, poll once per minute and, when a message is retrieved, revert to using the cron expression.

The advice (and poller) use a CompoundTrigger for this purpose. The trigger’s primary trigger can be a CronTrigger. When the advice detects that no message is received, it adds the secondary trigger to the CompoundTrigger. When the CompoundTrigger instance’s nextExecutionTime method is invoked, it delegates to the secondary trigger, if present. Otherwise, it delegates to the primary trigger.

The poller must also have a reference to the same CompoundTrigger.

The following example shows the configuration for the hourly cron expression with a fallback to every minute:

<int:inbound-channel-adapter channel="nullChannel" auto-startup="false">
    <bean class="org.springframework.integration.endpoint.PollerAdviceTests.Source" />
    <int:poller trigger="compoundTrigger">
        <int:advice-chain>
            <bean class="org.springframework.integration.aop.CompoundTriggerAdvice">
                <constructor-arg ref="compoundTrigger"/>
                <constructor-arg ref="secondary"/>
            </bean>
        </int:advice-chain>
    </int:poller>
</int:inbound-channel-adapter>

<bean id="compoundTrigger" class="org.springframework.integration.util.CompoundTrigger">
    <constructor-arg ref="primary" />
</bean>

<bean id="primary" class="org.springframework.scheduling.support.CronTrigger">
    <constructor-arg value="0 0 * * * *" /> <!-- top of every hour -->
</bean>

<bean id="secondary" class="org.springframework.scheduling.support.PeriodicTrigger">
    <constructor-arg value="60000" />
</bean>

	Important: Async Handoff
	CompoundTriggerAdvice modifies the trigger based on the receive() result. This works only if the advice is called on the poller thread. It does not work if the poller has a task-executor. To use this advice where you wish to use async operations after the result of a poll, do the async handoff later, perhaps by using an ExecutorChannel.

An inbound-channel-adapter element can invoke any method on a Spring-managed object and send a non-null return value to a MessageChannel after converting the method’s output to a Message. When the adapter’s subscription is activated, a poller tries to receive messages from the source. The poller is scheduled with the TaskScheduler according to the provided configuration. To configure the polling interval or cron expression for an individual channel adapter, you can provide a poller element with one of the scheduling attributes, such as fixed-rate or cron. The following example defines two inbound-channel-adapter instances:

<int:inbound-channel-adapter ref="source1" method="method1" channel="channel1">
    <int:poller fixed-rate="5000"/>
</int:inbound-channel-adapter>

<int:inbound-channel-adapter ref="source2" method="method2" channel="channel2">
    <int:poller cron="30 * 9-17 * * MON-FRI"/>
</int:channel-adapter>

See also Section 6.3.3, “Channel Adapter Expressions and Scripts”.

	Note
	If no poller is provided, then a single default poller must be registered within the context. See Section 10.1.4, “Endpoint Namespace Support” for more detail.

Important: Poller Configuration

Some inbound-channel-adapter types are backed by a SourcePollingChannelAdapter, which means they contain a poller configuration that polls the MessageSource (to invoke a custom method that produces the value that becomes a Message payload) based on the configuration specified in the Poller. The following example shows the configuration of two pollers: <int:poller max-messages-per-poll="1" fixed-rate="1000"/> <int:poller max-messages-per-poll="10" fixed-rate="1000"/> In the the first configuration, the polling task is invoked once per poll, and, during each task (poll), the method (which results in the production of the message) is invoked once, based on the max-messages-per-poll attribute value. In the second configuration, the polling task is invoked 10 times per poll or until it returns null, thus possibly producing ten messages per poll while each poll happens at one-second intervals. However, what happens if the configuration looks like the following example: <int:poller fixed-rate="1000"/> Note that there is no max-messages-per-poll specified. As we cover later, the identical poller configuration in the PollingConsumer (for example, service-activator, filter, router, and others) would have a default value of -1 for max-messages-per-poll, which means "execute the polling task non-stop unless the polling method returns null (perhaps because there are no more messages in the `QueueChannel)`" and then sleep for one second. However, in the SourcePollingChannelAdapter, it is a bit different. The default value for max-messages-per-poll is 1, unless you explicitly set it to a negative value (such as -1). This makes sure that the poller can react to lifecycle events (such as start and stop) and prevents it from potentially spinning in an infinite loop if the implementation of the custom method of the MessageSource has a potential to never return null and happens to be non-interruptible. However, if you are sure that your method can return null and you need to poll for as many sources as available per each poll, you should explicitly set max-messages-per-poll to a negative value, as the following example shows: <int:poller max-messages-per-poll="-1" fixed-rate="1000"/>

An outbound-channel-adapter element can also connect a MessageChannel to any POJO consumer method that should be invoked with the payload of messages sent to that channel. The following example shows how to define an outbound channel adapter:

<int:outbound-channel-adapter channel="channel1" ref="target" method="handle"/>

<beans:bean id="target" class="org.MyPojo"/>

If the channel being adapted is a PollableChannel, you must provide a poller sub-element, as the following example shows:

<int:outbound-channel-adapter channel="channel2" ref="target" method="handle">
    <int:poller fixed-rate="3000" />
</int:outbound-channel-adapter>

<beans:bean id="target" class="org.MyPojo"/>

You should use a ref attribute if the POJO consumer implementation can be reused in other <outbound-channel-adapter> definitions. However, if the consumer implementation is referenced by only a single definition of the <outbound-channel-adapter>, you can define it as an inner bean, as the following example shows:

<int:outbound-channel-adapter channel="channel" method="handle">
    <beans:bean class="org.Foo"/>
</int:outbound-channel-adapter>

	Note
	Using both the ref attribute and an inner handler definition in the same <outbound-channel-adapter> configuration is not allowed, as it creates an ambiguous condition. Such a configuration results in an exception being thrown.

Any channel adapter can be created without a channel reference, in which case it implicitly creates an instance of DirectChannel. The created channel’s name matches the id attribute of the <inbound-channel-adapter> or <outbound-channel-adapter> element. Therefore, if channel is not provided, id is required.

Like many other Spring Integration components, the <inbound-channel-adapter> and <outbound-channel-adapter> also provide support for SpEL expression evaluation. To use SpEL, provide the expression string in the expression attribute instead of providing the ref and method attributes that are used for method-invocation on a bean. When an expression is evaluated, it follows the same contract as method-invocation where: the expression for an <inbound-channel-adapter> generates a message any time the evaluation result is a non-null value, while the expression for an <outbound-channel-adapter> must be the equivalent of a void-returning method invocation.

Starting with Spring Integration 3.0, an <int:inbound-channel-adapter/> can also be configured with a SpEL <expression/> (or even with a <script/>) sub-element, for when more sophistication is required than can be achieved with the simple expression attribute. If you provide a script as a Resource by using the location attribute, you can also set refresh-check-delay, which allows the resource to be periodically refreshed. If you want the script to be checked on each poll, you would need to coordinate this setting with the poller’s trigger, as the following example shows:

<int:inbound-channel-adapter ref="source1" method="method1" channel="channel1">
    <int:poller max-messages-per-poll="1" fixed-delay="5000"/>
    <script:script lang="ruby" location="Foo.rb" refresh-check-delay="5000"/>
</int:inbound-channel-adapter>

See also the cacheSeconds property on the ReloadableResourceBundleExpressionSource when using the <expression/> sub-element. For more information regarding expressions, see Appendix A, Spring Expression Language (SpEL). For scripts, see Section 10.8, “Groovy support” and Section 10.7, “Scripting Support”.

Important

The <int:inbound-channel-adapter/> is endpoint starts a message flow by periodically triggering to poll some underlying MessageSource. Since, at the time of polling, there is no message object, expressions and scripts do not have access to a root Message, so there are no payload or headers properties that are available in most other messaging SpEL expressions. The script can generate and return a complete Message object with headers and payload or only a payload, which is added to a message with basic headers.

You can use the <bridge> element is used to create a messaging bridge between two message channels or channel adapters. To do so, provide the input-channel and output-channel attributes, as the following example shows:

<int:bridge input-channel="input" output-channel="output"/>

As mentioned above, a common use case for the messaging bridge is to connect a PollableChannel to a SubscribableChannel. When performing this role, the messaging bridge may also serve as a throttler:

<int:bridge input-channel="pollable" output-channel="subscribable">
     <int:poller max-messages-per-poll="10" fixed-rate="5000"/>
 </int:bridge>

You can use a similar mechanism to connecting channel adapters. The following example shows a simple "echo" between the stdin and stdout adapters from Spring Integration’s stream namespace:

<int-stream:stdin-channel-adapter id="stdin"/>

 <int-stream:stdout-channel-adapter id="stdout"/>

 <int:bridge id="echo" input-channel="stdin" output-channel="stdout"/>

Similar configurations work for other (potentially more useful) Channel Adapter bridges, such as file-to-JMS or mail-to-file. Upcoming chapters cover the various channel adapters.

	Note
	If no output-channel is defined on a bridge, the reply channel provided by the inbound message is used, if available. If neither an output nor a reply channel is available, an exception is thrown.

The following example shows how to configure a bridge in Java by using the @BridgeFrom annotation:

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

@Bean
@BridgeFrom(value = "polled", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "10"))
public SubscribableChannel direct() {
    return new DirectChannel();
}

The following example shows how to configure a bridge in Java by using the @BridgeTo annotation:

@Bean
@BridgeTo(value = "direct", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "10"))
public PollableChannel polled() {
    return new QueueChannel();
}

@Bean
public SubscribableChannel direct() {
    return new DirectChannel();
}

Alternately, you can use a BridgeHandler, as the following example shows:

@Bean
@ServiceActivator(inputChannel = "polled",
        poller = @Poller(fixedRate = "5000", maxMessagesPerPoll = "10"))
public BridgeHandler bridge() {
    BridgeHandler bridge = new BridgeHandler();
    bridge.setOutputChannelName("direct");
    return bridge;
}

You can use the Java Domain Specific Language (DSL) to configure a bridge, as the following example shows:

@Bean
public IntegrationFlow bridgeFlow() {
    return IntegrationFlows.from("polled")
            .bridge(e -> e.poller(Pollers.fixedDelay(5000).maxMessagesPerPoll(10)))
            .channel("direct")
            .get();
}