Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added 'endpoint' chapter and created sections within the 'channel' chapter

Browse Source
This commit is contained in:
Mark Fisher
2008-10-21 00:11:00 +00:00
parent 40e3dbe583
commit d2d06c459c
4 changed files with 143 additions and 426 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-integration-reference/src/channel.xml
View File

@@ -22,11 +22,14 @@
When sending a message, the return value will be <emphasis>true</emphasis> if the message is sent successfully.
If the send call times out or is interrupted, then it will return <emphasis>false</emphasis>.
</para>
<para>
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 <interfacename>PollableChannel</interfacename>.
<programlisting language="java">public interface PollableChannel extends MessageChannel {
<section id="channel-interfaces-pollablechannel">
<title>PollableChannel</title>
<para>
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 <interfacename>PollableChannel</interfacename>.
<programlisting language="java">public interface PollableChannel extends MessageChannel {
Message&lt;?&gt; receive();
@@ -37,21 +40,26 @@
List&lt;Message&lt;?&gt;&gt; purge(MessageSelector selector);
}</programlisting>
Similar to the send methods, when receiving a message, the return value will be <emphasis>null</emphasis> in the
case of a timeout or interrupt.
</para>
<para>
The <interfacename>SubscribableChannel</interfacename> base interface is implemented by channels that send
Messages directly to their subscribed consumers. Therefore, they do not provide receive methods for polling, but
instead define methods for handling those subscribers:
<programlisting language="java">public interface SubscribableChannel extends MessageChannel {
Similar to the send methods, when receiving a message, the return value will be <emphasis>null</emphasis> in the
case of a timeout or interrupt.
</para>
</section>
<section id="channel-interfaces-subscribablechannel">
<title>SubscribableChannel</title>
<para>
The <interfacename>SubscribableChannel</interfacename> base interface is implemented by channels that send
Messages directly to their subscribed consumers. Therefore, they do not provide receive methods for polling, but
instead define methods for handling those subscribers:
<programlisting language="java">public interface SubscribableChannel extends MessageChannel {
boolean subscribe(MessageConsumer consumer);
boolean unsubscribe(MessageConsumer consumer);
}</programlisting>
</para>
</para>
</section>
</section>
<section id="channel-implementations">

411
spring-integration-reference/src/core-api.xml
View File

@@ -1,371 +1,4 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE book PUBLIC "-//OASIS//DTD DocBook XML V4.5//EN" "http://www.oasis-open.org/docbook/xml/4.5/docbookx.dtd">
<chapter id="api">
<title>The Core API</title>
<section id="api-message">
<title>Message</title>
<para>
The Spring Integration <interfacename>Message</interfacename> is a generic container for data. Any object can
be provided as the payload, and each <interfacename>Message</interfacename> also includes headers containing
user-extensible properties as key-value pairs. Here is the definition of the
<interfacename>Message</interfacename> interface:
<programlisting language="java">public interface Message&lt;T&gt; {
T getPayload();
MessageHeaders getHeaders();
}</programlisting>
And the following headers are pre-defined:
<table id="api-message-headerproperties">
<title>Pre-defined Message Headers</title>
<tgroup cols="2">
<colspec align="left" />
<thead>
<row>
<entry align="center">Header Name</entry>
<entry align="center">Header Type</entry>
</row>
</thead>
<tbody>
<row>
<entry>ID</entry>
<entry>java.util.UUID</entry>
</row>
<row>
<entry>TIMESTAMP</entry>
<entry>java.lang.Long</entry>
</row>
<row>
<entry>EXPIRATION_DATE</entry>
<entry>java.lang.Long</entry>
</row>
<row>
<entry>CORRELATION_ID</entry>
<entry>java.lang.Object</entry>
</row>
<row>
<entry>REPLY_CHANNEL</entry>
<entry>java.lang.Object (can be a String or MessageChannel)</entry>
</row>
<row>
<entry>SEQUENCE_NUMBER</entry>
<entry>java.lang.Integer</entry>
</row>
<row>
<entry>SEQUENCE_SIZE</entry>
<entry>java.lang.Integer</entry>
</row>
<row>
<entry>PRIORITY</entry>
<entry>MessagePriority (an <emphasis>enum</emphasis>)</entry>
</row>
</tbody>
</tgroup>
</table>
</para>
<para>
Many inbound/outbound adapter implementations will also provide and/or expect certain headers, and additional
user-defined headers can also be configured.
</para>
<para>
The base implementation of the <interfacename>Message</interfacename> interface is
<classname>GenericMessage&lt;T&gt;</classname>, and it provides two constructors:
<programlisting language="java">new GenericMessage&lt;T&gt;(T payload);
new GenericMessage&lt;T&gt;(T payload, Map&lt;String, Object&gt; headers)</programlisting>
When a Message is created, a random unique id will be generated. The constructor that accepts a Map of headers
will copy the provided headers to the newly created Message. There are also two convenient subclasses available:
<classname>StringMessage</classname> and <classname>ErrorMessage</classname>. The latter accepts any
<classname>Throwable</classname> object as its payload.
</para>
<para>
You may notice that the Message interface defines retrieval methods for its payload and headers but no setters.
The reason for this is that a Message cannot be modified after its initial creation. Therefore, when a Message
instance is sent to multiple consumers (e.g. through a Publish Subscribe Channel), if one of those consumers
needs to send a reply with a different payload type, it will need to create a new Message. As a result, the
other consumers are not affected by those changes. Keep in mind, that multiple consumers may access the same
payload instance or header value, and whether such an instance is itself immutable is a decision left to the
developer. In other words, the contract for Messages is similar to that of an
<emphasis>unmodifiable Collection</emphasis>, and the MessageHeaders' map further exemplifies that; even though
the MessageHeaders class implements <interfacename>java.util.Map</interfacename>, any attempt to invoke a
<emphasis>put</emphasis> operation (or 'remove' or 'clear') on the MessageHeaders will result in an
<classname>UnsupportedOperationException</classname>.
</para>
<para>
Rather than requiring the creation and population of a Map to pass into the GenericMessage constructor, Spring
Integration does provide a far more convenient way to construct Messages: <classname>MessageBuilder</classname>.
The MessageBuilder provides two factory methods for creating Messages from either an existing Message or with a
payload Object. When building from an existing Message, the headers <emphasis>and payload</emphasis> of that
Message will be copied to the new Message:
<programlisting language="java">Message&lt;String&gt; message1 = MessageBuilder.withPayload("test")
.setHeader("foo", "bar")
.build();
Message&lt;String&gt; message2 = MessageBuilder.fromMessage(message1).build();
assertEquals("test", message2.getPayload());
assertEquals("bar", message2.getHeaders().get("foo"));</programlisting>
</para>
<para>
If you need to create a Message with a new payload but still want to copy the
headers from an existing Message, you can use one of the 'copy' methods.
<programlisting language="java">Message&lt;String&gt; message3 = MessageBuilder.fromPayload("test3")
.copyHeaders(message1.getHeaders())
.build();
Message&lt;String&gt; message4 = MessageBuilder.fromPayload("test4")
.setHeader("foo", 123)
.copyHeadersIfAbsent(message1.getHeaders())
.build();
assertEquals("bar", message3.getHeaders().get("foo"));
assertEquals(123, message4.getHeaders().get("foo"));</programlisting>
Notice that the <methodname>copyHeadersIfAbsent</methodname> does not overwrite existing values. Also, in the
second example above, you can see how to set any user-defined header with <methodname>setHeader</methodname>.
Finally, there are set methods available for the predefined headers as well as a non-destructive method for
setting any header (MessageHeaders also defines constants for the pre-defined header names).
<programlisting language="java">Message&lt;Integer&gt; importantMessage = MessageBuilder.fromPayload(99)
.setPriority(MessagePriority.HIGHEST)
.build();
assertEquals(MessagePriority.HIGHEST, importantMessage.getHeaders().getPriority());
Message&lt;Integer&gt; anotherMessage = MessageBuilder.fromMessage(importantMessage)
.setHeaderIfAbsent(MessageHeaders.PRIORITY, MessagePriority.LOW)
.build();
assertEquals(MessagePriority.HIGHEST, anotherMessage.getHeaders().getPriority());
</programlisting>
</para>
<para>
The <classname>MessagePriority</classname> is only considered when using a <classname>PriorityChannel</classname>
(as described in the next section). It is defined as an <emphasis>enum</emphasis> with five possible values:
<programlisting language="java">public enum MessagePriority {
HIGHEST,
HIGH,
NORMAL,
LOW,
LOWEST
}</programlisting>
</para>
<para>
The <interfacename>Message</interfacename> is obviously a very important part of the API. By encapsulating the
data in a generic wrapper, the messaging system can pass it around without any knowledge of the data's type. As
an application evolves to support new types, or when the types themselves are modified and/or extended, the
messaging system will not be affected by such changes. On the other hand, when some component in the messaging
system <emphasis>does</emphasis> require access to information about the <interfacename>Message</interfacename>,
such metadata can typically be stored to and retrieved from the metadata in the Message Headers.
</para>
</section>
<section id="api-messagechannel">
<title>MessageChannel</title>
<para>
While the <interfacename>Message</interfacename> plays the crucial role of encapsulating data, it is the
<interfacename>MessageChannel</interfacename> that decouples message producers from message consumers.
Spring Integration's top-level <interfacename>MessageChannel</interfacename> interface is defined as follows.
<programlisting language="java"><![CDATA[public interface MessageChannel {
String getName();
boolean send(Message message);
boolean send(Message message, long timeout);
}]]></programlisting>
When sending a message, the return value will be <emphasis>true</emphasis> if the message is sent successfully.
If the send call times out or is interrupted, then it will return <emphasis>false</emphasis>.
</para>
<para>
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 <interfacename>PollableChannel</interfacename>.
<programlisting language="java">public interface PollableChannel extends MessageChannel {
Message&lt;?&gt; receive();
Message&lt;?&gt; receive(long timeout);
List&lt;Message&lt;?&gt;&gt; clear();
List&lt;Message&lt;?&gt;&gt; purge(MessageSelector selector);
}</programlisting>
Similar to the send methods, when receiving a message, the return value will be <emphasis>null</emphasis> in the
case of a timeout or interrupt.
</para>
<para>
The <interfacename>SubscribableChannel</interfacename> base interface is implemented by channels that send
Messages directly to their subscribed consumers. Therefore, they do not provide receive methods for polling, but
instead define methods for handling those subscribers:
<programlisting language="java">public interface SubscribableChannel extends MessageChannel {
boolean subscribe(MessageConsumer consumer);
boolean unsubscribe(MessageConsumer consumer);
}</programlisting>
</para>
<para>
Spring Integration provides several different Message Channel implementations. Each is briefly described in the
sections below.
</para>
<section id="api-messagechannel-publishsubscribechannel">
<title>PublishSubscribeChannel</title>
<para>
The <classname>PublishSubscribeChannel</classname> implementation broadcasts any Message
sent to it to all of its subscribed consumers. This is most often used for sending
<emphasis>Event Messages</emphasis> whose primary role is notification as opposed to
<emphasis>Document Messages</emphasis> which are generally intended to be processed by
a single consumer. Note that the <classname>PublishSubscribeChannel</classname> is
intended for sending only. Since it broadcasts to its subscribers directly when its
<methodname>send(Message)</methodname> method is invoked, consumers cannot poll for
Messages (it does not implement <interfacename>PollableChannel</interfacename> and
therefore has no <methodname>receive()</methodname> method). Instead, any subscriber
must be a <interfacename>MessageConsumer</interfacename> itself, and the subscriber's
<methodname>send(Message)</methodname> method will be invoked in turn.
</para>
</section>
<section id="api-messagechannel-queuechannel">
<title>QueueChannel</title>
<para>
The <classname>QueueChannel</classname> implementation wraps a queue. Unlike, the
<classname>PublishSubscribeChannel</classname>, the <classname>QueueChannel</classname> 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 <code>Integer.MAX_VALUE</code>) as well as a constructor that accepts the queue capacity:
<programlisting language="java">public QueueChannel(int capacity)</programlisting>
A channel that has not reached its capacity limit will store messages in its internal queue, and the
<methodname>send()</methodname> 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.
Note however, that calling the no-arg versions of <methodname>send()</methodname> and
<methodname>receive()</methodname> will block indefinitely.
</para>
</section>
<section id="api-messagechannel-prioritychannel">
<title>PriorityChannel</title>
<para>
Whereas the <classname>QueueChannel</classname> enforces first-in/first-out (FIFO) ordering, the
<classname>PriorityChannel</classname> 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
'<literal>priority</literal>' header within each message. However, for custom priority determination
logic, a comparator of type <classname>Comparator&lt;Message&lt;?&gt;&gt;</classname> can be provided to the
<classname>PriorityChannel</classname>'s constructor.
</para>
</section>
<section id="api-messagechannel-rendezvouschannel">
<title>RendezvousChannel</title>
<para>
The <classname>RendezvousChannel</classname> enables a "direct-handoff" scenario where a sender will block
until another party invokes the channel's <methodname>receive()</methodname> method or vice-versa. Internally,
this implementation is quite similar to the <classname>QueueChannel</classname> except that it uses a
<classname>SynchronousQueue</classname> (a zero-capacity implementation of
<interfacename>BlockingQueue</interfacename>). 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
<classname>QueueChannel</classname>, the message would have been stored to the internal queue and potentially
never received.
</para>
<para>
The <classname>RendezvousChannel</classname> is also useful for implementing request-reply
operations. The sender can create a temporary, anonymous instance of <classname>RendezvousChannel</classname>
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.
</para>
</section>
<section id="api-messagechannel-directchannel">
<title>DirectChannel</title>
<para>
The <classname>DirectChannel</classname> has point-to-point semantics, but otherwise is more similar to the
<classname>PublishSubscribeChannel</classname> than any of the queue-based channel implementations described
above. It implements the <interfacename>SubscribableChannel</interfacename> interface instead of the
<interfacename>PollableChannel</interfacename> interface, so it dispatches Messages directly to a subscriber.
As a point-to-point channel, however, it differs from the <classname>PublishSubscribeChannel</classname> in
that it will only send each Message to a <emphasis>single</emphasis> subscribed
<classname>MessageConsumer</classname>. Its primary purpose is to enable a single thread to perform the
operations on "both sides" of the channel. For example, if a consumer is subscribed to a
<classname>DirectChannel</classname>, then sending a Message to that channel will trigger invocation of that
consumer's <methodname>onMessage(Message)</methodname> method <emphasis>directly in the sender's
thread</emphasis>. 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 consumer's invocation (e.g. updating a database record) can play a role in determining the ultimate result
of that transaction (commit or rollback).
<note>
Since the <classname>DirectChannel</classname> 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 <interfacename>PollableChannels</interfacename>. Likewise, if a channel needs to broadcast
messages, it should not be a <classname>DirectChannel</classname> but rather a
<classname>PublishSubscribeChannel</classname>. Below you will see how these can be configured.
</note>
</para>
</section>
<section id="api-messagechannel-threadlocalchannel">
<title>ThreadLocalChannel</title>
<para>
The final channel implementation type is <classname>ThreadLocalChannel</classname>. 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
<classname>DirectChannels</classname> 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
<classname>ThreadLocalChannel</classname>, the original sending thread can collect its replies from it.
</para>
</section>
</section>
<section id="api-channelinterceptor">
<title>ChannelInterceptor</title>
<para>
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
<interfacename>Messages</interfacename> are being sent to and received from
<interfacename>MessageChannels</interfacename>, those channels provide an opportunity for intercepting
the send and receive operations. The <interfacename>ChannelInterceptor</interfacename> strategy interface
provides methods for each of those operations:
<programlisting language="java"><![CDATA[public interface ChannelInterceptor {
Message<?> preSend(Message<?> message, MessageChannel channel);
void postSend(Message<?> message, MessageChannel channel, boolean sent);
boolean preReceive(MessageChannel channel);
Message<?> postReceive(Message<?> message, MessageChannel channel);
}]]></programlisting>
After implementing the interface, registering the interceptor with a channel is just a matter of calling:
<programlisting language="java">channel.addInterceptor(someChannelInterceptor);</programlisting>
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
<methodname>preReceive</methodname> method can return '<literal>false</literal>' to prevent the receive
operation from proceeding.
</para>
<para>
Because it is rarely necessary to implement all of the interceptor methods, a
<classname>ChannelInterceptorAdapter</classname> class is also available for sub-classing. It provides no-op
methods (the <literal>void</literal> method is empty, the <classname>Message</classname> returning methods
return the Message parameter as-is, and the <literal>boolean</literal> method returns <literal>true</literal>).
Therefore, it is often easiest to extend that class and just implement the method(s) that you need as in the
following example.
<programlisting language="java"><![CDATA[public class CountingChannelInterceptor extends ChannelInterceptorAdapter {
private final AtomicInteger sendCount = new AtomicInteger();
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
sendCount.incrementAndGet();
return message;
}
}]]></programlisting>
</para>
</section>
<section id="api-messagehandler">
<title>MessageHandler</title>
@@ -498,50 +131,6 @@ channel.addInterceptor(interceptor);</programlisting>
</para>
</section>
<section id="api-messageexchangetemplate">
<title>MessageExchangeTemplate</title>
<para>
Whereas the <interfacename>MessageHandler</interfacename> interface provides the foundation for many of the
components that enable non-invasive invocation of your application code <emphasis>from the messaging
system</emphasis>, sometimes it is necessary to invoke the messaging system <emphasis>from your application
code</emphasis>. Spring Integration provides a <classname>MessageExchangeTemplate</classname> that supports a
variety of message-exchanges, including request/reply scenarios. For example, it is possible to send a request
and wait for a reply.
<programlisting language="java">MessageExchangeTemplate template = new MessageExchangeTemplate();
Message reply = template.sendAndReceive(new StringMessage("test"), someChannel);</programlisting>
In that 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.
<programlisting language="java"><![CDATA[public boolean send(final Message<?> message, final MessageTarget target) { ... }
public Message<?> sendAndReceive(final Message<?> request, final MessageTarget target) { .. }
public Message<?> receive(final PollableSource<?> source) { ... }
public boolean receiveAndForward(final PollableSource<?> source, final MessageTarget target) { ... }]]></programlisting>
</para>
<para>
Additionally, a 'transactionManager' can be configured on a MessageExchangeTemplate as well as the various
transaction attributes:
<programlisting language="java">template.setTransactionManager(transactionManager);
template.setPropagationBehaviorName(propagationBehavior);
template.setIsolationLevelName(isolationLevel);
template.setTransactionTimeout(transactionTimeout);
template.setTransactionReadOnly(readOnly);
template.setReceiveTimeout(receiveTimeout);
template.setSendTimeout(sendTimeout);</programlisting>
Finally, there is a also an asynchronous version called <classname>AsyncMessageExchangeTemplate</classname>
whose constructor accepts a <interfacename>TaskExecutor</interfacename>, and whose Message-returning methods
return an <classname>AsyncMessage</classname>. That is essentially a wrapper for any Message that also
implements <interfacename>Future&lt;Message&lt;T&gt;&gt;</interfacename>:
<programlisting>AsyncMessageExchangeTemplate template = new AsyncMessageExchangeTemplate(taskExecutor);
Message reply = template.sendAndReceive(new StringMessage("test"), someChannel);
// do some work in the meantime
reply.getPayload(); // blocks if still waiting for actual reply
</programlisting>
</para>
</section>
<section id="api-gateway">
<title>MessagingGateway</title>
<para>

