Channel Adapter

A channel adapter is a message endpoint that enables connecting a single sender or receiver to a message channel. Spring Integration provides a number of adapters to support various transports, such as JMS, file, HTTP, web services, mail, and more. Upcoming chapters of this reference guide discuss each adapter. However, this chapter focuses on the simple but flexible method-invoking channel adapter support. There are both inbound and outbound adapters, and each may be configured with XML elements provided in the core namespace. These provide an easy way to extend Spring Integration, as long as you have a method that can be invoked as either a source or a destination.

Configuring An Inbound Channel Adapter

An inbound-channel-adapter element (a SourcePollingChannelAdapter in Java configuration) 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:

Java DSL
@Bean
public IntegrationFlow source1() {
    return IntegrationFlows.from(() -> new GenericMessage<>(...),
                             e -> e.poller(p -> p.fixedRate(5000)))
                ...
                .get();
}

@Bean
public IntegrationFlow source2() {
    return IntegrationFlows.from(() -> new GenericMessage<>(...),
                             e -> e.poller(p -> p.cron("30 * 9-17 * * MON-FRI")))
                ...
                .get();
}
Java
public class SourceService {

    @InboundChannelAdapter(channel = "channel1", poller = @Poller(fixedRate = "5000"))
    Object method1() {
        ...
    }

    @InboundChannelAdapter(channel = "channel2", poller = @Poller(cron = "30 * 9-17 * * MON-FRI"))
    Object method2() {
        ...
    }
}
Kotlin DSL
@Bean
fun messageSourceFlow() =
    integrationFlow( { GenericMessage<>(...) },
                    { poller { it.fixedRate(5000) } }) {
        ...
    }
XML
<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 Channel Adapter Expressions and Scripts.

If no poller is provided, then a single default poller must be registered within the context. See Endpoint Namespace Support for more detail.
Important: Poller Configuration

All the 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 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"/>

Starting with version 5.5, a 0 value for max-messages-per-poll has a special meaning - skip the MessageSource.receive() call altogether, which may be considered as pausing for this inbound channel adapter until the maxMessagesPerPoll is changed to a non-zero value at a later time, e.g. via a Control Bus.

Also see Global Default Poller for more information.

Configuring An Outbound Channel Adapter

An outbound-channel-adapter element (a @ServiceActivator for Java configuration) 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:

Java DSL
@Bean
public IntegrationFlow outboundChannelAdapterFlow(MyPojo myPojo) {
    return f -> f
             .handle(myPojo, "handle");
}
Java
public class MyPojo {

    @ServiceActivator(channel = "channel1")
    void handle(Object payload) {
        ...
    }

}
Kotlin DSL
@Bean
fun outboundChannelAdapterFlow(myPojo: MyPojo) =
    integrationFlow {
        handle(myPojo, "handle")
    }
XML
<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 (the @Poller sub-annotation on the @ServiceActivator), as the following example shows:

Java
public class MyPojo {

    @ServiceActivator(channel = "channel1", poller = @Poller(fixedRate = "3000"))
    void handle(Object payload) {
        ...
    }

}
XML
<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>
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.

Channel Adapter Expressions and Scripts

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 Spring Expression Language (SpEL). For scripts, see Groovy support and Scripting Support.

The <int:inbound-channel-adapter/> (SourcePollingChannelAdapter) is an endpoint which 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 by the framework.