From 6315a85b6235c4a4a41fa8e0b5606dc8de1f10a5 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 3 Jul 2009 19:52:23 +0000 Subject: [PATCH] INT-676 --- spring-integration-reference/src/channel.xml | 209 ++++++++++++++----- 1 file changed, 154 insertions(+), 55 deletions(-) diff --git a/spring-integration-reference/src/channel.xml b/spring-integration-reference/src/channel.xml index 5e91f88abd..5fc3044fa4 100644 --- a/spring-integration-reference/src/channel.xml +++ b/spring-integration-reference/src/channel.xml @@ -49,8 +49,8 @@ SubscribableChannel The SubscribableChannel base interface is implemented by channels that send - Messages directly to their subscribed handlers. Therefore, they do not provide receive methods for polling, but - instead define methods for handling those subscribers: + 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); @@ -95,10 +95,12 @@ 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. 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. + 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. The no-argument send and receive methods block indefinitely. Note however, that calling the no-arg versions of send() and receive() will block indefinitely. @@ -107,11 +109,11 @@ PriorityChannel 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 + 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. + logic, a comparator of type Comparator<Message<?>> can be provided + to the PriorityChannel's constructor.
@@ -122,47 +124,90 @@ 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 too dangerous. For - example, the sender's thread could roll back a transaction if the send operation times out, whereas with a - QueueChannel, the message would have been stored to the internal queue and potentially - never received. + 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. + + + 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 . 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. + 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 + 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. Its primary purpose is to enable 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. 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) can play a role in determining the ultimate result - of that transaction (commit or rollback). + 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). 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 needs to provide buffering to throttle input, and to modify those to be - queue-based PollableChannels. Likewise, if a channel needs to broadcast + 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 can have one of two dispatcher strategies. These determine how + invocations will be ordered in the case that there are multiple handlers subscribed to the same channel. + The default strategy is "round-robin" and essentially load-balances across the handlers in rotation. The + other strategy is "failover" and it will always try to invoke the first handler, falling back to any + subsequent handlers as necessary. 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. + +
+
+ ExecutorChannel + + The ExecutorChannel is a point-to-point channel that supports + the same dispatcher strategies as DirectChannel. The key difference is that + it 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. + + 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. + +
ThreadLocalChannel @@ -200,15 +245,25 @@ After implementing the interface, registering the interceptor with a channel is just a matter of calling: 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 an Exception). Also, the + 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. + + 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 will be sent directly to one or more subscribers + depending on the type of channel (e.g. a PublishSubscribeChannel sends to all of its subscribers). Therefore, + the preReceive(..) and postReceive(..) interceptor methods + are only invoked when the interceptor is applied to a PollableChannel. + Because it is rarely necessary to implement all of the interceptor methods, a ChannelInterceptorAdapter class is also available for sub-classing. It provides no-op methods (the void method is empty, the Message returning methods - return the Message parameter as-is, and the boolean method returns true). + return the Message as-is, and the boolean method returns true). Therefore, it is often easiest to extend that class and just implement the method(s) that you need as in the following example. - - 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 will be sent directly to one or more subscribers - depending on the type of channel (e.g. a PublishSubscribeChannel sends to all of its subscribers). Therefore, - the preReceive(..) and postReceive(..) interceptor methods - are only invoked when the interceptor is applied to a PollableChannel. - + + The order of invocation for the interceptor methods depends on the type of channel. As described above, + 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 separate + sender and receiver threads. For example, if a receiver is already blocked while waiting for a message + the order could be: preSend, preReceive, postReceive, postSend. However, if a receiver polls after the + sender has placed a message on the channel and already returned, the order would be: 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!). + Obviously, the type of queue also plays a role (e.g. rendezvous vs. priority). The bottom line is that + you cannot rely on the order beyond the fact that preSend will precede postSend and preReceive will + precede postReceive. +
@@ -256,6 +314,12 @@ public Message sendAndReceive(final Message request, final MessageChannel public Message receive(final PollableChannel channel) { ... }]]> + + + A less invasive approach that allows you to invoke simple interfaces with payload and/or header + values instead of Message instances is described in . + +
@@ -285,14 +349,15 @@ public Message receive(final PollableChannel channel) { ... }]]>SubscribableChannel). - However, you can also provide a variety of "queue" sub-elements to create the channel types (as described in + However, you can alternatively provide a variety of "queue" sub-elements to create any of + the pollable channel types (as described in ). Examples of each are shown below.
DirectChannel Configuration - As mentioned above, DirectChannel is the default type. - ]]> + As mentioned above, DirectChannel is the default type. + ]]>
@@ -300,12 +365,12 @@ public Message receive(final PollableChannel channel) { ... }]]> To create a QueueChannel, use the "queue" sub-element. You may specify the channel's capacity: - <channel id="exampleChannel"> + <channel id="queueChannel"> <queue capacity="25"/> </channel> 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's highly + 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. @@ -316,21 +381,41 @@ public Message receive(final PollableChannel channel) { ... }]]>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="exampleChannel" task-executor="someTaskExecutor"/> + <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 - for the channel. That will indicate that the channel should set the sequence-size and sequence-number - Message headers 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. This value is 'false' by default. - <publish-subscribe-channel id="exampleChannel" apply-sequence="true"/> + 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"/> + + 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. + + +
+
+ ExecutorChannel + + To create an ExecutorChannel, add the '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" task-executor="someExecutor"/>
PriorityChannel Configuration To create a PriorityChannel, use the "priority-queue" sub-element: - + ]]> By default, the channel will consult the MessagePriority header of the @@ -338,7 +423,7 @@ public Message receive(final PollableChannel channel) { ... }]]>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: - + @@ -349,8 +434,10 @@ public Message receive(final PollableChannel channel) { ... }]]>RendezvousChannel Configuration A RendezvousChannel is created when the queue sub-element is - a <rendezvous-queue>. It does not provide any additional configuration options. - + 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. + ]]> @@ -360,7 +447,7 @@ public Message receive(final PollableChannel channel) { ... }]]>ThreadLocalChannel Configuration The ThreadLocalChannel does not provide any additional configuration options. - ]]> + ]]>
@@ -376,5 +463,17 @@ public Message receive(final PollableChannel channel) { ... }]]> + + + If namespace support is enabled, there are also two special channels defined within the context by default: + errorChannel and nullChannel. The 'nullChannel' acts like /dev/null, + simply logging any Message sent to it at DEBUG level and returning immediately. Any time you face channel + resolution errors for a reply that you don't care about, you can set the affected component's 'output-channel' + to reference 'nullChannel' (the name 'nullChannel' is reserved within the context). The 'errorChannel' is + used internally for sending error messages, and it can be overridden with a custom configuration. It is + discussed in greater detail in . + +
+ \ No newline at end of file