Since Message Channels may or may not buffer Messages (as discussed in the overview), there are two sub-interfaces defining the buffering (pollable) and non-buffering (subscribable) channel behavior. Here is the definition of PollableChannel.

public interface PollableChannel extends MessageChannel {

    Message<?> receive();

    Message<?> receive(long timeout);

    List<Message<?>> clear();

    List<Message<?>> purge(MessageSelector selector);

}

Similar to the send methods, when receiving a message, the return value will be null in the case of a timeout or interrupt.

The SubscribableChannel base interface is implemented by channels that send Messages directly to their subscribed MessageHandlers. Therefore, they do not provide receive methods for polling, but instead define methods for managing those subscribers:

public interface SubscribableChannel extends MessageChannel {

    boolean subscribe(MessageHandler handler);

    boolean unsubscribe(MessageHandler handler);

}

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 be a MessageHandler itself, and the subscriber's handleMessage(Message) method will be invoked in turn.

The QueueChannel implementation wraps a queue. Unlike the PublishSubscribeChannel, the QueueChannel has point-to-point semantics. In other words, even if the channel has multiple consumers, only one of them should receive any Message sent to that channel. It provides a default no-argument constructor (providing an essentially unbounded capacity of Integer.MAX_VALUE) as well as a constructor that accepts the queue capacity:

public QueueChannel(int capacity)

A channel that has not reached its capacity limit will store messages in its internal queue, and the send() method will return immediately even if no receiver is ready to handle the message. If the queue has reached capacity, then the sender will block until room is available. Or, if using the send call that accepts a timeout, it will block until either room is available or the timeout period elapses, whichever occurs first. Likewise, a receive call will return immediately if a message is available on the queue, but if the queue is empty, then a receive call may block until either a message is available or the timeout elapses. In either case, it is possible to force an immediate return regardless of the queue's state by passing a timeout value of 0. Note however, that calls to the no-arg versions of send() and receive() will block indefinitely.

Whereas the QueueChannel enforces first-in/first-out (FIFO) ordering, the PriorityChannel is an alternative implementation that allows for messages to be ordered within the channel based upon a priority. By default the priority is determined by the 'priority' header within each message. However, for custom priority determination logic, a comparator of type Comparator<Message<?>> can be provided to the PriorityChannel's constructor.

The RendezvousChannel enables a "direct-handoff" scenario where a sender will block until another party invokes the channel's receive() method or vice-versa. 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 are operating in different threads but simply dropping the message in a queue asynchronously is not appropriate. In other words, with a RendezvousChannel at least 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. When persistence is required, you can either invoke a database operation within a handler or use Spring Integration's support for JMS-based Channel Adapters. The latter option allows you to take advantage of any JMS provider's implementation for message persistence, and it will be discussed in Chapter 22, JMS Support. However, when buffering in a queue is not necessary, the simplest approach is to rely upon the DirectChannel discussed next.

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.

The DirectChannel has point-to-point semantics but otherwise is more similar to the PublishSubscribeChannel than any of the queue-based channel implementations described above. 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 will only send 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 is subscribed to a DirectChannel, then sending a Message to that channel will trigger 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, then the outcome of the handler's invocation (e.g. updating a database record) will play 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 and then to consider which of those need to provide buffering or to throttle input, and then 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. Below you will see how each of these 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. The load-balancer determines how invocations will be ordered in the case that there are multiple handlers subscribed to the same channel. When using the namespace support described below, the default strategy is "round-robin" which essentially load-balances across the handlers in rotation.

	Note
	The "round-robin" strategy is currently the only implementation available out-of-the-box in Spring Integration. Other strategy implementations may be added in future versions.

The load-balancer also works in combination with a boolean failover property. If the "failover" value is true (the default), then the dispatcher will fall 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 are subscribed.

If a certain situation requires that the dispatcher always try to invoke the first handler, then fallback 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 will always begin 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 will determine that order.

	Note
	Keep in mind that load-balancing and failover only apply 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 in the "input-channel" attribute.

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 will not block, but it also means that the handler invocation may not occur in the sender's thread. It therefore does not support transactions spanning the sender and receiving handler.

Tip

Note that there are occasions where the sender may block. For example, when using a TaskExecutor with a rejection-policy that throttles back on the client (such as the ThreadPoolExecutor.CallerRunsPolicy), the sender's thread will execute the method directly anytime 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, that obviously cannot be relied upon for transactions.

The final channel implementation type is ThreadLocalChannel. This channel also delegates to a queue internally, but the queue is bound to the current thread. That way the thread that sends to the channel will later be able to receive those same Messages, but no other thread would be able to access them. While probably the least common type of channel, this is useful for situations where DirectChannels are being used to enforce a single thread of operation but any reply Messages should be sent to a "terminal" channel. If that terminal channel is a ThreadLocalChannel, the original sending thread can collect its replies from it.

As mentioned above, DirectChannel is the default type.

<channel id="directChannel"/>

