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, etc.) may be configured as Message header values by an upstream component. Developers can also store any arbitrary key-value pairs in the headers.

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.

A Message Channel may follow either Point-to-Point or Publish/Subscribe semantics. With a Point-to-Point channel, at most one consumer can receive each Message sent to the channel. Publish/Subscribe channels, on the other hand, will attempt to broadcast each Message to all of its subscribers. Spring Integration supports both of these.

Whereas "Point-to-Point" and "Publish/Subscribe" define the two options for how many consumers will 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. The variety of channel implementations available in Spring Integration will be discussed in detail in Section 4.1.2, “Message Channel Implementations”.

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 will necessarily connect your existing application code directly. Any real-world enterprise integration solution will require some amount of code focused upon integration concerns such as routing and transformation. The important thing is to achieve separation of concerns between such integration logic and business logic. In other words, as with the Model-View-Controller 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 will provide an overview of the Message Endpoint types that handle these responsibilities, and in upcoming chapters, you will 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 (e.g. from XML Document to java.lang.String). Similarly, a transformer may be used to 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, etc. If the Message is accepted, it is sent to the output channel, but if not it will be 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 filter to narrow down the set of Messages to be processed based on some criteria.

A Message Router is responsible for deciding what channel or channels should receive the Message next (if any). Typically the decision is based upon the Message’s content and/or 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 above.

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 then 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 sub-divided 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 or to discard 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.

The Service Activator invokes an operation on some service object to process the request Message, extracting the request Message’s payload and converting if necessary (if the method does not expect a Message-typed parameter). Whenever the service object’s method returns a value, that return value will likewise be converted to a reply Message if necessary (if it’s not already a Message). That reply Message is sent to the output channel. If no output channel has been configured, then the reply will be 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.

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 will do some mapping between the Message and whatever object or resource is received-from or sent-to the other system (File, HTTP Request, JMS Message, etc). Depending on the transport, the Channel Adapter may also populate or extract Message header values. Spring Integration provides a number of Channel Adapters, and they will be described in upcoming chapters.

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

When configuring with XML:

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

the bean names will be as follows:

When using EIP annotations, the names depend on several factors.

When Annotating POJO Methods

@Component
public class SomeComponent {

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

}

the bean names will be as follows:

Starting with version 5.0.4, these names can be modified using the @EndpointId annotation:

@Component
public class SomeComponent {

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

}

the bean names will be as follows:

i.e. @EndpointId creates names as created by the id attribute with XML configuration.

When Annotating @Beans

@Configuratiom
public class SomeConfiguration {

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

}

the bean names will be as follows:

Starting with version 5.0.4, these names can be modified using the @EndpointId annotation:

@Configuratiom
public class SomeConfiguration {

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

}

i.e. @EndpointId 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 as well as emitting 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’s the normal generated handler name).

Message Sources

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

When configuring with XML:

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

the bean names will be as follows:

Using @EndpointId with Java configuration:

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

the bean names will be as follows:

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

the bean names will be as follows:

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) (or eclipse with the Spring IDE plugins) or IntelliJ IDEA, for example. These IDEs know how to resolve the correct XML schema from the classpath (using the META-INF/spring.schemas file in the jar(s)). When using STS, or eclipse with the plugin, be sure to 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 will likely see false errors.

Each of these online schemas has a warning similar to this:

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 & Annotation 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 3.3, “Main Components” above). Their implementations (contracts) are:

The first two are simple enough to understand how to implement, configure and use, respectively; the last one deserves more review.

The AbstractEndpoint is widely used throughout the Framework for different component implementations; its main implementations are:

Using Messaging Annotations and/or Java DSL, you shouldn’t worry about these components, because the Framework produces them automatically via appropriate annotations and BeanPostProcessor s. When building components manually, the ConsumerEndpointFactoryBean should be used to help to 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, e.g. AggregatingMessageHandler, MessageTransformingHandler, AbstractMessageSplitter etc.; as well as the target protocol outbound adapters are implementations too, e.g. FileWritingMessageHandler, HttpRequestExecutingMessageHandler, AbstractMqttMessageHandler etc. When you develop Spring Integration applications with Java & Annotation Configuration, you should take a look into the Spring Integration module to find an appropriate MessageHandler implementation to be used for the @ServiceActivator configuration. For example to send an XMPP message (see Chapter 38, XMPP Support) we should configure something like this:

@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 to 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, e.g. AbstractMqttMessageDrivenChannelAdapter and ImapIdleChannelAdapter; and request-reply - MessagingGatewaySupport implementations, e.g. AmqpInboundGateway and AbstractWebServiceInboundGateway.

Polling inbound endpoints are for those protocols which don’t provide a listener API or aren’t intended for such a behavior. For example any File based protocol, as an FTP, any data bases (RDBMS or NoSQL) etc.

These inbound endpoints consist with two components: the poller configuration, to initiate the polling task periodically, and 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 shouldn’t 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, e.g. MongoDbMessageSource and AbstractTwitterMessageSource. With that in mind, our config for reading data from an RDBMS table with JDBC may look like:

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

All the required inbound and outbound classes for the target protocols you can find in the particular Spring Integration module, in most cases in the respective package. For example 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, for example:

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