119
spring-integration-reference/src/endpoint.xml Normal file
View File

@@ -0,0 +1,119 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE chapter PUBLIC "-//OASIS//DTD DocBook XML V4.5//EN"
"http://www.oasis-open.org/docbook/xml/4.5/docbookx.dtd">
<chapter id="endpoint">
<title>Message Endpoints</title>
<para>
As mentioned in the overview, Message Endpoints are responsible for connecting the various messaging components to
channels. Over the next several chapters, you will see a number of different components that consume Messages. Some
of these are also capable of sending reply Messages. Sending Messages is quite straightforward. As shown above in
<xref linkend="channel"/>, it's easy to <emphasis>send</emphasis> a Message to a Message Channel. However,
receiving is a bit more complicated. The main reason is that there are two types of consumers:
<ulink url="http://www.eaipatterns.com/PollingConsumer.html">Polling Consumers</ulink> and
<ulink url="http://www.eaipatterns.com/EventDrivenConsumer.html">Event-Driven Consumers</ulink>.
</para>
<para>
Of the two, Event-Driven Consumers are much simpler. Without any need to manage and schedule a separate poller
thread, they are essentially just listeners with a callback method. When connecting to one of Spring Integration's
subscribable Message Channels, this simple option works great. However, when connecting to a buffering, pollable
Message Channel, some component has to schedule and manage the polling thread(s). Spring Integration provides
two different endpoint implementations to accommodate these two types of consumers. Therefore, the consumers
themselves can simply implement the callback interface. When polling is required, the endpoint acts as a
"container" for the consumer instance. The benefit is similar to that of using a container for hosting
Message-Driven Beans, but since these consumers are simply Spring-managed Objects running within an
ApplicationContext, it more closely resembles Spring's own MessageListener containers.
</para>
<section id="endpoint-consumer">
<title>Message Consumer</title>
<para>
Spring Integration's <interfacename>MessageConsumer</interfacename> interface is defined as follows:
<programlisting language="java">public interface MessageConsumer {
void onMessage(Message&lt;?&gt; message);
}</programlisting>
Despite its simplicity, this provides the foundation for most of the components that will be covered in the
following chapters (Routers, Transformers, Splitters, Aggregators, Service Activators, etc). Those components
each perform very different functionality with the Messages they receive, but the requirements for actually
receiving a Message are the same, and the choice between polling and event-driven behavior is also the same.
Spring Integration provides two endpoint implementations that "host" these callback-based consumers and allow
them to be connected to Message Channels.
</para>
</section>
<section id="endpoint-eventdrivenconsumer">
<title>Event-Driven Consumer</title>
<para>
Because it is the simpler of the two, we will cover the Event-Driven Consumer endpoint first. You may recall that
the <interfacename>SubscribableChannel</interfacename> interface provides a <methodname>subscribe()</methodname>
method and that the method accepts a <interfacename>MessageConsumer</interfacename> parameter (as shown in
<xref linkend="channel-interfaces-subscribablechannel"/>):
<programlisting language="java">
subscribableChannel.subscribe(messageConsumer);
</programlisting>
Since a consumer that is subscribed to a channel does not have to actively poll that channel, this is an
Event-Driven Consumer, and the corresponding endpoint "container" class provided by Spring Integration accepts a
<interfacename>MessageConsumer</interfacename> and a <interfacename>SubscribableChannel</interfacename>:
<programlisting language="java">MessageConsumer consumer = new ExampleConsumer();
SubscribableChannel channel = (SubscribableChannel) context.getBean("exampleSubscribableChannel");
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(consumer, channel);</programlisting>
</para>
</section>
<section id="endpoint-pollingconsumer">
<title>Polling Consumer</title>
<para>
Spring Integration also provides a <classname>PollingConsumerEndpoint</classname>, and it can be instantiated in
the same way except that the channel must implement <interfacename>PollableChannel</interfacename>:
<programlisting language="java">MessageConsumer consumer = new ExampleConsumer();
PollableChannel channel = (PollableChannel) context.getBean("examplePollableChannel");
PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint(consumer, channel);</programlisting>
</para>
<para>
There are many other configuration options for the polling endpoint. For example, the trigger can be provided:
<programlisting language="java">
PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint(consumer, channel);
endpoint.setTrigger(new IntervalTrigger(30, TimeUnit.SECONDS));</programlisting>
Likewise, other polling-related configuration properties may be specified:
<programlisting language="java">
PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint(consumer, channel);
endpoint.setMaxMessagesPerPoll(10);
endpoint.setReceiveTimeout(5000);</programlisting>
A polling consumer may even delegate to a Spring <interfacename>TaskExecutor</interfacename> and
participate in Spring-managed transactions. The following example shows the configuration of both:
<programlisting language="java">
PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint(consumer, channel);
TaskExecutor taskExecutor = (TaskExecutor) context.getBean("exampleExecutor");
endpoint.setTaskExecutor(taskExecutor);
PlatformTransactionManager txManager = (PlatformTransationManager) context.getBean("exampleTxManager");
endpoint.setTransactionManager(txManager);</programlisting>
The examples above show dependency lookups, but keep in mind that these endpoints will most often be configured
as Spring <emphasis>bean definitions</emphasis>. In fact, Spring Integration also provides a
<interfacename>FactoryBean</interfacename> that creates the appropriate endpoint type based on the type of
channel, and there is full XML namespace support to even further hide those details. The namespace-based
configuration will be featured as each component type is introduced.
<note>
Interestingly, many of the <interfacename>MessageConsumer</interfacename> implementations are also capable of
generating reply Messages. As mentioned above, sending Messages is trivial when compared to the Message
reception. Nevertheless, <emphasis>when</emphasis> and <emphasis>how many</emphasis> reply Messages are sent
depends on the consumer type. For example, an <emphasis>Aggregator</emphasis> waits for a number of Messages to
arrive and is often a downstream consumer for a <emphasis>Splitter</emphasis> which may generate multiple
replies for each Message it consumes. When using the namespace configuration, you do not strictly need to know
all of the details, but it still might be worth knowing that several of these components share a common base
class, the <classname>AbstractReplyProducingMessageConsumer</classname>, and it provides a
<methodname>setOutputChannel(..)</methodname> method.
</note>
</para>
</section>
</chapter>

3
spring-integration-reference/src/spring-integration-reference.xml
View File

@@ -43,8 +43,9 @@
<xi:include href="./overview.xml"/>
<xi:include href="./message.xml"/>
<xi:include href="./channel.xml"/>
<xi:include href="./transformation.xml"/>
<xi:include href="./endpoint.xml"/>
<xi:include href="./routing.xml"/>
<xi:include href="./transformation.xml"/>
<xi:include href="./splitting.xml"/>
<xi:include href="./aggregator-resequencer.xml"/>
<xi:include href="./adapters.xml"/>