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

Removed the 'core-api' section. Added 'message' and 'channel' chapters.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-20 19:11:07 +00:00
parent 379fed3f5b
commit a4d056ef9f
3 changed files with 462 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

242
spring-integration-reference/src/channel.xml Normal file
View File

@@ -0,0 +1,242 @@
<?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="channel">
<title>Message Channels</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.
</para>
<section id="channel-interfaces">
<title>The MessageChannel Interface</title>
<para>
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>
</section>
<section id="channel-implementations">
<title>Message Channel Implementations</title>
<para>
Spring Integration provides several different Message Channel implementations. Each is briefly described in the
sections below.
</para>
<section id="channel-implementations-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="channel-implementations-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="channel-implementations-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="channel-implementations-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="channel-implementations-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="channel-implementations-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="channel-interceptors">
<title>Channel Interceptors</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="channel-template">
<title>MessageChannelTemplate</title>
<para>
As you will see when the endpoints and their various configuration options are introduced, Spring Integration
provides a foundation for messaging components that enables non-invasive invocation of your application code
<emphasis>from the messaging system</emphasis>. However, sometimes it is necessary to invoke the messaging system
<emphasis>from your application code</emphasis>. For convenience when implementing such use-cases, Spring
Integration provides a <classname>MessageChannelTemplate</classname> that supports a variety of operations across
the Message Channels, including request/reply scenarios. For example, it is possible to send a request
and wait for a reply.
<programlisting language="java">MessageChannelTemplate template = new MessageChannelTemplate();
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 MessageChannel channel) { ... }
public Message<?> sendAndReceive(final Message<?> request, final MessageChannel channel) { .. }
public Message<?> receive(final PollableChannel<?> channel) { ... }]]></programlisting>
</para>
</section>
</chapter>

218
spring-integration-reference/src/message.xml Normal file
View File

@@ -0,0 +1,218 @@
<?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="message">
<title>Message Construction</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.
</para>
<section id="message-interface">
<title>The Message Interface</title>
<para>Here is the definition of the <interfacename>Message</interfacename> interface:
<programlisting language="java">public interface Message&lt;T&gt; {
T getPayload();
MessageHeaders getHeaders();
}</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="message-headers">
<title>Message Headers</title>
<para>
Just as Spring Integration allows any Object to be used as the payload of a Message, it also supports any Object
types as header values. In fact, the <classname>MessageHeaders</classname> class implements the
<emphasis>java.util.Map</emphasis> interface:
<programlisting language="java">public final class MessageHeaders implements Map&lt;String, Object&gt;, Serializable {
...
}</programlisting>
<note>
Even though the MessageHeaders implements Map, it is effectively a read-only implementation. Any attempt to
<emphasis>put</emphasis> a value in the Map will result in an <classname>UnsupportedOperationException</classname>.
The same applies for <emphasis>remove</emphasis> and <emphasis>clear</emphasis>. Since Messages may be passed to
multiple consumers, the structure of the Map cannot be modified. Likewise, the Message's payload Object can not
be <emphasis>set</emphasis> after the initial creation. However, the mutability of the header values themselves
(or the payload Object) is intentionally left as a decision for the framework user.
</note>
</para>
<para>
As an implementation of Map, the headers can obviously be retrieved by calling <methodname>get(..)</methodname>
with the name of the header. Alternatively, you can provide the expected <emphasis>Class</emphasis> as an
additional parameter. Even better, when retrieving one of the pre-defined values, convenient getters are
available. Here is an example of each of these three options:
<programlisting language="java"> Object someValue = message.getHeaders().get("someKey");
CustomerId customerId = message.getHeaders().get("customerId", CustomerId.class);
Long timestamp = message.getHeaders().getTimestamp();
</programlisting>
</para>
<para>
The following Message headers are pre-defined:
<table id="message-headers-table">
<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 and outbound adapter implementations will also provide and/or expect certain headers, and additional
user-defined headers can also be configured.
</para>
</section>
<section id="message-implementations">
<title>Message Implementations</title>
<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.
</para>
<para>
There are also two convenient subclasses available: <classname>StringMessage</classname> and
<classname>ErrorMessage</classname>. The former accepts a String as its payload:
<programlisting language="java">StringMessage message = new StringMessage("hello world");
String s = message.getPayload();</programlisting>
And, the latter accepts any <classname>Throwable</classname> object as its payload:
<programlisting language="java">ErrorMessage message = new ErrorMessage(someThrowable);
Throwable t = message.getPayload();</programlisting>
Notice that these implementations take advantage of the fact that the <classname>GenericMessage</classname>
base class is parameterized. Therefore, as shown in both examples, no casting is necessary when retrieving
the Message payload Object.
</para>
</section>
<section id="message-builder">
<title>The MessageBuilder Helper Class</title>
<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 chapter). 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>
</section>
</chapter>

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

@@ -41,7 +41,8 @@
<toc></toc>
<xi:include href="./overview.xml"/>
<xi:include href="./core-api.xml"/>
<xi:include href="./message.xml"/>
<xi:include href="./channel.xml"/>
<xi:include href="./transformation.xml"/>
<xi:include href="./routing.xml"/>
<xi:include href="./splitting.xml"/>