A default channel will have a round-robin load-balancer and will also have failover enabled (See the discussion in Section 3.2.5, “DirectChannel” for more detail). To disable one or both of these, add a <dispatcher/> sub-element and configure the attributes:

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

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

To create a QueueChannel, use the "queue" sub-element. You may specify the channel's capacity:

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

	Note
	If you do not provide a value for the 'capacity' attribute on this <queue/> sub-element, the resulting queue will be unbounded. To avoid issues such as OutOfMemoryErrors, it is highly recommended to set an explicit value for a bounded queue.

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 simply publishes in the sender's thread):

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

If you are providing a Resequencer or Aggregator downstream from a PublishSubscribeChannel, then you can set the 'apply-sequence' property on the channel to true. That will indicate that the channel should set the sequence-size and sequence-number Message headers as well as the correlation id prior to passing the Messages along. For example, if there are 5 subscribers, the sequence-size would be set to 5, and the Messages would have sequence-number header values ranging from 1 to 5.

<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, the channel creates new Message instances with the same payload reference but different header values when the flag is set to true.

To create an ExecutorChannel, add the <dispatcher> sub-element along with a 'task-executor' attribute. Its value can reference any TaskExecutor within the context. For example, this enables configuration of a thread-pool for dispatching messages to subscribed handlers. As mentioned above, this does break the "single-threaded" execution context between sender and receiver so that any active transaction context will not be shared by the invocation of the handler (i.e. the handler may throw an Exception, but the send invocation has already returned successfully).

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

Note

The "load-balancer" and "failover" options are also both available on the dispatcher sub-element as described above in Section 3.5.1, “DirectChannel Configuration”. The same defaults apply as well. So, the channel will have a round-robin load-balancing strategy with failover enabled unless explicit configuration is provided for one or both of those attributes. <channel id="executorChannelWithoutFailover"> <dispatcher task-executor="someExecutor" failover="false"/> </channel>

To create a PriorityChannel, use the "priority-queue" sub-element:

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

By default, the channel will consult the MessagePriority header of the message. However, a custom Comparator reference may be provided instead. Also, note that the PriorityChannel (like the other types) does support the "datatype" attribute. As with the QueueChannel, it also supports a "capacity" attribute. The following example demonstrates all of these:

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

A RendezvousChannel is created when the queue sub-element is a <rendezvous-queue>. It does not provide any additional configuration options to those described above, and its queue does not accept any capacity value since it is a 0-capacity direct handoff queue.

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

The ThreadLocalChannel does not provide any additional configuration options.

<thread-local-channel id="threadLocalChannel"/>

Message channels may also have interceptors as described in Section 3.3, “Channel Interceptors”. The <interceptors> sub-element can be added within <channel> (or the more specific element types). Provide the "ref" attribute to reference any Spring-managed object that implements the ChannelInterceptor interface:

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

In general, it is a good idea to define the interceptor implementations in a separate location since they usually provide common behavior that can be reused across multiple channels.

Channel Interceptors allow you for a clean and concise way of applying cross-cutting behavior per individual channel. But what 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. The better way would be to configure interceptors globally and apply them on multiple channels in one shot. Spring Integration provides capabilities to configure Global Interceptors and apply them on multiple channels. Look at the example below:

<int:channel-interceptor pattern="input*, bar*, foo" order="3">
    <bean class="foo.barSampleInterceptor"/>
</int:channel-interceptor>

or

<int:channel-interceptor ref="myInterceptor" pattern="input*, bar*, foo" order="3"/>

<bean id="myInterceptor" class="foo.barSampleInterceptor"/>

<channel-interceptor> element allows you to define a global interceptor which will be applied on all channels that match patterns defined via pattern attribute. In the above case the global interceptor will be applied on 'foo' channel and all other channels that begin with 'bar' and 'input'. The order attribute allows you to manage the place where this interceptor will be injected. For example, channel 'inputChannel' could have individual interceptors configured locally (see below):

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

The reasonable question would be how global interceptor will be injected in relation to other interceptors configured locally or through other global interceptor definitions? Current implementation provides a very simple and clever mechanism of handling this. Positive number in the order attribute will ensure interceptor injection after existing interceptors and negative number will ensure that such interceptors injected before. This means that in the above example global interceptor will be injected AFTER (since its order is greater then 0) 'wire-tap' interceptor configured locally. If there was another global interceptor with matching pattern their order would be determined based on who's got the higher or lower value in order attribute. To inject global interceptor BEFORE the existing interceptors use negative value for the order attribute.

	Note
	Note that order and pattern attributes are optional. The default value for order will be 0 and for pattern is '*'

As mentioned above, Spring Integration provides a simple Wire Tap interceptor out of the box. You can configure a Wire Tap on any channel within an 'interceptors' element. This is especially useful for debugging, and can be used in conjunction with Spring Integration's logging Channel Adapter as follows:

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

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

	Tip
	The 'logging-channel-adapter' also accepts a boolean attribute: 'log-full-message'. That is false by default so that only the payload is logged. Setting that to true enables logging of all headers in addition to the payload.