tceic.com
学霸学习网 这下你爽了
广告
当前位置:首页 >> IT/计算机 >>

Spring


Spring Integration Reference Manual

2.0.0.M6

Copyright ? 2005-2007 Mark Fisher, Marius Bogoevici, Iwein Fuld, Jonas Partner, Oleg Zhurakousky, Gary Russell, Josh Long

? SpringSource Inc., 2010

1. Spring Integration Overview .......................................................................................................... 1 1.1. Background ....................................................................................................................... 1 1.2. Goals and Principles ........................................................................................................... 1 1.3. Main Components .............................................................................................................. 2 1.3.1. Message .................................................................................................................. 2 1.3.2. Message Channel .................................................................................................... 2 1.3.3. Message Endpoint ................................................................................................... 3 1.4. Message Endpoints ............................................................................................................. 3 1.4.1. Transformer ............................................................................................................ 3 1.4.2. Filter ...................................................................................................................... 3 1.4.3. Router .................................................................................................................... 4 1.4.4. Splitter .................................................................................................................... 4 1.4.5. Aggregator .............................................................................................................. 4 1.4.6. Service Activator ..................................................................................................... 4 1.4.7. Channel Adapter ..................................................................................................... 5 2. Message Construction ................................................................................................................... 6 2.1. The Message Interface ........................................................................................................ 6 2.2. Message Headers ............................................................................................................... 6 2.3. Message Implementations ................................................................................................... 7 2.4. The MessageBuilder Helper Class ....................................................................................... 7 3. Message Channels ....................................................................................................................... 10 3.1. The MessageChannel Interface .......................................................................................... 10 3.1.1. PollableChannel .................................................................................................... 10 3.1.2. SubscribableChannel ............................................................................................. 10 3.2. Message Channel Implementations .................................................................................... 11 3.2.1. PublishSubscribeChannel ....................................................................................... 11 3.2.2. QueueChannel ....................................................................................................... 11 3.2.3. PriorityChannel ..................................................................................................... 11 3.2.4. RendezvousChannel .............................................................................................. 11 3.2.5. DirectChannel ....................................................................................................... 12 3.2.6. ExecutorChannel ................................................................................................... 13 3.2.7. ThreadLocalChannel ............................................................................................. 13 3.3. Channel Interceptors ......................................................................................................... 14 3.4. MessageChannelTemplate ................................................................................................ 15 3.5. Configuring Message Channels ......................................................................................... 15 3.5.1. DirectChannel Configuration .................................................................................. 16 3.5.2. QueueChannel Configuration ................................................................................. 16 3.5.3. PublishSubscribeChannel Configuration ................................................................. 16 3.5.4. ExecutorChannel ................................................................................................... 17 3.5.5. PriorityChannel Configuration ................................................................................ 17 3.5.6. RendezvousChannel Configuration ......................................................................... 18 3.5.7. ThreadLocalChannel Configuration ........................................................................ 18 3.5.8. Channel Interceptor Configuration .......................................................................... 18 3.5.9. Global Channel Interceptor Configuration ............................................................... 18 3.5.10. Wire Tap ............................................................................................................. 19 4. Message Endpoints ..................................................................................................................... 20 4.1. Message Handler .............................................................................................................. 20 4.2. Event Driven Consumer .................................................................................................... 20 4.3. Polling Consumer ............................................................................................................. 21 4.4. Namespace Support .......................................................................................................... 22 4.5. Payload Type Conversion ................................................................................................. 24 5. Service Activator ........................................................................................................................ 25 Spring-WS (2.0.0.M6) ii

Spring Integration 2.0.0.M6 5.1. Introduction ..................................................................................................................... 25 5.2. The <service-activator/> Element ...................................................................................... 25 6. Channel Adapter ......................................................................................................................... 27 6.1. The <inbound-channel-adapter> element ........................................................................... 27 6.2. The <outbound-channel-adapter/> element ........................................................................ 27 7. Router ........................................................................................................................................ 29 7.1. Router Implementations .................................................................................................... 29 7.1.1. PayloadTypeRouter ............................................................................................... 29 7.1.2. HeaderValueRouter ............................................................................................... 29 7.1.3. RecipientListRouter ............................................................................................... 30 7.2. The <router> element ....................................................................................................... 31 7.3. The @Router Annotation .................................................................................................. 31 8. Filter .......................................................................................................................................... 33 8.1. Introduction ..................................................................................................................... 33 8.2. The <filter> Element ........................................................................................................ 33 9. Transformer ................................................................................................................................ 35 9.1. Introduction ..................................................................................................................... 35 9.2. The <transformer> Element .............................................................................................. 35 9.3. The @Transformer Annotation ......................................................................................... 37 10. Splitter ..................................................................................................................................... 39 10.1. Introduction ................................................................................................................... 39 10.2. Programming model ....................................................................................................... 39 10.3. Configuring a Splitter using XML ................................................................................... 40 10.4. Configuring a Splitter with Annotations ........................................................................... 40 11. Aggregator ............................................................................................................................... 41 11.1. Introduction ................................................................................................................... 41 11.2. Functionality .................................................................................................................. 41 11.3. Programming model ....................................................................................................... 41 11.3.1. CorrelatingMessageHandler ................................................................................. 41 11.3.2. ReleaseStrategy ................................................................................................... 43 11.3.3. CorrelationStrategy .............................................................................................. 43 11.4. Configuring an Aggregator with XML ............................................................................. 44 11.5. Managing State in an Aggregator: MessageGroupStore ..................................................... 46 11.6. Configuring an Aggregator with Annotations ................................................................... 47 12. Resequencer ............................................................................................................................. 49 12.1. Introduction ................................................................................................................... 49 12.2. Functionality .................................................................................................................. 49 12.3. Configuring a Resequencer with XML ............................................................................. 49 13. Delayer .................................................................................................................................... 51 13.1. Introduction ................................................................................................................... 51 13.2. The <delayer> Element ................................................................................................... 51 14. Message Handler Chain ............................................................................................................. 52 14.1. Introduction ................................................................................................................... 52 14.2. The <chain> Element ..................................................................................................... 52 15. Messaging Bridge ..................................................................................................................... 54 15.1. Introduction ................................................................................................................... 54 15.2. The <bridge> Element .................................................................................................... 54 16. Inbound Messaging Gateways .................................................................................................... 55 16.1. SimpleMessagingGateway .............................................................................................. 55 16.2. GatewayProxyFactoryBean ............................................................................................. 55 17. Message Publishing ................................................................................................................... 58 17.1. Message Publishing Configuration .................................................................................. 58

Spring-WS (2.0.0.M6)

iii

Spring Integration 2.0.0.M6 17.1.1. Annotation-driven approach via @Publisher annotation ......................................... 58 17.1.2. XML-based approach via <publisher> element ...................................................... 59 18. File Support .............................................................................................................................. 61 18.1. Introduction ................................................................................................................... 61 18.2. Reading Files ................................................................................................................. 61 18.3. Writing files ................................................................................................................... 62 18.4. File Transformers ........................................................................................................... 63 19. JDBC Support ........................................................................................................................... 65 19.1. Inbound Channel Adapter ............................................................................................... 65 19.1.1. Polling and Transactions ...................................................................................... 65 19.2. Outbound Channel Adapter ............................................................................................. 66 19.3. Message Store ................................................................................................................ 66 19.3.1. Initializing the Database ....................................................................................... 67 19.3.2. Partitioning a Message Store ................................................................................ 67 20. JMS Support ............................................................................................................................. 68 20.1. Inbound Channel Adapter ............................................................................................... 68 20.2. Message-Driven Channel Adapter ................................................................................... 69 20.3. Outbound Channel Adapter ............................................................................................. 69 20.4. Inbound Gateway ........................................................................................................... 70 20.5. Outbound Gateway ......................................................................................................... 71 20.6. Message Conversion, Marshalling and Unmarshalling ...................................................... 71 20.7. JMS Backed Message Channels ...................................................................................... 71 20.8. JMS Samples ................................................................................................................. 73 21. Web Services Support ............................................................................................................... 74 21.1. Outbound Web Service Gateways .................................................................................... 74 21.2. Inbound Web Service Gateways ...................................................................................... 74 21.3. Web Service Namespace Support .................................................................................... 74 22. RMI Support ............................................................................................................................. 77 22.1. Introduction ................................................................................................................... 77 22.2. Outbound RMI ............................................................................................................... 77 22.3. Inbound RMI ................................................................................................................. 77 22.4. RMI namespace support ................................................................................................. 77 23. HttpInvoker Support .................................................................................................................. 79 23.1. Introduction ................................................................................................................... 79 23.2. HttpInvoker Inbound Gateway ........................................................................................ 79 23.3. HttpInvoker Outbound Gateway ...................................................................................... 79 23.4. HttpInvoker Namespace Support ..................................................................................... 80 24. HTTP Support .......................................................................................................................... 81 24.1. Introduction ................................................................................................................... 81 24.2. Http Inbound Gateway .................................................................................................... 81 24.3. Http Outbound Gateway ................................................................................................. 82 24.4. HTTP Namespace Support .............................................................................................. 82 25. TCP and UDP Support .............................................................................................................. 84 25.1. Introduction ................................................................................................................... 84 25.2. UDP Adapters ................................................................................................................ 84 25.3. TCP Connection Factories .............................................................................................. 86 25.4. TCP Adapters ................................................................................................................ 88 25.5. TCP Adapters - Legacy - These Adapters Will be Removed Before 2.0 GA ....................... 89 25.6. TCP Gateways ............................................................................................................... 90 25.7. IP Configuration Attributes ............................................................................................. 91 26. Mail Support ............................................................................................................................. 98 26.1. Mail-Sending Channel Adapter ....................................................................................... 98

Spring-WS (2.0.0.M6)

iv

Spring Integration 2.0.0.M6 26.2. Mail-Receiving Channel Adapter .................................................................................... 98 26.3. Mail Namespace Support ................................................................................................ 99 27. JMX Support .......................................................................................................................... 101 27.1. Notification Listening Channel Adapter ......................................................................... 101 27.2. Notification Publishing Channel Adapter ....................................................................... 101 27.3. Attribute Polling Channel Adapter ................................................................................. 102 27.4. Operation Invoking Channel Adapter ............................................................................. 102 27.5. Operation Invoking outbound Gateway .......................................................................... 102 27.6. Control Bus .................................................................................................................. 103 28. XMPP Support ........................................................................................................................ 104 28.1. Introduction ................................................................................................................. 104 28.2. Using The Spring Integration XMPP Namespace ............................................................ 104 28.3. XMPP Connection ........................................................................................................ 105 28.4. XMPP Messages .......................................................................................................... 105 28.4.1. Inbound Message Adapter .................................................................................. 105 28.4.2. Outbound Message Adapter ............................................................................... 106 28.5. XMPP Presence ........................................................................................................... 107 28.5.1. Inbound Presence Adapter .................................................................................. 108 28.5.2. Outbound Presence Adapter ............................................................................... 108 29. Stream Support ....................................................................................................................... 109 29.1. Introduction ................................................................................................................. 109 29.2. Reading from streams ................................................................................................... 109 29.3. Writing to streams ........................................................................................................ 109 29.4. Stream namespace support ............................................................................................ 109 30. Spring ApplicationEvent Support ............................................................................................. 111 30.1. Receiving Spring ApplicationEvents .............................................................................. 111 30.2. Sending Spring ApplicationEvents ................................................................................ 111 31. Dealing with XML Payloads .................................................................................................... 112 31.1. Introduction ................................................................................................................. 112 31.2. Transforming xml payloads ........................................................................................... 112 31.3. Namespace support for xml transformers ....................................................................... 113 31.4. Splitting xml messages ................................................................................................. 114 31.5. Routing xml messages using XPath ............................................................................... 115 31.6. Selecting xml messages using XPath ............................................................................. 115 31.7. XPath components namespace support ........................................................................... 116 32. Security in Spring Integration .................................................................................................. 118 32.1. Introduction ................................................................................................................. 118 32.2. Securing channels ......................................................................................................... 118 A. Spring Integration Samples ....................................................................................................... 119 A.1. The Cafe Sample ........................................................................................................... 119 A.2. The XML Messaging Sample ......................................................................................... 122 A.3. The OSGi Samples ........................................................................................................ 123 B. Configuration ........................................................................................................................... 127 B.1. Introduction .................................................................................................................. 127 B.2. Namespace Support ....................................................................................................... 127 B.3. Configuring the Task Scheduler ...................................................................................... 128 B.4. Error Handling .............................................................................................................. 129 B.5. Annotation Support ........................................................................................................ 130 B.6. Message Mapping rules and conventions ......................................................................... 132 B.6.1. Simple Scenarios ................................................................................................ 132 B.6.2. Complex Scenarios ............................................................................................. 134 C. Additional Resources ................................................................................................................ 137

Spring-WS (2.0.0.M6)

v

Spring Integration 2.0.0.M6 C.1. Spring Integration Home ................................................................................................ 137

Spring-WS (2.0.0.M6)

vi

Chapter 1. Spring Integration Overview
1.1. Background
One of the key themes of the Spring Framework is inversion of control. In its broadest sense, this means that the framework handles responsibilities on behalf of the components that are managed within its context. The components themselves are simplified since they are relieved of those responsibilities. For example, dependency injection relieves the components of the responsibility of locating or creating their dependencies. Likewise, aspect-oriented programming relieves business components of generic cross-cutting concerns by modularizing them into reusable aspects. In each case, the end result is a system that is easier to test, understand, maintain, and extend. Furthermore, the Spring framework and portfolio provide a comprehensive programming model for building enterprise applications. Developers benefit from the consistency of this model and especially the fact that it is based upon well-established best practices such as programming to interfaces and favoring composition over inheritance. Spring's simplified abstractions and powerful support libraries boost developer productivity while simultaneously increasing the level of testability and portability. Spring Integration is a new member of the Spring portfolio motivated by these same goals and principles. It extends the Spring programming model into the messaging domain and builds upon Spring's existing enterprise integration support to provide an even higher level of abstraction. It supports message-driven architectures where inversion of control applies to runtime concerns, such as when certain business logic should execute and where the response should be sent. It supports routing and transformation of messages so that different transports and different data formats can be integrated without impacting testability. In other words, the messaging and integration concerns are handled by the framework, so business components are further isolated from the infrastructure and developers are relieved of complex integration responsibilities. As an extension of the Spring programming model, Spring Integration provides a wide variety of configuration options including annotations, XML with namespace support, XML with generic "bean" elements, and of course direct usage of the underlying API. That API is based upon well-defined strategy interfaces and non-invasive, delegating adapters. Spring Integration's design is inspired by the recognition of a strong affinity between common patterns within Spring and the well-known Enterprise Integration Patterns as described in the book of the same name by Gregor Hohpe and Bobby Woolf (Addison Wesley, 2004). Developers who have read that book should be immediately comfortable with the Spring Integration concepts and terminology.

1.2. Goals and Principles
Spring Integration is motivated by the following goals: ? Provide a simple model for implementing complex enterprise integration solutions. ? Facilitate asynchronous, message-driven behavior within a Spring-based application. ? Promote intuitive, incremental adoption for existing Spring users. Spring Integration is guided by the following principles: ? Components should be loosely coupled for modularity and testability.

Spring-WS (2.0.0.M6)

1

Spring Integration Overview

? The framework should enforce separation of concerns between business logic and integration logic. ? Extension points should be abstract in nature but within well-defined boundaries to promote reuse and portability.

1.3. Main Components
From the vertical perspective, a layered architecture facilitates separation of concerns, and interface-based contracts between layers promote loose coupling. Spring-based applications are typically designed this way, and the Spring framework and portfolio provide a strong foundation for following this best practice for the full-stack of an enterprise application. Message-driven architectures add a horizontal perspective, yet these same goals are still relevant. Just as "layered architecture" is an extremely generic and abstract paradigm, messaging systems typically follow the similarly abstract "pipes-and-filters" model. The "filters" represent any component that is capable of producing and/or consuming messages, and the "pipes" transport the messages between filters so that the components themselves remain loosely-coupled. It is important to note that these two high-level paradigms are not mutually exclusive. The underlying messaging infrastructure that supports the "pipes" should still be encapsulated in a layer whose contracts are defined as interfaces. Likewise, the "filters" themselves would typically be managed within a layer that is logically above the application's service layer, interacting with those services through interfaces much in the same way that a web-tier would.

1.3.1. Message
In Spring Integration, a Message is a generic wrapper for any Java object combined with metadata used by the framework while handling that object. It consists of a payload and headers. The payload can be of any type and the headers hold commonly required information such as id, timestamp, expiration, and return address. Headers are also used for passing values to and from connected transports. For example, when creating a Message from a received File, the file name may be stored in a header to be accessed by downstream components. Likewise, if a Message's content is ultimately going to be sent by an outbound Mail adapter, the various properties (to, from, cc, subject, etc.) may be configured as Message header values by an upstream component. Developers can also store any arbitrary key-value pairs in the headers.

1.3.2. Message Channel
A Message Channel represents the "pipe" of a pipes-and-filters architecture. Producers send Messages to a channel, and consumers receive Messages from a channel. The Message Channel therefore decouples the messaging components, and also provides a convenient point for interception and monitoring of Messages. A Message Channel may follow either Point-to-Point or Publish/Subscribe semantics. With a Point-to-Point channel, at most one consumer can receive each Message sent to the channel. Publish/Subscribe channels, on the other hand, will attempt to broadcast each Message to all of its subscribers. Spring Integration supports both of these. Whereas "Point-to-Point" and "Publish/Subscribe" define the two options for how many consumers will ultimately receive each Message, there is another important consideration: should the channel buffer messages? In Spring Integration, Pollable Channels are capable of buffering Messages within a queue. The advantage of buffering is that it allows for throttling the inbound Messages and thereby prevents overloading a consumer. However, as the name suggests, this also adds some complexity, since a consumer can only receive the Messages from such a channel if a poller is configured. On the other hand, a consumer connected to a Subscribable Channel is simply Message-driven. The variety of channel implementations available in Spring Integration will be discussed in detail in Section 3.2, “Message Channel Implementations”.

Spring-WS (2.0.0.M6)

2

Spring Integration Overview

1.3.3. Message Endpoint
One of the primary goals of Spring Integration is to simplify the development of enterprise integration solutions through inversion of control. This means that you should not have to implement consumers and producers directly, and you should not even have to build Messages and invoke send or receive operations on a Message Channel. Instead, you should be able to focus on your specific domain model with an implementation based on plain Objects. Then, by providing declarative configuration, you can "connect" your domain-specific code to the messaging infrastructure provided by Spring Integration. The components responsible for these connections are Message Endpoints. This does not mean that you will necessarily connect your existing application code directly. Any real-world enterprise integration solution will require some amount of code focused upon integration concerns such as routing and transformation. The important thing is to achieve separation of concerns between such integration logic and business logic. In other words, as with the Model-View-Controller paradigm for web applications, the goal should be to provide a thin but dedicated layer that translates inbound requests into service layer invocations, and then translates service layer return values into outbound replies. The next section will provide an overview of the Message Endpoint types that handle these responsibilities, and in upcoming chapters, you will see how Spring Integration's declarative configuration options provide a non-invasive way to use each of these.

1.4. Message Endpoints
A Message Endpoint represents the "filter" of a pipes-and-filters architecture. As mentioned above, the endpoint's primary role is to connect application code to the messaging framework and to do so in a non-invasive manner. In other words, the application code should ideally have no awareness of the Message objects or the Message Channels. This is similar to the role of a Controller in the MVC paradigm. Just as a Controller handles HTTP requests, the Message Endpoint handles Messages. Just as Controllers are mapped to URL patterns, Message Endpoints are mapped to Message Channels. The goal is the same in both cases: isolate application code from the infrastructure. These concepts are discussed at length along with all of the patterns that follow in the Enterprise Integration Patterns book. Here, we provide only a high-level description of the main endpoint types supported by Spring Integration and their roles. The chapters that follow will elaborate and provide sample code as well as configuration examples.

1.4.1. Transformer
A Message Transformer is responsible for converting a Message's content or structure and returning the modified Message. Probably the most common type of transformer is one that converts the payload of the Message from one format to another (e.g. from XML Document to java.lang.String). Similarly, a transformer may be used to add, remove, or modify the Message's header values.

1.4.2. Filter
A Message Filter determines whether a Message should be passed to an output channel at all. This simply requires a boolean test method that may check for a particular payload content type, a property value, the presence of a header, etc. If the Message is accepted, it is sent to the output channel, but if not it will be dropped (or for a more severe implementation, an Exception could be thrown). Message Filters are often used in conjunction with a Publish Subscribe channel, where multiple consumers may receive the same Message and use the filter to narrow down the set of Messages to be processed based on some criteria.

Note
Be careful not to confuse the generic use of "filter" within the Pipes-and-Filters architectural pattern with this specific endpoint type that selectively narrows down the Messages flowing Spring-WS (2.0.0.M6) 3

Spring Integration Overview

between two channels. The Pipes-and-Filters concept of "filter" matches more closely with Spring Integration's Message Endpoint: any component that can be connected to Message Channel(s) in order to send and/or receive Messages.

1.4.3. Router
A Message Router is responsible for deciding what channel or channels should receive the Message next (if any). Typically the decision is based upon the Message's content and/or metadata available in the Message Headers. A Message Router is often used as a dynamic alternative to a statically configured output channel on a Service Activator or other endpoint capable of sending reply Messages. Likewise, a Message Router provides a proactive alternative to the reactive Message Filters used by multiple subscribers as described above.

1.4.4. Splitter
A Splitter is another type of Message Endpoint whose responsibility is to accept a Message from its input channel, split that Message into multiple Messages, and then send each of those to its output channel. This is typically used for dividing a "composite" payload object into a group of Messages containing the sub-divided payloads.

1.4.5. Aggregator
Basically a mirror-image of the Splitter, the Aggregator is a type of Message Endpoint that receives multiple Messages and combines them into a single Message. In fact, Aggregators are often downstream consumers in a pipeline that includes a Splitter. Technically, the Aggregator is more complex than a Splitter, because it is required to maintain state (the Messages to-be-aggregated), to decide when the complete group of Messages is available, and to timeout if necessary. Furthermore, in case of a timeout, the Aggregator needs to know whether to send the partial results or to discard them to a separate channel. Spring Integration provides a CompletionStrategy as well as configurable settings for timeout, whether to send partial results upon timeout, and the discard channel.

1.4.6. Service Activator
A Service Activator is a generic endpoint for connecting a service instance to the messaging system. The input Message Channel must be configured, and if the service method to be invoked is capable of returning a value, an output Message Channel may also be provided.

Note
The output channel is optional, since each Message may also provide its own 'Return Address' header. This same rule applies for all consumer endpoints. The Service Activator invokes an operation on some service object to process the request Message, extracting the request Message's payload and converting if necessary (if the method does not expect a Message-typed parameter). Whenever the service object's method returns a value, that return value will likewise be converted to a reply Message if necessary (if it's not already a Message). That reply Message is sent to the output channel. If no output channel has been configured, then the reply will be sent to the channel specified in the Message's "return address" if available. A request-reply "Service Activator" endpoint connects a target object's method to input and output Message Channels.

Spring-WS (2.0.0.M6)

4

Spring Integration Overview

1.4.7. Channel Adapter
A Channel Adapter is an endpoint that connects a Message Channel to some other system or transport. Channel Adapters may be either inbound or outbound. Typically, the Channel Adapter will do some mapping between the Message and whatever object or resource is received-from or sent-to the other system (File, HTTP Request, JMS Message, etc). Depending on the transport, the Channel Adapter may also populate or extract Message header values. Spring Integration provides a number of Channel Adapters, and they will be described in upcoming chapters. An inbound "Channel Adapter" endpoint connects a source system to a MessageChannel. An outbound "Channel Adapter" endpoint connects a MessageChannel to a target system.

Spring-WS (2.0.0.M6)

5

Chapter 2. Message Construction
The Spring Integration Message is a generic container for data. Any object can be provided as the payload, and each Message also includes headers containing user-extensible properties as key-value pairs.

2.1. The Message Interface
Here is the definition of the Message interface:
public interface Message<T> { T getPayload(); MessageHeaders getHeaders(); }

The Message 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 does require access to information about the Message, such metadata can typically be stored to and retrieved from the metadata in the Message Headers.

2.2. Message Headers
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 MessageHeaders class implements the java.util.Map interface:
public final class MessageHeaders implements Map<String, Object>, Serializable { ... }

Note
Even though the MessageHeaders implements Map, it is effectively a read-only implementation. Any attempt to put a value in the Map will result in an UnsupportedOperationException. The same applies for remove and clear. 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 set 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. As an implementation of Map, the headers can obviously be retrieved by calling get(..) with the name of the header. Alternatively, you can provide the expected Class 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:
Object someValue = message.getHeaders().get("someKey"); CustomerId customerId = message.getHeaders().get("customerId", CustomerId.class); Long timestamp = message.getHeaders().getTimestamp();

Spring-WS (2.0.0.M6)

6

Message Construction

The following Message headers are pre-defined: Table 2.1. Pre-defined Message Headers Header Name ID TIMESTAMP EXPIRATION_DATE CORRELATION_ID REPLY_CHANNEL ERROR_CHANNEL SEQUENCE_NUMBER SEQUENCE_SIZE PRIORITY java.util.UUID java.lang.Long java.lang.Long java.lang.Object java.lang.Object (can be a String or MessageChannel) java.lang.Object (can be a String or MessageChannel) java.lang.Integer java.lang.Integer MessagePriority (an enum) Header Type

Many inbound and outbound adapter implementations will also provide and/or expect certain headers, and additional user-defined headers can also be configured.

2.3. Message Implementations
The base implementation of the Message interface is GenericMessage<T>, and it provides two constructors:
new GenericMessage<T>(T payload); new GenericMessage<T>(T payload, Map<String, Object> headers)

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: StringMessage and ErrorMessage. The former accepts a String as its payload:
StringMessage message = new StringMessage("hello world"); String s = message.getPayload();

And, the latter accepts any Throwable object as its payload:
ErrorMessage message = new ErrorMessage(someThrowable); Throwable t = message.getPayload();

Notice that these implementations take advantage of the fact that the GenericMessage base class is parameterized. Therefore, as shown in both examples, no casting is necessary when retrieving the Message payload Object.

2.4. The MessageBuilder Helper Class
Spring-WS (2.0.0.M6) 7

Message Construction

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 unmodifiable Collection, and the MessageHeaders' map further exemplifies that; even though the MessageHeaders class implements java.util.Map, any attempt to invoke a put operation (or 'remove' or 'clear') on the MessageHeaders will result in an UnsupportedOperationException. 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: MessageBuilder. 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 and payload of that Message will be copied to the new Message:
Message<String> message1 = MessageBuilder.withPayload("test") .setHeader("foo", "bar") .build(); Message<String> message2 = MessageBuilder.fromMessage(message1).build(); assertEquals("test", message2.getPayload()); assertEquals("bar", message2.getHeaders().get("foo"));

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.
Message<String> message3 = MessageBuilder.withPayload("test3") .copyHeaders(message1.getHeaders()) .build(); Message<String> message4 = MessageBuilder.withPayload("test4") .setHeader("foo", 123) .copyHeadersIfAbsent(message1.getHeaders()) .build(); assertEquals("bar", message3.getHeaders().get("foo")); assertEquals(123, message4.getHeaders().get("foo"));

Notice that the copyHeadersIfAbsent does not overwrite existing values. Also, in the second example above, you can see how to set any user-defined header with setHeader. 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).
Message<Integer> importantMessage = MessageBuilder.withPayload(99) .setPriority(MessagePriority.HIGHEST) .build(); assertEquals(MessagePriority.HIGHEST, importantMessage.getHeaders().getPriority()); Message<Integer> anotherMessage = MessageBuilder.fromMessage(importantMessage) .setHeaderIfAbsent(MessageHeaders.PRIORITY, MessagePriority.LOW) .build(); assertEquals(MessagePriority.HIGHEST, anotherMessage.getHeaders().getPriority());

The MessagePriority is only considered when using a PriorityChannel (as described in the next chapter). It is defined as an enum with five possible values:
public enum MessagePriority {

Spring-WS (2.0.0.M6)

8

Message Construction

HIGHEST, HIGH, NORMAL, LOW, LOWEST }

Spring-WS (2.0.0.M6)

9

Chapter 3. Message Channels
While the Message plays the crucial role of encapsulating data, it is the MessageChannel that decouples message producers from message consumers.

3.1. The MessageChannel Interface
Spring Integration's top-level MessageChannel interface is defined as follows.
public interface MessageChannel { String getName(); boolean send(Message message); boolean send(Message message, long timeout); }

When sending a message, the return value will be true if the message is sent successfully. If the send call times out or is interrupted, then it will return false.

3.1.1. PollableChannel
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.

3.1.2. SubscribableChannel
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); }

Spring-WS (2.0.0.M6)

10

Message Channels

3.2. Message Channel Implementations
Spring Integration provides several different Message Channel implementations. Each is briefly described in the sections below.

3.2.1. PublishSubscribeChannel
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.

3.2.2. QueueChannel
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.

3.2.3. 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 '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.

3.2.4. RendezvousChannel
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. Spring-WS (2.0.0.M6) 11

Message Channels

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 20, 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.

3.2.5. DirectChannel
The point-to-point semantics but otherwise is more similar to the 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.
DirectChannel PublishSubscribeChannel

has

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.

Spring-WS (2.0.0.M6)

12

Message Channels

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.

3.2.6. ExecutorChannel
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.

3.2.7. ThreadLocalChannel
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.

Spring-WS (2.0.0.M6)

13

Message Channels

3.3. Channel Interceptors
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 Messages are being sent to and received from MessageChannels, those channels provide an opportunity for intercepting the send and receive operations. The ChannelInterceptor strategy interface provides methods for each of those operations:
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); }

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 a RuntimeException). Also, the preReceive method can return 'false' to prevent the receive operation from proceeding.

Note
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. Spring Integration also provides an implementation of the Wire Tap pattern. It is a simple interceptor that sends the Message to another channel without otherwise altering the existing flow. It can be very useful for debugging and monitoring. An example is shown in Section 3.5.10, “Wire Tap”. 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 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.
public class CountingChannelInterceptor extends ChannelInterceptorAdapter { private final AtomicInteger sendCount = new AtomicInteger(); @Override public Message<?> preSend(Message<?> message, MessageChannel channel) { sendCount.incrementAndGet(); return message; } }

Tip
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

Spring-WS (2.0.0.M6)

14

Message Channels

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.

3.4. MessageChannelTemplate
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 from the messaging system. However, sometimes it is necessary to invoke the messaging system from your application code. For convenience when implementing such use-cases, Spring Integration provides a MessageChannelTemplate 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.
MessageChannelTemplate template = new MessageChannelTemplate(); Message reply = template.sendAndReceive(new StringMessage("test"), someChannel);

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.
public boolean send(final Message<?> message, final MessageChannel channel) { ... } public Message<?> sendAndReceive(final Message<?> request, final MessageChannel channel) { .. } public Message<?> receive(final PollableChannel<?> channel) { ... }

Note
A less invasive approach that allows you to invoke simple interfaces with payload and/or header values instead of Message instances is described in Section 16.2, “GatewayProxyFactoryBean”.

3.5. Configuring Message Channels
To create a Message Channel instance, you can use the 'channel' element:
<channel id="exampleChannel"/>

The default channel type is Point to Point. To create a Publish Subscribe channel, use the "publish-subscribe-channel" element:
<publish-subscribe-channel id="exampleChannel"/>

To create a Datatype Channel that only accepts messages containing a certain payload type, provide the

Spring-WS (2.0.0.M6)

15

Message Channels fully-qualified class name in the channel element's datatype attribute:
<channel id="numberChannel" datatype="java.lang.Number"/>

Note that the type check passes for any type that is assignable to the channel's datatype. In other words, the "numberChannel" above would accept messages whose payload is java.lang.Integer or java.lang.Double. Multiple types can be provided as a comma-delimited list:
<channel id="stringOrNumberChannel" datatype="java.lang.String,java.lang.Number"/>

When using the "channel" element without any sub-elements, it will create a DirectChannel instance (a SubscribableChannel). However, you can alternatively provide a variety of "queue" sub-elements to create any of the pollable channel types (as described in Section 3.2, “Message Channel Implementations”). Examples of each are shown below.

3.5.1. DirectChannel Configuration
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>

3.5.2. QueueChannel Configuration
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.

3.5.3. PublishSubscribeChannel Configuration
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):

Spring-WS (2.0.0.M6)

16

Message Channels

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

3.5.4. ExecutorChannel
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>

3.5.5. PriorityChannel Configuration
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:

Spring-WS (2.0.0.M6)

17

Message Channels

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

3.5.6. RendezvousChannel Configuration
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>

3.5.7. ThreadLocalChannel Configuration
The ThreadLocalChannel does not provide any additional configuration options.
<thread-local-channel id="threadLocalChannel"/>

3.5.8. Channel Interceptor Configuration
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.

3.5.9. Global Channel Interceptor Configuration
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 Spring-WS (2.0.0.M6) 18

Message 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 '*'

3.5.10. Wire Tap
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.

Note
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 Section B.4, “Error Handling”.

Spring-WS (2.0.0.M6)

19

Chapter 4. Message Endpoints
The first part of this chapter covers some background theory and reveals quite a bit about the underlying API that drives Spring Integration's various messaging components. This information can be helpful if you want to really understand what's going on behind the scenes. However, if you want to get up and running with the simplified namespace-based configuration of the various elements, feel free to skip ahead to Section 4.4, “Namespace Support” for now. 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 Chapter 3, Message Channels, it's easy to send a Message to a Message Channel. However, receiving is a bit more complicated. The main reason is that there are two types of consumers: Polling Consumers and Event Driven Consumers. 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.

4.1. Message Handler
Spring Integration's MessageHandler interface is implemented by many of the components within the framework. In other words, this is not part of the public API, and a developer would not typically implement MessageHandler directly. Nevertheless, it is used by a Message Consumer for actually handling the consumed Messages, and so being aware of this strategy interface does help in terms of understanding the overall role of a consumer. The interface is defined as follows:
public interface MessageHandler { void handleMessage(Message<?> message); }

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 handle, 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 handlers and allow them to be connected to Message Channels.

4.2. Event Driven Consumer
Because it is the simpler of the two, we will cover the Event Driven Consumer endpoint first. You may recall that the SubscribableChannel interface provides a subscribe() method and that the method accepts a MessageHandler parameter (as shown in Section 3.1.2, “SubscribableChannel”):

Spring-WS (2.0.0.M6)

20

Message Endpoints

subscribableChannel.subscribe(messageHandler);

Since a handler that is subscribed to a channel does not have to actively poll that channel, this is an Event Driven Consumer, and the implementation provided by Spring Integration accepts a a SubscribableChannel and a MessageHandler:
SubscribableChannel channel = (SubscribableChannel) context.getBean("subscribableChannel"); EventDrivenConsumer consumer = new EventDrivenConsumer(channel, exampleHandler);

4.3. Polling Consumer
Spring Integration also provides a PollingConsumer, and it can be instantiated in the same way except that the channel must implement PollableChannel:
PollableChannel channel = (PollableChannel) context.getBean("pollableChannel"); PollingConsumer consumer = new PollingConsumer(channel, exampleHandler);

There are many other configuration options for the Polling Consumer. For example, the trigger is a required property:
PollingConsumer consumer = new PollingConsumer(channel, handler); consumer.setTrigger(new IntervalTrigger(30, TimeUnit.SECONDS));

Spring Integration currently provides two implementations of the Trigger interface: IntervalTrigger and CronTrigger. The IntervalTrigger is typically defined with a simple interval (in milliseconds), but also supports an 'initialDelay' property and a boolean 'fixedRate' property (the default is false, i.e. fixed delay):
IntervalTrigger trigger = new IntervalTrigger(1000); trigger.setInitialDelay(5000); trigger.setFixedRate(true);

The CronTrigger simply requires a valid cron expression (see the Javadoc for details):
CronTrigger trigger = new CronTrigger("*/10 * * * * MON-FRI");

In addition to the trigger, several other polling-related configuration properties may be specified:
PollingConsumer consumer = new PollingConsumer(channel, handler); consumer.setMaxMessagesPerPoll(10); consumer.setReceiveTimeout(5000);

The 'maxMessagesPerPoll' property specifies the maximum number of messages to receive within a given poll operation. This means that the poller will continue calling receive() without waiting until either null is returned or that max is reached. For example, if a poller has a 10 second interval trigger and a 'maxMessagesPerPoll' setting of 25, and it is polling a channel that has 100 messages in its queue, all 100 messages can be retrieved within 40 seconds. It grabs 25, waits 10 seconds, grabs the next 25, and so on. The 'receiveTimeout' property specifies the amount of time the poller should wait if no messages are available when it invokes the receive operation. For example, consider two options that seem similar on the surface but Spring-WS (2.0.0.M6) 21

Message Endpoints

are actually quite different: the first has an interval trigger of 5 seconds and a receive timeout of 50 milliseconds while the second has an interval trigger of 50 milliseconds and a receive timeout of 5 seconds. The first one may receive a message up to 4950 milliseconds later than it arrived on the channel (if that message arrived immediately after one of its poll calls returned). On the other hand, the second configuration will never miss a message by more than 50 milliseconds. The difference is that the second option requires a thread to wait, but as a result it is able to respond much more quickly to arriving messages. This technique, known as "long polling", can be used to emulate event-driven behavior on a polled source. A Polling Consumer may also delegate to a Spring TaskExecutor, and it can be configured to participate in Spring-managed transactions. The following example shows the configuration of both:
PollingConsumer consumer = new PollingConsumer(channel, handler); TaskExecutor taskExecutor = (TaskExecutor) context.getBean("exampleExecutor"); consumer.setTaskExecutor(taskExecutor); PlatformTransactionManager txManager = (PlatformTransationManager) context.getBean("exampleTxManager"); consumer.setTransactionManager(txManager);

The examples above show dependency lookups, but keep in mind that these consumers will most often be configured as Spring bean definitions. In fact, Spring Integration also provides a FactoryBean that creates the appropriate consumer 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
Many of the MessageHandler implementations are also capable of generating reply Messages. As mentioned above, sending Messages is trivial when compared to the Message reception. Nevertheless, when and how many reply Messages are sent depends on the handler type. For example, an Aggregator waits for a number of Messages to arrive and is often configured as a downstream consumer for a Splitter which may generate multiple replies for each Message it handles. 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 AbstractReplyProducingMessageHandler, and it provides a setOutputChannel(..) method.

4.4. Namespace Support
Throughout the reference manual, you will see specific configuration examples for endpoint elements, such as router, transformer, service-activator, and so on. Most of these will support an "input-channel" attribute and many will support an "output-channel" attribute. After being parsed, these endpoint elements produce an instance of either the PollingConsumer or the EventDrivenConsumer depending on the type of the "input-channel" that is referenced: PollableChannel or SubscribableChannel respectively. When the channel is pollable, then the polling behavior is determined based on the endpoint element's "poller" sub-element. For example, a simple interval-based poller with a 1-second interval would be configured like this:
<transformer input-channel="pollable" ref="transformer" output-channel="output"> <poller> <interval-trigger interval="1000"/> </poller> </transformer>

For a poller based on a Cron expression, use the "cron-trigger" child element instead: Spring-WS (2.0.0.M6) 22

Message Endpoints

<transformer input-channel="pollable" ref="transformer" output-channel="output"> <poller> <cron-trigger expression="*/10 * * * * MON-FRI"/> </poller> </transformer>

If the input channel is a PollableChannel, then the poller configuration is required. Specifically, as mentioned above, the 'trigger' is a required property of the PollingConsumer class. Therefore, if you omit the "poller" sub-element for a Polling Consumer endpoint's configuration, an Exception may be thrown. However, it is also possible to create top-level pollers in which case only a "ref" is required:
<poller id="weekdayPoller"> <cron-trigger expression="*/10 * * * * MON-FRI"/> </poller> <transformer input-channel="pollable" ref="transformer" output-channel="output"> <poller ref="weekdayPoller"/> </transformer>

In fact, to simplify the configuration, you can define a global default poller. A single top-level poller within an ApplicationContext may have the default attribute with a value of "true". In that case, any endpoint with a PollableChannel for its input-channel that is defined within the same ApplicationContext and has no explicitly configured 'poller' sub-element will use that default.
<poller id="defaultPoller" default="true" max-messages-per-poll="5"> <interval-trigger interval="3" time-unit="SECONDS"/> </poller> <!-- No <poller/> sub-element is necessary since there is a default --> <transformer input-channel="pollable" ref="transformer" output-channel="output"/>

Spring Integration also provides transaction support for the pollers so that each receive-and-forward operation can be performed as an atomic unit-of-work. To configure transactions for a poller, simply add the <transactional/> sub-element. The attributes for this element should be familiar to anyone who has experience with Spring's Transaction management:
<poller> <interval-trigger interval="1000"/> <transactional transaction-manager="txManager" propagation="REQUIRED" isolation="REPEATABLE_READ" timeout="10000" read-only="false"/> </poller>

The polling threads may be executed by any instance of Spring's TaskExecutor abstraction. This enables concurrency for an endpoint or group of endpoints. As of Spring 3.0, there is a "task" namespace in the core Spring Framework, and its <executor/> element supports the creation of a simple thread pool executor. That element accepts attributes for common concurrency settings such as pool-size and queue-capacity. Configuring a thread-pooling executor can make a substantial difference in how the endpoint performs under load. These settings are available per-endpoint since the performance of an endpoint is one of the major factors to consider (the other major factor being the expected volume on the channel to which the endpoint subscribes). To enable concurrency for a polling endpoint that is configured with the XML namespace support, provide the 'task-executor' reference on its <poller/> element and then provide one or more of the properties shown below:
<poller task-executor="pool"/>

Spring-WS (2.0.0.M6)

23

Message Endpoints

<interval-trigger interval="5" time-unit="SECONDS"/> </poller> <task:executor id="pool" pool-size="5-25" queue-capacity="20" keep-alive="120"/>

If no 'task-executor' is provided, the consumer's handler will be invoked in the caller's thread. Note that the "caller" is usually the default TaskScheduler (see Section B.3, “Configuring the Task Scheduler”). Also, keep in mind that the 'task-executor' attribute can provide a reference to any implementation of Spring's TaskExecutor interface by specifying the bean name. The "executor" element above is simply provided for convenience. As mentioned in the background section for Polling Consumers above, you can also configure a Polling Consumer in such a way as to emulate event-driven behavior. With a long receive-timeout and a short interval-trigger, you can ensure a very timely reaction to arriving messages even on a polled message source. Note that this will only apply to sources that have a blocking wait call with a timeout. For example, the File poller does not block, each receive() call returns immediately and either contains new files or not. Therefore, even if a poller contains a long receive-timeout, that value would never be usable in such a scenario. On the other hand when using Spring Integration's own queue-based channels, the timeout value does have a chance to participate. The following example demonstrates how a Polling Consumer will receive Messages nearly instantaneously.
<service-activator input-channel="someQueueChannel" output-channel="output"> <poller receive-timeout="30000"> <interval-trigger interval="10"/> </poller> </service-activator>

Using this approach does not carry much overhead since internally it is nothing more then a timed-wait thread which does not require nearly as much CPU resource usage as a thrashing, infinite while loop for example.

4.5. Payload Type Conversion
Throughout the reference manual, you will also see specific configuration and implementation examples of various endpoints which can accept a Message or any arbitrary Object as an input parameter. In the case of an Object, such parameter will be mapped to a Message payload or part of the payload or header (when using Spring Expression Language). However there are times when the type of input parameter of the endpoint method does not match the type of the payload or its part. In this scenario we need to perform type conversion. Spring Integration provides a convenient way for registering type converters (using Spring 3.x ConversionService) within its own instance of the conversion service bean named integrationConversionService which is automatically created as soon as the first converter is defined. To register such converter all you need is to implement org.springframework.core.convert.converter.Converter and register via cionvinient namespace support:
<int:converter ref="sampleConverter"/> <bean id="sampleConverter" class="foo.bar.TestConverter"/>

or
<int:converter> <bean class="org.springframework.integration.config.xml.ConverterParserTests$TestConverter3"/> </int:converter>

Spring-WS (2.0.0.M6)

24

Chapter 5. Service Activator
5.1. Introduction
The Service Activator is the endpoint type for connecting any Spring-managed Object to an input channel so that it may play the role of a service. If the service produces output, it may also be connected to an output channel. Alternatively, an output producing service may be located at the end of a processing pipeline or message flow in which case, the inbound Message's "replyChannel" header can be used. This is the default behavior if no output channel is defined, and as with most of the configuration options you'll see here, the same behavior actually applies for most of the other components we have seen.

5.2. The <service-activator/> Element
To create a Service Activator, use the 'service-activator' element with the 'input-channel' and 'ref' attributes:
<service-activator input-channel="exampleChannel" ref="exampleHandler"/>

The configuration above assumes that "exampleHandler" either contains a single method annotated with the @ServiceActivator annotation or that it contains only one public method at all. To delegate to an explicitly defined method of any object, simply add the "method" attribute.
<service-activator input-channel="exampleChannel" ref="somePojo" method="someMethod"/>

In either case, when the service method returns a non-null value, the endpoint will attempt to send the reply message to an appropriate reply channel. To determine the reply channel, it will first check if an "output-channel" was provided in the endpoint configuration:
<service-activator input-channel="exampleChannel" output-channel="replyChannel" ref="somePojo" method="someMethod"/>

If no "output-channel" is available, it will then check the Message's REPLY_CHANNEL header value. If that value is available, it will then check its type. If it is a MessageChannel, the reply message will be sent to that channel. If it is a String, then the endpoint will attempt to resolve the channel name to a channel instance. If the channel cannot be resolved, then a ChannelResolutionException will be thrown. The argument in the service method could be either a Message or an arbitrary type. If the latter, then it will be assumed that it is a Message payload, which will be extracted from the message and injected into such service method. This is generally the recommended approach as it follows and promotes a POJO model when working with Spring Integration. Arguments may also have @Header, @Headers or @MessageMapping annotations as described in Section B.5, “Annotation Support”

Note
Since v1.0.3 of Spring Integration, the service method is not required to have an argument at all, which means you can now implement event-style Service Activators, where all you care about is an invocation of the service method, not worrying about the contents of the message. Think of it as a NULL JMS message. An example use-case for such an implementation could be a simple counter/monitor of messages deposited on the input channel. Using a "ref" attribute is generally recommended if the custom Service Activator handler implementation can Spring-WS (2.0.0.M6) 25

Service Activator

be reused in other <service-activator> definitions. However if the custom Service Activator handler implementation should be scoped to a single definition of the <service-activator>, you can use an inner bean definition:
<service-activator id="exampleServiceActivator" input-channel="inChannel" output-channel = "outChannel" method="foo"> <beans:bean class="org.foo.ExampleServiceActivator"/> </service-activator>

Note
Using both the "ref" attribute and an inner handler definition in the same <service-activator> configuration is not allowed, as it creates an ambiguous condition and will result in an Exception being thrown.

Spring-WS (2.0.0.M6)

26

Chapter 6. Channel Adapter
A Channel Adapter is a Message Endpoint that enables connecting a single sender or receiver to a Message Channel. Spring Integration provides a number of adapters out of the box to support various transports, such as JMS, File, HTTP, Web Services, and Mail. Those will be discussed in upcoming chapters of this reference guide. However, this chapter focuses on the simple but flexible Method-invoking Channel Adapter support. There are both inbound and outbound adapters, and each may be configured with XML elements provided in the core namespace.

6.1. The <inbound-channel-adapter> element
An "inbound-channel-adapter" element can invoke any method on a Spring-managed Object and send a non-null return value to a MessageChannel after converting it to a Message. When the adapter's subscription is activated, a poller will attempt to receive messages from the source. The poller will be scheduled with the TaskScheduler according to the provided configuration. To configure the polling interval or cron expression for an individual channel-adapter, provide a 'poller' element with either an 'interval-trigger' (in milliseconds) or 'cron-trigger' sub-element.
<inbound-channel-adapter ref="source1" method="method1" channel="channel1"> <poller> <interval-trigger interval="5000"/> </poller> </inbound-channel-adapter> <inbound-channel-adapter ref="source2" method="method2" channel="channel2"> <poller> <cron-trigger expression="30 * 9-17 * * MON-FRI"/> </poller> </channel-adapter>

Note
If no poller is provided, then a single default poller must be registered within the context. See Section 4.4, “Namespace Support” for more detail.

6.2. The <outbound-channel-adapter/> element
An "outbound-channel-adapter" element can also connect a MessageChannel to any POJO consumer method that should be invoked with the payload of Messages sent to that channel.
<outbound-channel-adapter channel="channel1" ref="target1" method="method1"/>

If the channel being adapted is a PollableChannel, provide a poller sub-element:
<outbound-channel-adapter channel="channel2" ref="target2" method="method2"> <poller> <interval-trigger interval="3000"/> </poller> </outbound-channel-adapter> <beans:bean id="target1" class="org.bar.Foo"/>

Using a "ref" attribute is generally recommended if the POJO consumer implementation can be reused in other <outbound-channel-adapter> definitions. However if the consumer implementation should be scoped to a

Spring-WS (2.0.0.M6)

27

Channel Adapter single definition of the <outbound-channel-adapter>, you can define it as inner bean:
<outbound-channel-adapter channel="channel2" method="method2"> <beans:bean class="org.bar.Foo"/> </outbound-channel-adapter>

Note
Using attribute and an inner handler definition in the same configuration is not allowed, as it creates an ambiguous condition and will result in an Exception being thrown.
<outbound-channel-adapter>

both

the

"ref"

Any Channel Adapter can be created without a "channel" reference in which case it will implicitly create an instance of DirectChannel. The created channel's name will match the "id" attribute of the <inbound-channel-adapter/> or <outbound-channel-adapter element. Therefore, if the "channel" is not provided, the "id" is required.

Spring-WS (2.0.0.M6)

28

Chapter 7. Router
7.1. Router Implementations
Since content-based routing often requires some domain-specific logic, most use-cases will require Spring Integration's options for delegating to POJOs using the XML namespace support and/or Annotations. Both of these are discussed below, but first we present a couple implementations that are available out-of-the-box since they fulfill generic, but common, requirements.

7.1.1. PayloadTypeRouter
A PayloadTypeRouter will send Messages to the channel as defined by payload-type mappings.
<bean id="payloadTypeRouter" class="org.springframework.integration.router.PayloadTypeRouter"> <property name="payloadTypeChannelMap"> <map> <entry key="java.lang.String" value-ref="stringChannel"/> <entry key="java.lang.Integer" value-ref="integerChannel"/> </map> </property> </bean>

Configuration of PayloadTypeRouter is also supported via the namespace provided by Spring Integration (see Section B.2, “Namespace Support”), which essentially simplifies configuration by combining <router/> configuration and its corresponding implementation defined using <bean/> element into a single and more concise configuration element. The example below demonstrates PayloadTypeRouter configuration which is equivalent to the one above using Spring Integration's namespace support:
<payload-type-router input-channel="routingChannel"> <mapping type="java.lang.String" channel="stringChannel" /> <mapping type="java.lang.Integer" channel="integerChannel" /> </payload-type-router>

7.1.2. HeaderValueRouter
A HeaderValueRouter will send Messages to the channel based on the individual header value mappings. When HeaderValueRouter is created it is initialized with the name of the header to be evaluated, using constructor-arg. The value of the header could be one of two things: 1. Arbitrary value 2. Channel name If arbitrary value, then a channelResolver should be provided to map header values to channel names. The example below uses MapBasedChannelResolver to set up a map of header values to channel names.
<bean id="myHeaderValueRouter" class="org.springframework.integration.router.HeaderValueRouter"> <constructor-arg value="someHeaderName" /> <property name="channelResolver"> <bean class="org.springframework.integration.channel.MapBasedChannelResolver"> <property name="channelMap"> <map> <entry key="someHeaderValue" value-ref="channelA" /> <entry key="someOtherHeaderValue" value-ref="channelB" />

Spring-WS (2.0.0.M6)

29

Router

</map> </property> </bean> </property> </bean>

If channelResolver is not specified, then the header value will be treated as a channel name making configuration much simpler, where no channelResolver needs to be specified.
<bean id="myHeaderValueRouter" class="org.springframework.integration.router.HeaderValueRouter"> <constructor-arg value="someHeaderName" /> </bean>

Similar to the PayloadTypeRouter, configuration of HeaderValueRouter is also supported via namespace support provided by Spring Integration (see Section B.2, “Namespace Support”). The example below demonstrates two types of namespace-based configuration of HeaderValueRouter which are equivalent to the ones above using Spring Integration namespace support: 1. Configuration where mapping of header values to channels is required
<header-value-router input-channel="routingChannel" header-name="testHeader"> <mapping value="someHeaderValue" channel="channelA" /> <mapping value="someOtherHeaderValue" channel="channelB" /> </header-value-router>

2. Configuration where mapping of header values is not required if header values themselves represent the channel names
<header-value-router input-channel="routingChannel" header-name="testHeader"/>

Note
The two router implementations shown above share some common properties, such as "defaultOutputChannel" and "resolutionRequired". If "resolutionRequired" is set to "true", and the router is unable to determine a target channel (e.g. there is no matching payload for a PayloadTypeRouter and no "defaultOutputChannel" has been specified), then an Exception will be thrown.

7.1.3. RecipientListRouter
A RecipientListRouter will send each received Message to a statically-defined list of Message Channels:
<bean id="recipientListRouter" class="org.springframework.integration.router.RecipientListRouter"> <property name="channels"> <list> <ref bean="channel1"/> <ref bean="channel2"/> <ref bean="channel3"/> </list> </property> </bean>

Configuration for RecipientListRouter is also supported via namespace support provided by Spring Integration (see Section B.2, “Namespace Support”). The example below demonstrates namespace-based

Spring-WS (2.0.0.M6)

30

Router configuration of RecipientListRouter and all the supported attributes using Spring Integration namespace support:
<recipient-list-router id="customRouter" input-channel="routingChannel" timeout="1234" ignore-send-failures="true" apply-sequence="true"> <recipient channel="channel1"/> <recipient channel="channel2"/> </recipient-list-router>

Note
The 'apply-sequence' flag here has the same affect as it does for a publish-subscribe-channel, and like publish-subscribe-channel it is disabled by default on the recipient-list-router. Refer to Section 3.5.3, “PublishSubscribeChannel Configuration” for more information.

7.2. The <router> element
The "router" element provides a simple way to connect a router to an input channel, and also accepts the optional default output channel. The "ref" may provide the bean name of a custom Router implementation (extending AbstractMessageRouter):
<router ref="payloadTypeRouter" input-channel="input1" default-output-channel="defaultOutput1"/> <router ref="recipientListRouter" input-channel="input2" default-output-channel="defaultOutput2"/> <router ref="customRouter" input-channel="input3" default-output-channel="defaultOutput3"/> <beans:bean id="customRouterBean class="org.foo.MyCustomRouter"/>

Alternatively, the "ref" may point to a simple Object that contains the @Router annotation (see below), or the "ref" may be combined with an explicit "method" name. When specifying a "method", the same behavior applies as described in the @Router annotation section below.
<router input-channel="input" ref="somePojo" method="someMethod"/>

Using a "ref" attribute is generally recommended if the custom router implementation can be reused in other <router> definitions. However if the custom router implementation should be scoped to a concrete definition of the <router>, you can provide an inner bean definition:
<router method="someMethod" input-channel="input3" default-output-channel="defaultOutput3"> <beans:bean class="org.foo.MyCustomRouter"/> </router>

Note
Using both the "ref" attribute and an inner handler definition in the same <router> configuration is not allowed, as it creates an ambiguous condition and will result in an Exception being thrown.

7.3. The @Router Annotation
When using the @Router annotation, the annotated method can return either the MessageChannel or String type. In the case of the latter, the endpoint will resolve the channel name as it does for the default output.

Spring-WS (2.0.0.M6)

31

Router Additionally, the method can return either a single value or a collection. When a collection is returned, the reply message will be sent to multiple channels. To summarize, the following method signatures are all valid.
@Router public MessageChannel route(Message message) {...} @Router public List<MessageChannel> route(Message message) {...} @Router public String route(Foo payload) {...} @Router public List<String> route(Foo payload) {...}

In addition to payload-based routing, a common requirement is to route based on metadata available within the message header as either a property or attribute. Rather than requiring use of the Message type as the method parameter, the @Router annotation may also use the @Header parameter annotation that is documented in Section B.5, “Annotation Support”.
@Router public List<String> route(@Header("orderStatus") OrderStatus status)

Note
For routing of XML-based Messages, including XPath support, see Chapter 31, Dealing with XML Payloads.

Spring-WS (2.0.0.M6)

32

Chapter 8. Filter
8.1. Introduction
Message Filters are used to decide whether a Message should be passed along or dropped based on some criteria such as a Message Header value or even content within the Message itself. Therefore, a Message Filter is similar to a router, except that for each Message received from the filter's input channel, that same Message may or may not be sent to the filter's output channel. Unlike the router, it makes no decision regarding which Message Channel to send to but only decides whether to send.

Note
As you will see momentarily, the Filter does also support a discard channel, so in certain cases it can play the role of a very simple router (or "switch") based on a boolean condition. In Spring Integration, a Message Filter may be configured as a Message Endpoint that delegates to some implementation of the MessageSelector interface. That interface is itself quite simple:
public interface MessageSelector { boolean accept(Message<?> message); }

The MessageFilter constructor accepts a selector instance:
MessageFilter filter = new MessageFilter(someSelector);

In combination with the namespace and SpEL very powerful filters can be configured with very little java code.

8.2. The <filter> Element
The <filter> element is used to create a Message-selecting endpoint. In addition to "input-channel" and "output-channel" attributes, it requires a "ref". The "ref" may point to a MessageSelector implementation:
<filter input-channel="input" ref="selector" output-channel="output"/> <bean id="selector" class="example.MessageSelectorImpl"/>

Alternatively, the "method" attribute can be added at which point the "ref" may refer to any object. The referenced method may expect either the Message type or the payload type of inbound Messages. The return value of the method must be a boolean value. Any time the method returns 'true', the Message will be passed along to the output-channel.
<filter input-channel="input" output-channel="output" ref="exampleObject" method="someBooleanReturningMethod"/> <bean id="exampleObject" class="example.SomeObject"/>

If the selector or adapted POJO method returns false, there are a few settings that control the fate of the rejected Message. By default (if configured like the example above), the rejected Messages will be silently dropped. If rejection should instead indicate an error condition, then set the 'throw-exception-on-rejection' flag to true:

Spring-WS (2.0.0.M6)

33

Filter

<filter input-channel="input" ref="selector" output-channel="output" throw-exception-on-rejection="true"/>

If you want the rejected messages to go to a specific channel, provide that reference as the 'discard-channel':
<filter input-channel="input" ref="selector" output-channel="output" discard-channel="rejectedMessages"/>

Note
A common usage for Message Filters is in conjunction with a Publish Subscribe Channel. Many filter endpoints may be subscribed to the same channel, and they decide whether or not to pass the Message for the next endpoint which could be any of the supported types (e.g. Service Activator). This provides a reactive alternative to the more proactive approach of using a Message Router with a single Point-to-Point input channel and multiple output channels. Using a "ref" attribute is generally recommended if the custom filter implementation can be reused in other <filter> definitions. However if the custom filter implementation should be scoped to a single <filter> element, provide an inner bean definition:
<filter method="someMethod" input-channel="inChannel" output-channel="outChannel"> <beans:bean class="org.foo.MyCustomFilter"/> </filter>

Note
Using both the "ref" attribute and an inner handler definition in the same <filter> configuration is not allowed, as it creates an ambiguous condition, and it will therefore result in an Exception being thrown. With the introduction of SpEL Spring Integration has added the expression attribute to the filter element. It can be used to avoid Java entirely for simple filters.
<filter input-channel="input" expression="payload.equals(nonsense)"/>

The string passed as the expression attribute will be evaluated as a SpEL expression in the context of the message. If it is needed to include the result of an expression in the scope of the application context you can use the #{} notation as defined in the SpEL reference documentation SpEL reference documentation .
<filter input-channel="input" expression="payload.matches(#{filterPatterns.nonsensePattern})"/>

Spring-WS (2.0.0.M6)

34

Chapter 9. Transformer
9.1. Introduction
Message Transformers play a very important role in enabling the loose-coupling of Message Producers and Message Consumers. Rather than requiring every Message-producing component to know what type is expected by the next consumer, Transformers can be added between those components. Generic transformers, such as one that converts a String to an XML Document, are also highly reusable. For some systems, it may be best to provide a Canonical Data Model, but Spring Integration's general philosophy is not to require any particular format. Rather, for maximum flexibility, Spring Integration aims to provide the simplest possible model for extension. As with the other endpoint types, the use of declarative configuration in XML and/or Annotations enables simple POJOs to be adapted for the role of Message Transformers. These configuration options will be described below.

Note
For the same reason of maximizing flexibility, Spring does not require XML-based Message payloads. Nevertheless, the framework does provide some convenient Transformers for dealing with XML-based payloads if that is indeed the right choice for your application. For more information on those transformers, see Chapter 31, Dealing with XML Payloads.

9.2. The <transformer> Element
The <transformer> element is used to create a Message-transforming endpoint. In addition to "input-channel" and "output-channel" attributes, it requires a "ref". The "ref" may either point to an Object that contains the @Transformer annotation on a single method (see below) or it may be combined with an explicit method name value provided via the "method" attribute.
<transformer id="testTransformer" ref="testTransformerBean" input-channel="inChannel" method="transform" output-channel="outChannel"/> <beans:bean id="testTransformerBean" class="org.foo.TestTransformer" />

Using a "ref" attribute is generally recommended if the custom transformer handler implementation can be reused in other <transformer> definitions. However if the custom transformer handler implementation should be scoped to a single definition of the <transformer>, you can define an inner bean definition:
<transformer id="testTransformer" input-channel="inChannel" method="transform" output-channel="outChannel"> <beans:bean class="org.foo.TestTransformer"/> </transformer>

Note
Using both the "ref" attribute and an inner handler definition in the same <transformer> configuration is not allowed, as it creates an ambiguous condition and will result in an Exception being thrown. The method that is used for transformation may expect either the Message type or the payload type of inbound Messages. It may also accept Message header values either individually or as a full map by using the @Header Spring-WS (2.0.0.M6) 35

Transformer and @Headers parameter annotations respectively. The return value of the method can be any type. If the return value is itself a Message, that will be passed along to the transformer's output channel. If the return type is a Map, and the original Message payload was not a Map, the entries in that Map will be added to the Message headers of the original Message (the keys must be Strings). If the return value is null, then no reply Message will be sent (effectively the same behavior as a Message Filter returning false). Otherwise, the return value will be sent as the payload of an outbound reply Message. There are a also a few Transformer implementations available out of the box. Because, it is fairly common to use the toString() representation of an Object, Spring Integration provides an ObjectToStringTransformer whose output is a Message with a String payload. That String is the result of invoking the toString operation on the inbound Message's payload.
<object-to-string-transformer input-channel="in" output-channel="out"/>

A potential example for this would be sending some arbitrary object to the 'outbound-channel-adapter' in the file namespace. Whereas that Channel Adapter only supports String, byte-array, or java.io.File payloads by default, adding this transformer immediately before the adapter will handle the necessary conversion. Of course, that works fine as long as the result of the toString() call is what you want to be written to the File. Otherwise, you can just provide a custom POJO-based Transformer via the generic 'transformer' element shown previously.

Tip
When debugging, this transformer is not typically necessary since the 'logging-channel-adapter' is capable of logging the Message payload. Refer to Section 3.5.10, “Wire Tap” for more detail. If you need to serialize an Object to a byte array or deserialize a byte array back into an Object, Spring Integration provides symmetrical serialization transformers.
<payload-serializing-transformer input-channel="objectsIn" output-channel="bytesOut"/> <payload-deserializing-transformer input-channel="bytesIn" output-channel="objectsOut"/>

If you only need to add headers to a Message, and they are not dynamically determined by Message content, then referencing a custom implementation may be overkill. For that reason, Spring Integration provides the 'header-enricher' element.
<header-enricher input-channel="in" output-channel="out"> <header name="foo" value="123"/> <header name="bar" ref="someBean"/> </header-enricher>

As added convenience, Spring Integration also provides Object-to-Map and Map-to-Object transformers which utilize Spring Expression Language (SpEL) to serialize and de-serialize the object graphs. Object hierarchy is introspected to the most primitive types (e.g., String, int etc.). The path to this type is described via SpEL, which becomes the keykey in the transformed Map with primitive type being the value. For example:
public class Parent{ private Child child; private String name; // setters and getters are omitted } public class Child{ private String name; private List<String> nickNames; // setters and getters are omitted

Spring-WS (2.0.0.M6)

36

Transformer

}

... will be transformed to a Map which looks like this: {person.name=George, person.child.name=Jenna,
person.child.nickNames[0]=Bimbo . . . etc}

SpEL-based Map allows you to describe the object structure without sharing the actual types allowing you to restore/rebuild the object graph into a differently typed Object graph as long as you maintain the structure. For example: The above structure could be easily restored back to the following Object graph via Map-to-Object transformer:
public class Father{ private Kid child; private String name; // setters and getters are omitted } public class Kid{ private String name; private List<String> nickNames; // setters and getters are omitted }

To configure these transformers, Spring Integration provides namespace support Object-to-Map:
<object-to-map-transformer input-channel="directInput" output-channel="output"/>

Map-to-Object
<int:map-to-object-transformer input-channel="input" output-channel="output" type="org.foo.Person"/>

or
<int:map-to-object-transformer input-channel="inputA" output-channel="outputA" ref="person"/> <bean id="person" class="org.foo.Person" scope="prototype"/>

Note
NOTE: 'ref' and 'type' attributes are mutually exclusive. You can only use either one. Also, if using 'ref' attribute you must point to a 'prototype' scoped bean, otherwise BeanCreationException will be thrown.

9.3. The @Transformer Annotation
The @Transformer annotation can also be added to methods that expect either the Message type or the message payload type. The return value will be handled in the exact same way as described above in the section describing the <transformer> element.
@Transformer Order generateOrder(String productId) { return new Order(productId); }

Transformer methods may also accept the @Header and @Headers annotations that is documented in

Spring-WS (2.0.0.M6)

37

Transformer Section B.5, “Annotation Support”
@Transformer Order generateOrder(String productId, @Header("customerName") String customer) { return new Order(productId, customer); }

Spring-WS (2.0.0.M6)

38

Chapter 10. Splitter
10.1. Introduction
The Splitter is a component whose role is to partition a message in several parts, and send the resulting messages to be processed independently. Very often, they are upstream producers in a pipeline that includes an Aggregator.

10.2. Programming model
The API for performing splitting consists from one base class, AbstractMessageSplitter, which is a MessageHandler implementation, encapsulating features which are common to splitters, such as filling in the appropriate message headers CORRELATION_ID, SEQUENCE_SIZE, and SEQUENCE_NUMBER on the messages that are produced. This allows to track down the messages and the results of their processing (in a typical scenario, these headers would be copied over to the messages that are produced by the various transforming endpoints), and use them, for example, in a Composed Message Processor scenario. An excerpt from AbstractMessageSplitter can be seen below:
public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageConsumer { ... protected abstract Object splitMessage(Message<?> message); }

For implementing a specific Splitter in an application, a developer can extend AbstractMessageSplitter and implement the splitMessage method, thus defining the actual logic for splitting the messages. The return value can be one of the following: ? a Collection (or subclass thereof) or an array of Message objects - in this case the messages will be sent as such (after the CORRELATION_ID, SEQUENCE_SIZE and SEQUENCE_NUMBER are populated). Using this approach gives more control to the developer, for example for populating custom message headers as part of the splitting process. ? a Collection (or subclass thereof) or an array of non-Message objects - works like the prior case, except that each collection element will be used as a Message payload. Using this approach allows developers to focus on the domain objects without having to consider the Messaging system and produces code that is easier to test. ? a Message or non-Message object (but not a Collection or an Array) - it works like the previous cases, except that there is a single message to be sent out. In Spring Integration, any POJO can implement the splitting algorithm, provided that it defines a method that accepts a single argument and has a return value. In this case, the return value of the method will be interpreted as described above. The input argument might either be a Message or a simple POJO. In the latter case, the splitter will receive the payload of the incoming message. Since this decouples the code from the Spring Integration API and will typically be easier to test, it is the recommended approach.

Spring-WS (2.0.0.M6)

39

Splitter

10.3. Configuring a Splitter using XML
A splitter can be configured through XML as follows:
<channel id="inputChannel"/> <splitter id="splitter" ? ref="splitterBean" ? method="split" ? input-channel="inputChannel" ? output-channel="outputChannel" ?/> <channel id="outputChannel"/> <beans:bean id="splitterBean" class="sample.PojoSplitter"/>

? ?

? ? ?

The id of the splitter is optional. A reference to a bean defined in the application context. The bean must implement the splitting logic as described in the section above. Optional. If reference to a bean is not provided, then it is assumed that the payload of the Message that arrived on the input-channel is an implementation of java.util.Collection and the default splitting logic will be applied on such Collection, incorporating each individual element into a Message and depositing it on the output-channel. The method (defined on the bean specified above) that implements the splitting logic. Optional. The input channel of the splitter. Required. The channel where the splitter will send the results of splitting the incoming message. Optional (because incoming messages can specify a reply channel themselves).

Using a "ref" attribute is generally recommended if the custom splitter handler implementation can be reused in other <splitter> definitions. However if the custom splitter handler implementation should be scoped to a single definition of the <splitter>, you can configure an inner bean definition:
<splitter id="testSplitter" input-channel="inChannel" method="split" output-channel="outChannel"> <beans:bean class="org.foo.TestSplitter"/> </spliter>

Note
Using both a "ref" attribute and an inner handler definition in the same <splitter> configuration is not allowed, as it creates an ambiguous condition and will result in an Exception being thrown.

10.4. Configuring a Splitter with Annotations
The @Splitter annotation is applicable to methods that expect either the Message type or the message payload type, and the return values of the method should be a collection of any type. If the returned values are not actual Message objects, then each of them will be sent as the payload of a message. Those messages will be sent to the output channel as designated for the endpoint on which the @Splitter is defined.
@Splitter List<LineItem> extractItems(Order order) { return order.getItems() }

Spring-WS (2.0.0.M6)

40

Chapter 11. Aggregator
11.1. Introduction
Basically a mirror-image of the Splitter, the Aggregator is a type of Message Handler that receives multiple Messages and combines them into a single Message. In fact, Aggregators are often downstream consumers in a pipeline that includes a Splitter. Technically, the Aggregator is more complex than a Splitter, because it is required to maintain state (the Messages to be aggregated), to decide when the complete group of Messages is available. In order to do this it requires a MessageStore

11.2. Functionality
The Aggregator combines a group of related messages, by correlating and storing them, until the group is deemed complete. At that point, the Aggregator will create a single message by processing the whole group, and will send that aggregated message as output. An main aspect of implementing an Aggregator is providing the logic that has to be executed when the aggregation (creation of a single message out of many) takes place. The other two aspects are correlation and release In Spring Integration, the grouping of the messages for aggregation (correlation) is done by default based on their CORRELATION_ID message header (i.e. the messages with the same CORRELATION_ID will be grouped together). However, this can be customized, and the users can opt for other ways of specifying how the messages should be grouped together, by using a CorrelationStrategy (see below). To determine whether or not a group of messages may be processed, a ReleaseStrategy is consulted. The default release strategy for aggregator will release groups that have all messages from the sequence, but this can be entirely customized

11.3. Programming model
The Aggregation API consists of a number of classes: ? The interface
MessageGroupProcessor

AbstractAggregatingMessageGroupProcessor MethodInvokingAggregatingMessageGroupProcessor

and and

related its

base

class subclass

? The ReleaseStrategy interface and its default implementation SequenceSizeReleaseStrategy ? The CorrelationStrategy interface and its default implementation HeaderAttributeCorrelationStrategy

11.3.1. CorrelatingMessageHandler
The CorrelatingMessageHandler is a MessageHandler implementation, encapsulating the common functionalities of an Aggregator (and other correlating use cases), which are:

Spring-WS (2.0.0.M6)

41

Aggregator

? correlating messages into a group to be aggregated ? maintaining those messages in a MessageStore until the group may be released ? deciding when the group is in fact may be released ? processing the released group into a single aggregated message ? recognizing and responding to an expired group The responsibility of deciding how the messages should be grouped together is delegated to a CorrelationStrategy instance. The responsibility of deciding whether the message group can be released is delegated to a ReleaseStrategy instance. Here is a brief highlight of the base AbstractAggregatingMessageGroupProcessor (the responsibility of implementing the aggregateMessages method is left to the developer):
public abstract class AbstractAggregatingMessageGroupProcessor implements MessageGroupProcessor { protected Map<String, Object> aggregateHeaders(MessageGroup group) { .... } protected abstract Object aggregatePayloads(MessageGroup group); }

The CorrelationStrategy is owned by the CorrelatingMessageHandler and it has a default value based on the correlation ID message header:
private volatile CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID);

When appropriate, the simplest option is the DefaultAggregatingMessageGroupProcessor. It creates a single Message whose payload is a List of the payloads received for a given group. It uses the default CorrelationStrategy and CompletionStrategy as shown above. This works well for simple Scatter Gather implementations with either a Splitter, Publish Subscribe Channel, or Recipient List Router upstream.

Note
When using a Publish Subscribe Channel or Recipient List Router in this type of scenario, be sure to enable the flag to apply-sequence. That will add the necessary headers (correlation id, sequence number and sequence size). That behavior is enabled by default for Splitters in Spring Integration, but it is not enabled for the Publish Subscribe Channel or Recipient List Router because those components may be used in a variety of contexts where those headers are not necessary. When implementing a specific aggregator object for an application, a developer can extend AbstractAggregatingMessageGroupProcessor and implement the aggregatePayloads method. However, there are better suited (which reads, less coupled to the API) solutions for implementing the aggregation logic, which can be configured easily either through XML or through annotations. In general, any ordinary Java class (i.e. POJO) can implement the aggregation algorithm. For doing so, it must provide a method that accepts as an argument a single java.util.List (parametrized lists are supported as well). This method will be invoked for aggregating messages, as follows: ? if the argument is a parametrized java.util.List, and the parameter type is assignable to Message, then the

Spring-WS (2.0.0.M6)

42

Aggregator

whole list of messages accumulated for aggregation will be sent to the aggregator ? if the argument is a non-parametrized java.util.List or the parameter type is not assignable to Message, then the method will receive the payloads of the accumulated messages ? if the return type is not assignable to Message, then it will be treated as the payload for a Message that will be created automatically by the framework.

Note
In the interest of code simplicity, and promoting best practices such as low coupling, testability, etc., the preferred way of implementing the aggregation logic is through a POJO, and using the XML or annotation support for setting it up in the application.

11.3.2. ReleaseStrategy
The ReleaseStrategy interface is defined as follows:
public interface ReleaseStrategy { boolean canRelease(MessageGroup messages); }

In general, any ordinary Java class (i.e. POJO) can implement the completion decision mechanism. For doing so, it must provide a method that accepts as an argument a single java.util.List (parametrized lists are supported as well), and returns a boolean value. This method will be invoked after the arrival of a new message, to decide whether the group is complete or not, as follows: ? if the argument is a parametrized java.util.List, and the parameter type is assignable to Message, then the whole list of messages accumulated in the group will be sent to the method ? if the argument is a non-parametrized java.util.List or the parameter type is not assignable to Message, then the method will receive the payloads of the accumulated messages ? the method must return true if the message group is ready for aggregation, and false otherwise. When the group is released for aggregation, all its unmarked messages are processed and then marked so they will not be processed again. If the group is also complete (i.e. if all messages from a sequence have arrived or if there is no sequence defined) then the group is removed from the message store. Partial sequences can be released, in which case the next time the ReleaseStrategy is called it will be presented with a group containing marked messages (already processed) and unmarked messages (a potential new partial sequence) Spring an out-of-the box implementation for ReleaseStrategy, the SequenceSizerReleaseStrategy. This implementation uses the SEQUENCE_NUMBER and SEQUENCE_SIZE of the arriving messages for deciding when a message group is complete and ready to be aggregated. As shown above, it is also the default strategy. Integration provides

11.3.3. CorrelationStrategy
The CorrelationStrategy interface is defined as follows:
public interface CorrelationStrategy {

Spring-WS (2.0.0.M6)

43

Aggregator

Object getCorrelationKey(Message<?> message); }

The method shall return an Object which represents the correlation key used for grouping messages together. The key must satisfy the criteria used for a key in a Map with respect to the implementation of equals() and hashCode(). In general, any ordinary Java class (i.e. POJO) can implement the correlation decision mechanism, and the rules for mapping a message to a method's argument (or arguments) are the same as for a ServiceActivator (including support for @Header annotations). The method must return a value, and the value must not be null. Spring out-of-the box implementation for CorrelationStrategy, the This implementation returns the value of one of the message headers (whose name is specified by a constructor argument) as the correlation key. By default, the correlation strategy is a HeaderAttributeCorrelationStrategy returning the value of the CORRELATION_ID header attribute.
HeaderAttributeCorrelationStrategy.

Integration

provides

an

11.4. Configuring an Aggregator with XML
Spring Integration supports the configuration of an aggregator via XML through the <aggregator/> element. Below you can see an example of an aggregator with all optional parameters defined.
<channel id="inputChannel"/> <aggregator id="completelyDefinedAggregator" ? input-channel="inputChannel" ? output-channel="outputChannel" ? discard-channel="discardChannel" ? ref="aggregatorBean" ? method="add" ? release-strategy="releaseStrategyBean" ? release-strategy-method="canRelease" ? correlation-strategy="correlationStrategyBean" ? correlation-strategy-method="groupNumbersByLastDigit" ? message-store="messageStore" 11 send-partial-result-on-expiry="true" 12 send-timeout="86420000" 13 /> <channel id="outputChannel"/> <bean id="aggregatorBean" class="sample.PojoAggregator"/> <bean id="releaseStrategyBean" class="sample.PojoReleaseStrategy"/> <bean id="correlationStrategyBean" class="sample.PojoCorrelationStrategy"/>

? ? ? ? ? ? ?

The id of the aggregator is optional. The input channel of the aggregator. Required. The channel where the aggregator will send the aggregation results. Optional (because incoming messages can specify a reply channel themselves). The channel where the aggregator will send the messages that timed out (if send-partial-results-on-timeout is false). Optional. A reference to a bean defined in the application context. The bean must implement the aggregation logic as described above. Required. A method defined on the bean referenced by ref, that implements the message aggregation algorithm. Optional, with restrictions (see above). A reference to a bean that implements the decision algorithm as to whether a given message group is complete. The bean can be an implementation of the CompletionStrategy interface or a POJO. In the latter Spring-WS (2.0.0.M6) 44

Aggregator case the completion-strategy-method attribute must be defined as well. Optional (by default, the aggregator will use sequence size) . A method defined on the bean referenced by release-strategy, that implements the completion decision algorithm. Optional, with restrictions (requires completion-strategy to be present). A reference to a bean that implements the correlation strategy. The bean can be an implementation of the CorrelationStrategy interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined as well. Optional (by default, the aggregator will use the correlation id header attribute) . A method defined on the bean referenced by correlation-strategy, that implements the correlation key algorithm. Optional, with restrictions (requires correlation-strategy to be present). A reference to a MessageGroupStore that can be used to store groups of messages under their correlation key until they are complete. Optional with default a volatile in-memory store. Whether upon the expiration of the message group, the aggregator will try to aggregate the messages that have already arrived. Optional (false by default). The timeout for sending the aggregated messages to the output or reply channel. Optional.

? ?

?
11

12

13

Using a "ref" attribute is generally recommended if a custom aggregator handler implementation can be reused in other <aggregator> definitions. However if a custom aggregator handler implementation should be scoped to a concrete definition of the <aggregator>, you can use an inner bean definition (starting with version 1.0.3) for custom aggregator handlers within the <aggregator> element:
<aggregator input-channel="input" method="sum" output-channel="output"> <beans:bean class="org.foo.ExampleAggregator"/> </aggregator>

Note
Using both a "ref" attribute and an inner bean definition in the same <aggregator> configuration is not allowed, as it creates an ambiguous condition. In such cases, an Exception will be thrown. An example implementation of the aggregator bean looks as follows:
public class PojoAggregator { public Long add(List<Long> results) { long total = 0l; for (long partialResult: results) { total += partialResult; } return total; } }

An implementation of the completion strategy bean for the example above may be as follows:
public class PojoReleaseStrategy { ... public boolean canRelease(List<Long> numbers) { int sum = 0; for (long number: numbers) { sum += number; } return sum >= maxValue; } }

Note

Spring-WS (2.0.0.M6)

45

Aggregator

Wherever it makes sense, the release strategy method and the aggregator method can be combined in a single bean. An implementation of the correlation strategy bean for the example above may be as follows:
public class PojoCorrelationStrategy { ... public Long groupNumbersByLastDigit(Long number) { return number % 10; } }

For example, this aggregator would group numbers by some criterion (in our case the remainder after dividing by 10) and will hold the group until the sum of the numbers which represents the payload exceeds a certain value.

Note
Wherever it makes sense, the release strategy method, correlation strategy method and the aggregator method can be combined in a single bean (all of them or any two).

11.5. Managing State in an Aggregator: MessageGroupStore
Aggregator (and some other patterns in Spring Integration) is a stateful pattern that requires decisions to be made based on a group of messages that have arrived over a period of time, all with the same correlation key. The design of the interfaces in the stateful patterns (e.g. ReleaseStrategy) is driven by the principle that the components (framework and user) should be to remain stateless. All state is carried by the MessageGroup and its management is delegated to the MessageGroupStore. The MessageGroupStore accumulates state information in MessageGroups, potentially forever. So to prevent stale state from hanging around, and for volatile stores to provide a hook for cleaning up when the application shots down, the MessageGroupStore allows the user to register callbacks to apply to MessageGroups when they expire. The interface is very straighforward:
public interface MessageGroupCallback { void execute(MessageGroupStore messageGroupStore, MessageGroup group); }

The callback has access directly to the store and the message group so it can manage the persistent state (e.g. by removing the group from the store entirely). The MessageGroupStore maintains a list of these callbacks which it applies when asked to all messages whose timestamp is earlier than a time supplied as a parameter:
public interface MessageGroupStore { void registerMessageGroupExpiryCallback(MessageGroupCallback callback); int expireMessageGroups(long timeout); }

The expireMessageGroups method can be called with a timeout value: any message older than the current time Spring-WS (2.0.0.M6) 46

Aggregator

minus this value wiull be expired, and have the callbacks applied. Thus it is the user of the store that defines what is meant by message group "expiry". As a convenience for users, Spring Integration provides a wrapper for the message expiry in the form of a MessageGroupStoreReaper:
<bean id="reaper" class="org...MessageGroupStoreReaper"> <property name="messageGroupStore" ref="messageStore"/> <property name="timeout" value="10"/> </bean> <task:scheduled-tasks scheduler="scheduler"> <task:scheduled ref="reaper" method="run" fixed-rate="10000"/> </task:scheduled-tasks>

The reaper is a Runnable, and all that is happening is that the message group store's expire method is being called in the sample above once every 10 seconds. In addition to the reaper, the expiry callbacks are invoked when the application shuts down via a lifecycle callback in the CorrelatingMessageHandler. The CorrelatingMessageHandler registers its own expiry callback, and this is the link with the boolean flag send-partial-result-on-expiry in the XML configuration of the aggregator. If the flag is set to true, then when the expiry callback is invoked then any unmarked messages in groups that are not yet released can be sent on to the downstream channel.

11.6. Configuring an Aggregator with Annotations
An aggregator configured using annotations can look like this.
public class Waiter { ... @Aggregator ? public Delivery aggregatingMethod(List<OrderItem> items) { ... } @ReleaseStrategy ? public boolean releaseChecker(List<Message<?>> messages) { ... } @CorrelationStrategy ? public String correlateBy(OrderItem item) { ... } }

? ? ?

An annotation indicating that this method shall be used as an aggregator. Must be specified if this class will be used as an aggregator. An annotation indicating that this method shall be used as the release strategy of an aggregator. If not present on any method, the aggregator will use the SequenceSizeCompletionStrategy. An annotation indicating that this method shall be used as the correlation strategy of an aggregator. If no correlation strategy is indicated, the aggregator will use the HeaderAttributeCorrelationStrategy based on CORRELATION_ID.

All of the configuration options provided by the xml element are also available for the @Aggregator annotation. The aggregator can be either referenced explicitly from XML or, if the @MessageEndpoint is defined on the Spring-WS (2.0.0.M6) 47

Aggregator

class, detected automatically through classpath scanning.

Spring-WS (2.0.0.M6)

48

Chapter 12. Resequencer
12.1. Introduction
Related to the Aggregator, albeit different from a functional standpoint, is the Resequencer.

12.2. Functionality
The Resequencer works in a similar way to the Aggregator, in the sense that it uses the CORRELATION_ID to store messages in groups, the difference being that the Resequencer does not process the messages in any way. It simply releases them in the order of their SEQUENCE_NUMBER header values. With respect to that, the user might opt to release all messages at once (after the whole sequence, according to the SEQUENCE_SIZE, has been released), or as soon as a valid sequence is available.

12.3. Configuring a Resequencer with XML
Configuring a resequencer requires only including the appropriate element in XML. A sample resequencer configuration is shown below.
<channel id="inputChannel"/> <channel id="outputChannel"/> <resequencer id="completelyDefinedResequencer" ? input-channel="inputChannel" ? output-channel="outputChannel" ? discard-channel="discardChannel" ? release-partial-sequences="true" ? message-store="messageStore" ? send-partial-result-on-expiry="true" ? send-timeout="86420000" ? />

? ? ? ? ?

? ?

The id of the resequencer is optional. The input channel of the resequencer. Required. The channel where the resequencer will send the reordered messages. Optional. The channel where the resequencer will send the messages that timed out (if send-partial-result-on-timeout is false). Optional. Whether to send out ordered sequences as soon as they are available, or only after the whole message group arrives. Optional (false by default). If this flag is not specified (so a complete sequence is defined by the sequence headers) then it can make sense to provide a custom Comparator to be used to order the messages when sending (use the XML attribute comparator to point to a bean definition). If release-partial-sequences is true then there is no way with a custom comparator to define a partial sequence. To do that you would have to provide a release-strategy (also a reference to another bean definition, either a POJO or a ReleaseStrategy ). A reference to a MessageGroupStore that can be used to store groups of messages under their correlation key until they are complete. Optional with default a volatile in-memory store. Whether, upon the expiration of the group, the ordered group should be sent out (even if some of the messages are missing). Optional (false by default). See Section 11.5, “Managing State in an Aggregator: MessageGroupStore”. Spring-WS (2.0.0.M6) 49

Resequencer

?

The timeout for sending out messages. Optional.

Note
Since there is no custom behavior to be implemented in Java classes for resequencers, there is no annotation support for it.

Spring-WS (2.0.0.M6)

50

Chapter 13. Delayer
13.1. Introduction
A Delayer is a simple endpoint that allows a Message flow to be delayed by a certain interval. When a Message is delayed, the original sender will not block. Instead, the delayed Messages will be scheduled with an instance of java.util.concurrent.ScheduledExecutorService to be sent to the output channel after the delay has passed. This approach is scalable even for rather long delays, since it does not result in a large number of blocked sender Threads. On the contrary, in the typical case a thread pool will be used for the actual execution of releasing the Messages. Below you will find several examples of configuring a Delayer.

13.2. The <delayer> Element
The <delayer> element is used to delay the Message flow between two Message Channels. As with the other endpoints, you can provide the "input-channel" and "output-channel" attributes, but the delayer also requires at least the 'default-delay' attribute with the number of milliseconds that each Message should be delayed.
<delayer input-channel="input" default-delay="3000" output-channel="output"/>

If you need per-Message determination of the delay, then you can also provide the name of a header within the 'delay-header-name' attribute:
<delayer input-channel="input" output-channel="output" default-delay="3000" delay-header-name="delay"/>

In the example above the 3 second delay would only apply in the case that the header value is not present for a given inbound Message. If you only want to apply a delay to Messages that have an explicit header value, then you can set the 'default-delay' to 0. For any Message that has a delay of 0 (or less), the Message will be sent directly. In fact, if there is not a positive delay value for a Message, it will be sent to the output channel on the calling Thread.

Tip
The delay handler actually supports header values that represent an interval in milliseconds (any Object whose toString() method produces a value that can be parsed into a Long) as well as java.util.Date instances representing an absolute time. In the former case, the milliseconds will be counted from the current time (e.g. a value of 5000 would delay the Message for at least 5 seconds from the time it is received by the Delayer). In the latter case, with an actual Date instance, the Message will not be released until that Date occurs. In either case, a value that equates to a non-positive delay, or a Date in the past, will not result in any delay. Instead, it will be sent directly to the output channel in the original sender's Thread. The delayer delegates to an instance of Spring's TaskScheduler abstraction. The default scheduler is a ThreadPoolTaskScheduler instance with a pool size of 1. If you want to delegate to a different scheduler, you can provide a reference through the delayer element's 'scheduler' attribute:
<delayer input-channel="input" output-channel="output" default-delay="0" delay-header-name="delay" scheduler="exampleTaskScheduler"/> <task:scheduler id="exampleTaskScheduler" pool-size="3"/>

Spring-WS (2.0.0.M6)

51

Chapter 14. Message Handler Chain
14.1. Introduction
The MessageHandlerChain is an implementation of MessageHandler that can be configured as a single Message Endpoint while actually delegating to a chain of other handlers, such as Filters, Transformers, Splitters, and so on. This can lead to a much simpler configuration when several handlers need to be connected in a fixed, linear progression. For example, it is fairly common to provide a Transformer before other components. Similarly, when providing a Filter before some other component in a chain, you are essentially creating a Selective Consumer. In either case, the chain only requires a single input-channel and a single output-channel as opposed to the configuration of channels for each individual component.

Tip
Spring Integration's Filter provides a boolean property 'throwExceptionOnRejection'. When providing multiple Selective Consumers on the same point-to-point channel with different acceptance criteria, this value should be set to 'true' (the default is false) so that the dispatcher will know that the Message was rejected and as a result will attempt to pass the Message on to other subscribers. If the Exception were not thrown, then it would appear to the dispatcher as if the Message had been passed on successfully even though the Filter had dropped the Message to prevent further processing. The handler chain simplifies configuration while internally maintaining the same degree of loose coupling between components, and it is trivial to modify the configuration if at some point a non-linear arrangement is required. Internally, the chain will be expanded into a linear setup of the listed endpoints, separated by direct channels. The reply channel header will not be taken into account within the chain: only after the last handler is invoked will the resulting message be forwarded on to the reply channel or the chain's output channel. Because of this setup all handlers except the last require a setOutputChannel implementation. The last handler only needs an output channel if the outputChannel on the MessageHandlerChain is set.

Note
As with other endpoints, the output-channel is optional. If there is a reply Message at the end of the chain, the output-channel takes precedence, but if not available, the chain handler will check for a reply channel header on the inbound Message. In most cases there is no need to implement MessageHandlers yourself. The next section will focus on namespace support for the chain element. Most Spring Integration endpoints, like Service Activators and Transformers, are suitable for use within a MessageHandlerChain.

14.2. The <chain> Element
The <chain> element provides an 'input-channel' attribute, and if the last element in the chain is capable of producing reply messages (optional), it also supports an 'output-channel' attribute. The sub-elements are then filters, transformers, splitters, and service-activators. The last element may also be a router.
<chain input-channel="input" output-channel="output"> <filter ref="someSelector" throw-exception-on-rejection="true"/>

Spring-WS (2.0.0.M6)

52

Message Handler Chain

<header-enricher error-channel="customErrorChannel"> <header name="foo" value="bar"/> </header-enricher> <service-activator ref="someService" method="someMethod"/> </chain>

The <header-enricher> element used in the above example will set a message header with name "foo" and value "bar" on the message. A header enricher is a specialization of Transformer that touches only header values. You could obtain the same result by implementing a MessageHandler that did the header modifications and wiring that as a bean. Some time you need to make a nested call to another chain from within the chain and then come back and continue execution within the original chain. To accomplish this you can utilize Messaging Gateway by including light-configuration via <gateway> element. For example:
<si:chain id="main-chain" input-channel="inputA" output-channel="inputB"> <si:header-enricher> <si:header name="name" value="Many" /> </si:header-enricher> <si:service-activator> <bean class="org.foo.SampleService" /> </si:service-activator> <si:gateway request-channel="inputC"/> </si:chain> <si:chain id="nested-chain-a" input-channel="inputC"> <si:header-enricher> <si:header name="name" value="Moe" /> </si:header-enricher> <si:gateway request-channel="inputD"/> <si:service-activator> <bean class="org.foo.SampleService" /> </si:service-activator> </si:chain> <si:chain id="nested-chain-b" input-channel="inputD"> <si:header-enricher> <si:header name="name" value="Jack" /> </si:header-enricher> <si:service-activator> <bean class="org.foo.SampleService" /> </si:service-activator> </si:chain>

In the above example the nested-chain-a will be called at the end of main-chain processing by the 'gateway' element configured there. While in nested-chain-a a call to a nested-chain-b will be made after header enrichment and then it will come back to finish execution in nested-chain-b finally getting back to the main-chain. When light version of <gateway> element is defined in the chain SI will construct an instance SimpleMessagingGateway (no need to provide 'service-interface' configuration) which will take the message in its current state and will place it on the channel defined via 'request-channel' attribute. Upon processing Message will be returned to the gateway and continue its journey within the current chain.

Spring-WS (2.0.0.M6)

53

Chapter 15. Messaging Bridge
15.1. Introduction
A Messaging Bridge is a relatively trivial endpoint that simply connects two Message Channels or Channel Adapters. For example, you may want to connect a PollableChannel to a SubscribableChannel so that the subscribing endpoints do not have to worry about any polling configuration. Instead, the Messaging Bridge provides the polling configuration. By providing an intermediary poller between two channels, a Messaging Bridge can be used to throttle inbound Messages. The poller's trigger will determine the rate at which messages arrive on the second channel, and the poller's "maxMessagesPerPoll" property will enforce a limit on the throughput. Another valid use for a Messaging Bridge is to connect two different systems. In such a scenario, Spring Integration's role would be limited to making the connection between these systems and managing a poller if necessary. It is probably more common to have at least a Transformer between the two systems to translate between their formats, and in that case, the channels would be provided as the 'input-channel' and 'output-channel' of a Transformer endpoint. If data format translation is not required, the Messaging Bridge may indeed be sufficient.

15.2. The <bridge> Element
The <bridge> element is used to create a Messaging Bridge between two Message Channels or Channel Adapters. Simply provide the "input-channel" and "output-channel" attributes:
<bridge input-channel="input" output-channel="output"/>

As mentioned above, a common use case for the Messaging Bridge is to connect a PollableChannel to a SubscribableChannel, and when performing this role, the Messaging Bridge may also serve as a throttler:
<bridge input-channel="pollable" output-channel="subscribable"> <poller max-messages-per-poll="10"> <interval-trigger interval="5" time-unit="SECONDS"/> </poller> </bridge>

Connecting Channel Adapters is just as easy. Here is a simple echo example between the "stdin" and "stdout" adapters from Spring Integration's "stream" namespace.
<stream:stdin-channel-adapter id="stdin"/> <stream:stdout-channel-adapter id="stdout"/> <bridge id="echo" input-channel="stdin" output-channel="stdout"/>

Of course, the configuration would be similar for other (potentially more useful) Channel Adapter bridges, such as File to JMS, or Mail to File. The various Channel Adapters will be discussed in upcoming chapters.

Note
If no 'output-channel' is defined on a bridge, the reply channel provided by the inbound Message will be used, if available. If neither output or reply channel is available, an Exception will be thrown.

Spring-WS (2.0.0.M6)

54

Chapter 16. Inbound Messaging Gateways
16.1. SimpleMessagingGateway
Even though the MessageChannelTemplate is fairly straightforward, it does not hide the details of messaging from your application code. To support working with plain Objects instead of messages, Spring Integration provides SimpleMessagingGateway with the following methods:
public void send(Object object) { ... } public Object receive() { ... } public Object sendAndReceive(Object object) { ... } Message<?> sendAndReceiveMessage(Object object);

It enables configuration of a request and/or reply channel and delegates to instances of the InboundMessageMapper and OutboundMessageMapper strategy interfaces.
SimpleMessagingGateway gateway = new SimpleMessagingGateway(inboundMapper, outboundMapper); gateway.setRequestChannel(requestChannel); gateway.setReplyChannel(replyChannel); Object result = gateway.sendAndReceive("test");

16.2. GatewayProxyFactoryBean
Working with Objects instead of Messages is an improvement. However, it would be even better to have no dependency on the Spring Integration API at all - including the gateway class. For that reason, Spring Integration also provides a GatewayProxyFactoryBean that generates a proxy for any interface and internally invokes the gateway methods shown above. Namespace support is also provided as demonstrated by the following example.
<gateway id="fooService" service-interface="org.example.FooService" default-request-channel="requestChannel" default-reply-channel="replyChannel"/>

Then, the "fooService" can be injected into other beans, and the code that invokes the methods on that proxied instance of the FooService interface has no awareness of the Spring Integration API. The general approach is similar to that of Spring Remoting (RMI, HttpInvoker, etc.). See the "Samples" Appendix for an example that uses this "gateway" element (in the Cafe demo). The reason that the attributes on the 'gateway' element are named 'default-request-channel' and 'default-reply-channel' is that you may also provide per-method channel references by using the @Gateway annotation.
public interface Cafe { @Gateway(requestChannel="orders") void placeOrder(Order order); }

It is also possible to pass values to be interpreted as Message headers on the Message that is created and sent to the request channel by using the @Header annotation:

Spring-WS (2.0.0.M6)

55

Inbound Messaging Gateways

public interface FileWriter { @Gateway(requestChannel="filesOut") void write(byte[] content, @Header(FileHeaders.FILENAME) String filename); }

If you prefer XML way of configuring Gateway methods, you can provide method sub-elements to the gateway configuration (see below)
<si:gateway id="myGateway" service-interface="org.foo.bar.TestGateway" default-request-channel="inputC"> <si:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/> <si:method name="echoUpperCase" request-channel="inputB"/> <si:method name="echoViaDefault"/> </si:gateway>

You can also provide individual headers per method invocation via XML. This could be very useful if headers you want to set are static in nature and you don't want to embed them in the gateway's method signature via @Header annotation. For example; in the Loan Broker example we want to influence how aggregation of the Loan quotes will be done based on what type of request was initiated (single quote or all quotes). Determining the type of the request by evaluating what gateway's method was invoked, although possible would violate the separation of concerns paradigm (method is a java artifact), but expressing your intention (meta information) via Message headers is natural to Messaging architecture.
<int:gateway id="loanBrokerGateway" service-interface="org.springframework.integration.loanbroker.LoanBrokerGateway"> <int:method name="getLoanQuote" request-channel="loanBrokerPreProcessingChannel"> <int:header name="RESPONSE_TYPE" value="BEST"/> </int:method> <int:method name="getAllLoanQuotes" request-channel="loanBrokerPreProcessingChannel"> <int:header name="RESPONSE_TYPE" value="ALL"/> </int:method> </int:gateway>

In the above case you can clearly see how a different header value will be set for the 'RESPONSE_TYPE' header based on the gateway's method. As with anything else, Gateway invocation might result in error. By default all error that have occurred downstream will be re-thrown as MessagingExeption (RuntimeException) upon Gateway's method invocation. However there are times when you may want to treat Exception as a valid reply, by mapping it to a Message. To accomplish this Gateway provides support for Exception mappers via exception-mapper attribute.
<si:gateway id="sampleGateway" default-request-channel="gatewayChannel" service-interface="foo.bar.SimpleGateway" exception-mapper="exceptionMapper"/> <bean id="exceptionMapper" class="foo.bar.SampleExceptionMapper"/>

foo.bar.SampleExceptionMapper is the org.springframework.integration.message.InboundMessageMapper toMessage(Object object).

which

implementation only defines one

of method

public static class SampleExceptionMapper implements InboundMessageMapper<Throwable>{ public Message<?> toMessage(Throwable object) throws Exception { MessageHandlingException ex = (MessageHandlingException) object; return MessageBuilder.withPayload("Error happened in message: " + ex.getFailedMessage().getPayload()).build(); } }

Spring-WS (2.0.0.M6)

56

Inbound Messaging Gateways

Spring-WS (2.0.0.M6)

57

Chapter 17. Message Publishing
The AOP Message Publishing feature allows you to construct and send a message as a by-product of method invocation. For example, imagine you have a component and every time the state of this component changes you would like to be notified via a Message. The easiest way to send such notifications would be to send a message to a dedicated channel, but how would you connect the method invocation that changes the state of the object to a message sending process, and how should the notification Message be structured? The AOP Message Publishing feature handles these responsibilities with a configuration-driven approach.

17.1. Message Publishing Configuration
Spring Integration provides two approaches: XML and Annotation-driven.

17.1.1. Annotation-driven approach via @Publisher annotation
The annotation-driven approach allows you to annotate any method with the @Publisher annotation and provide configuration attributes which will dictate the structure of a Message. The invocation of the annotated method will be proxied through PublisherAnnotationAdvisor which will construct a Message and send it to a Message Channel. Internally PublisherAnnotationAdvisor uses the Spring 3.0 Expression Language support, giving you considerable flexibility and control over the structure of the Message it will build. Here's an example:
@Publisher(payload="#return", channel="testChannel", headers="bar='123',fname=#args.fname") public String setName(String fname, String lname){ return fname + " " + lname; }

PublisherAnnotationAdvisor

defines and binds the following variables:

? #return - will bind to a return value allowing you to reference it or its attributes (e.g., #return.foo where 'foo' is an attribute of the object bound to #return) ? #exception - will bind to an exception if one is thrown by the method invocation. ? #args - will bind to method arguments, so individual arguments could be extracted by name (e.g., #args.fname as in the above method) In the above example the Message will be constructed with the following structure: ? Message payload - will be of type String and contain the value returned by the method. ? Message headers will contain 'bar' with a value of "123" and 'fname' with the value of the 'fname' argument passed to the method at runtime. As with most other annotation-driven features in Spring, you will need to register a post-processor (PublisherAnnotationBeanPostProcessor).
<bean class="org.springframework.integration.aop.PublisherAnnotationBeanPostProcessor"/>

Spring-WS (2.0.0.M6)

58

Message Publishing

17.1.2. XML-based approach via <publisher> element
The XML-based approach allows you to configure the same AOP-based Message Publishing functionality with simple namespace-based configuration of a MessagePublishingInterceptor. It certainly has some benefits over the annotation-driven approach since it allows you to use AOP pointcut expressions, thus possibly intercepting multiple methods at once or intercepting and publishing methods to which you don't have the source code. To configure Message Publishing via XML, you only need to do the following two things: ? Provide configuration for MessagePublishingInterceptor via the <publisher> XML element. ? Provide AOP configuration to apply the MessagePublishingInterceptor to managed objects.

<beans:bean id="testBean" class="org.foo.bar.TestBean" /> <aop:config> <aop:advisor advice-ref="interceptor" pointcut="bean(testBean)" /> </aop:config> <publisher id="interceptor" default-channel="defaultChannel"> <method pattern="echo" payload="'Echoing: ' + #return" headers="foo='bar'" channel="echoChannel"/> <method pattern="echoDef*" payload="#return"/> <method pattern="foo*"/> </publisher>

As you can see the <publisher> element expects the same variables as the PublisherAnnotationAdvisor and also utilizes the power of the Spring 3.0 Expression Language. In the above example the execution of the echo method of a testBean will render a Message with the following structure: ? The Message payload will be of type String and value of "Echoing: [value]" where value is the value returned by an executed method. ? The Message headers will contain the key "foo" with a value of "bar". ? The Message will be sent to echoChannel. The second method, mapping the execution of any method that begins with echoDef of testBean, will produce a Message with the following structure. ? The Message payload will be the value returned by an executed method. ? Since the channel attribute is not provided explicitly, the Message will be sent to the defaultChannel defined by the publisher. The third mapping is almost identical to the previous one (with the exception of method pattern), since the return value will be mapped to the Message payload by default if nothing else is specified. For simple mapping rules you can rely on the publisher defaults. For example:
<publisher id="anotherInterceptor"/>

Spring-WS (2.0.0.M6)

59

Message Publishing This will map the return value of every method that matches the pointcut expression to a payload and will be sent to a default-channel. If the defaultChannelis not specified (as above) the messages will be sent to the global nullChannel.

Spring-WS (2.0.0.M6)

60

Chapter 18. File Support
18.1. Introduction
Spring Integration's File support extends the Spring Integration Core with a dedicated vocabulary to deal with reading, writing, and transforming files. It provides a namespace that enables elements defining Channel Adapters dedicated to files and support for Transformers that can read file contents into strings or byte arrays. This section will explain the workings of FileReadingMessageSource and FileWritingMessageHandler and how to configure them as beans. Also the support for dealing with files through file specific implementations of Transformer will be discussed. Finally the file specific namespace will be explained.

18.2. Reading Files
A FileReadingMessageSource can be used to consume files from the filesystem. This is an implementation of MessageSource that creates messages from a file system directory.
<bean id="pollableFileSource" class="org.springframework.integration.file.FileReadingMessageSource" p:inputDirectory="file:${input.directory}"/>

To prevent creating messages for certain files, you may supply a FileListFilter. By default, an AcceptOnceFileListFilter is used. This filter ensures files are picked up only once from the directory.
<bean id="pollableFileSource" class="org.springframework.integration.file.FileReadingMessageSource" p:inputDirectory="file:${input.directory}" p:filter-ref="customFilterBean"/>

A common problem with reading files is that a file may be detected before it is ready. The default AcceptOnceFileListFilter does not prevent this. In most cases, this can be prevented if the file-writing process renames each file as soon as it is ready for reading. A pattern-matching filter that accepts only files that are ready (e.g. based on a known suffix), composed with the default AcceptOnceFileListFilter allows for this. The CompositeFileListFilter enables the composition.
<bean id="pollableFileSource" class="org.springframework.integration.file.FileReadingMessageSource" p:inputDirectory="file:${input.directory}" p:filter-ref="compositeFilter"/> <bean id="compositeFilter" class="org.springframework.integration.file.CompositeFileListFilter"> <constructor-arg> <list> <bean class="org.springframework.integration.file.AcceptOnceFileListFilter" /> <bean class="org.springframework.integration.file.PatternMatchingFileListFilter"> <constructor-arg value="^test.*$"/> </bean> </list> </constructor-arg> </bean>

The configuration can be simplified using the file specific namespace. To do this use the following template.
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:integration="http://www.springframework.org/schema/integration" xmlns:file="http://www.springframework.org/schema/integration/file"

Spring-WS (2.0.0.M6)

61

File Support

xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file-1.0.xsd"> </beans>

Within this namespace you can reduce the FileReadingMessageSource and wrap it in an inbound Channel Adapter like this:
<file:inbound-channel-adapter id="filesIn" directory="file:${input.directory}" prevent-duplicates="true"/> <file:inbound-channel-adapter id="filesIn" directory="file:${input.directory}" filter="customFilterBean" /> <file:inbound-channel-adapter id="filesIn" directory="file:${input.directory}" filename-pattern="^test.*$" />

The first channel adapter is relying on the default filter that just prevents duplication, the second is using a custom filter, and the third is using the filename-pattern attribute to add a Pattern based filter to the FileReadingMessageSource. The file-name-pattern and filter attributes are mutually exclusive, but you can use a CompositeFileListFilter to use any combination of filters, including a pattern based filter to fit your particular needs. When multiple processes are reading from the same directory it can be desirable to lock files to prevent them from being picked up concurrently. To do this you can use a FileLocker. There is a java.nio based implementation available out of the box, but it is also possible to implement your own locking scheme. The nio locker can be injected as follows
<file:inbound-channel-adapter id="filesIn" directory="file:${input.directory}" prevent-duplicates="true"> <file:nio-locker/> </file:inbound-channel-adapter>

A custom locker you can configure like this:
<file:inbound-channel-adapter id="filesIn" directory="file:${input.directory}" prevent-duplicates="true"> <file:locker ref="customLocker"/> </file:inbound-channel-adapter>

When filtering and locking files is not enough it might be needed to control the way files are listed entirely. To implement this type of requirement you can use an implementation of DirectoryScanner. This scanner allows you to determine entirely what files are listed each poll. This is also the interface that Spring Integration uses internally to wire FileListFilters FileLocker to the FileReadingMessageSource. A custom DirectoryScanner can be injected into the <file:inbound-channel-adapter/> on the scanner attribute.
<file:inbound-channel-adapter id="filesIn" directory="file:${input.directory}" prevent-duplicates="true" scanner="customDirectoryScanner"/>

This gives you full freedom to choose the ordering, listing and locking strategies.

18.3. Writing files

Spring-WS (2.0.0.M6)

62

File Support

To write messages to the file system you can use a FileWritingMessageHandler. This class can deal with File, String, or byte array payloads. In its simplest form the FileWritingMessageHandler only requires a destination directory for writing the files. The name of the file to be written is determined by the handler's FileNameGenerator. The default implementation looks for a Message header whose key matches the constant defined as FileHeaders.FILENAME. Additionally, you can configure the encoding and the charset that will be used in case of a String payload. To make things easier you can configure the FileWritingMessageHandler as part of an outbound channel adapter using the namespace.
<file:outbound-channel-adapter id="filesOut" directory="file:${input.directory.property}"/>

The namespace based configuration also supports a delete-source-files attribute. If set to true, it will trigger deletion of the original source files after writing to a destination. The default value for that flag is false.
<file:outbound-channel-adapter id="filesOut" directory="file:${output.directory}" delete-source-files="true"/>

Note
The delete-source-files attribute will only have an effect if the inbound Message has a File payload or if the FileHeaders.ORIGINAL_FILE header value contains either the source File instance or a String representing the original file path. In cases where you want to continue processing messages based on the written File you can use the outbound-gateway instead. It plays a very similar role as the outbound-channel-adapter. However after writing the File, it will also send it to the reply channel as the payload of a Message.
<file:outbound-gateway id="mover" request-channel="moveInput" reply-channel="output" directory="${output.directory}" delete-source-files="true"/>

Note
The 'outbound-gateway' works well in cases where you want to first move a File and then send it through a processing pipeline. In such cases, you may connect the file namespace's 'inbound-channel-adapter' element to the 'outbound-gateway' and then connect that gateway's reply-channel to the beginning of the pipeline. If you have more elaborate requirements or need to support additional payload types as input to be converted to file content you could extend the FileWritingMessageHandler, but a much better option is to rely on a Transformer.

18.4. File Transformers
To transform data read from the file system to objects and the other way around you need to do some work. Contrary to FileReadingMessageSource and to a lesser extent FileWritingMessageHandler, it is very likely that you will need your own mechanism to get the job done. For this you can implement the Transformer interface. Or extend the AbstractFilePayloadTransformer for inbound messages. Some obvious Spring-WS (2.0.0.M6) 63

File Support implementations have been provided. transforms Files into byte[]s using Spring's FileCopyUtils. It is often better to use a sequence of transformers than to put all transformations in a single class. In that case the File to byte[] conversion might be a logical first step.
FileToByteArrayTransformer

will convert Files to Strings as the name suggests. If nothing else, this can be useful for debugging (consider using with a Wire Tap).
FileToStringTransformer

To configure File specific transformers you can use the appropriate elements from the file namespace.
<file-to-bytes-transformer input-channel="input" output-channel="output" delete-files="true"/> <file:file-to-string-transformer input-channel="input" output-channel="output delete-files="true" charset="UTF-8"/>

The delete-files option signals to the transformer that it should delete the inbound File after the transformation is complete. This is in no way a replacement for using the AcceptOnceFileListFilter when the FileReadingMessageSource is being used in a multi-threaded environment (e.g. Spring Integration in general).

Spring-WS (2.0.0.M6)

64

Chapter 19. JDBC Support
Spring Integration provides Channel Adapters for receiving and sending messages via database queries.

19.1. Inbound Channel Adapter
The main function of an inbound Channel Adapter is to execute a SQL SELECT query and turn the result set into a message. The message payload is the whole result set, expressed as a List, and the types of the items in the list depends on the row-mapping strategy that is used. The default strategy is a generic mapper that just returns a Map for each row i nthe query. Optionally this can be changed by adding a reference to requires a reference to a RowMapper instance (see the Spring JDBC documentation for more detailed information about row mapping).

Note
If you want to convert rows in the SELECT query result to individual messages you can use a downstream splitter. The inbound adapter also requires a reference to either JdbcTemplate instance or DataSource. The following example defines an inbound Channel Adapter with a DataSource reference.
<jdbc:inbound-channel-adapter query="select * from item where status=2" channel="target" data-source="dataSource" update="update item set status=10 where id in (:idList)" />

Note
The parameters in the update query are specified with a colon (:) prefix to the name of a map key. This is a standard feature of the named parameter JDBC support in Spring JDBC. As well as the SELECT statement to generate the messages, the adapter above also has an UPDATE statement that is being used to mark the records as processed, so they don't show up in the next poll. The update is parameterised by the list of ids from the original select. This is done through a naming convention by default (a column in the input result set called "id" is translated into a list in the parameter map for the update called "idList"). To change the parameter generation strategy you can inject a SqlParameterSourceFactory into the adapter to override the default behaviour (the adapter has a sql-parameter-source-factory attribute).

19.1.1. Polling and Transactions
The inbound adapter accepts a regular Spring Integration poller as a sub element, so for instance the frequency of the polling can be controlled. A very important feature of the poller for JDBC usage is the option to wrap the poll operation in a transaction, for example:
<jdbc:inbound-channel-adapter query="..." channel="target" data-source="dataSource" update="..."> <poller> <interval-trigger interval="1000"/> <transactional/> </poller> </jdbc:inbound-channel-adapter>

Spring-WS (2.0.0.M6)

65

JDBC Support

Note
If a poller is not explicitly specified a default value will be used (and as per normal with Spring Integration can be defined as a top level bean) In this example the database is polled every 1000 milliseconds, and the update and select queries are both executed in the same transaction. The transaction manager configuration is not shown, but as long as it is aware of the data source then the poll is transactional. A common use case is for the downstream channels to be direct channels (the default), so that the endpoints are invoked in the same thread, and hence the same transaction. then if any of them fails, the transaction rolls back and the input data are reverted to their original state.

19.2. Outbound Channel Adapter
The outbound Channel Adapter is the inverse of the inbound: its role is to handle a message and use it to execute a SQL query. The message payload and headers are available by default as input parameters to the query, for instance:
<jdbc:outbound-channel-adapter query="insert into foos (id, status, name) values (:headers[$id], 0, :payload[foo])" channel="input" data-source="dataSource"/>

In the example above, messages arriving on the channel "input" have a payload of a map with key "foo", so the [] operator dereferences that value from the map. The headers are also accessed as a map.

Note
The parameters in the query above are bean paths in the incoming message (they are not Spring EL expressions). This behaviour is part of the MapSqlParameterSource in Spring JDBC, which is the default source created by the outbound adapter. Other behaviour is possible in the adapter, and only requires the user to inject a different SqlParameterSourceFactory . The outbound adapter requires a reference to either a DataSource or a JdbcTemplate. It can also have a SqlParameterSourceFactory injected to control the binding of incoming message to the query. If the input channel is a direct channel then the outbound adapter runs its query in the same thread, and therefor ethe same transaction (if there is one) as the sender of the message.

19.3. Message Store
The JDBC module provides an implementation of the Spring Integration MessageStore (important in the Claim Check pattern) and MessageGroupStore (important in stateful patterns like Aggregator) backed by a database. Both interfaces are implemented by the JdbcMessageStore and there is also support for configuring store instances in XML. For example:
<jdbc:message-store id="messageStore" data-source="dataSource"/>

A JdbcTemplate can be specified instead of a DataSource. Other optional attributes are show in the next example:
<jdbc:message-store id="messageStore" data-source="dataSource" lob-handler="lobHandler" table-prefix="MY_INT_"/>

Spring-WS (2.0.0.M6)

66

JDBC Support Here we have specified a LobHandler for dealing with messages as large objects (e.g. often necessary if using Oracle) and a prefix for the table names in the queries generated by the store. The table name prefix defaults to "INT_".

19.3.1. Initializing the Database
Spring Integration ships with some sample scripts that can be used to initialize a database. In the spring-integration-jdbc JAR file you will find scripts in the org.springframework.integration.jdbc package: there is a create and a drop script example for a range of common database platforms. A common way to use these scripts is to reference them in a Spring JDBC data source initializer. Note that the scripts are provided as samples or specifications of the the required table and column names. You may find that you need to enhance them for production use (e.g. with index declarations).

19.3.2. Partitioning a Message Store
It is common to use a JdbcMessageStore as a global store for a group of applications, or nodes in the same application. To provide some portection against name clashes, and to give control over the database meta-data configuration, the message store allows the tables to be partitioned in two ways. One is to use separate table names, by changing the prefix as described above, and the other is to specify a "region" name for partitioning data within a single table. An important use case for this is using the store to manage persistent queues backing a Spring Integration channel. The message data for a persistent channel is keyed in the store on the channel name, so if the channel names are not globally unique then there is the danger of channels picking up data that was not intended for them. To avoid this the message store region can be used to keep data separate for different physical channels that happen to have the same logical name.

Spring-WS (2.0.0.M6)

67

Chapter 20. JMS Support
Spring Integration provides Channel Adapters for receiving and sending JMS messages. There are actually two JMS-based inbound Channel Adapters. The first uses Spring's JmsTemplate to receive based on a polling period. The second is "message-driven" and relies upon a Spring MessageListener container. There is also an outbound Channel Adapter which uses the JmsTemplate to convert and send a JMS Message on demand. Whereas the JMS Channel Adapters are intended for unidirectional Messaging (send-only or receive-only), Spring Integration also provides inbound and outbound JMS Gateways for request/reply operations. The inbound gateway relies on one of Spring's MessageListener container implementations for Message-driven reception that is also capable of sending a return value to the "reply-to" Destination as provided by the received Message. The outbound Gateway sends a JMS Message to a "request-destination" and then receives a reply Message. The "reply-destination" reference (or "reply-destination-name") can be configured explicitly or else the outbound gateway will use a JMS TemporaryQueue.

20.1. Inbound Channel Adapter
The inbound Channel Adapter requires a reference to either a single JmsTemplate instance or both ConnectionFactory and Destination (a 'destinationName' can be provided in place of the 'destination' reference). The following example defines an inbound Channel Adapter with a Destination reference.
<jms:inbound-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel"> <integration:poller> <integration:interval-trigger interval="30" time-unit="SECONDS"/> </integration:poller> </jms:inbound-channel-adapter>

Tip
Notice from the configuration that the inbound-channel-adapter is a Polling Consumer. That means that it invokes receive() when triggered. This should only be used in situations where polling is done relatively infrequently and timeliness is not important. For all other situations (a vast majority of JMS-based use-cases), the message-driven-channel-adapter described below is a better option.

Note
All of the JMS adapters that require a reference to the ConnectionFactory will automatically look for a bean named "connectionFactory" by default. That is why you don't see a "connection-factory" attribute in many of the examples. However, if your JMS ConnectionFactory has a different bean name, then you will need to provide that attribute. If 'extract-payload' is set to true (which is the default), the received JMS Message will be passed through the MessageConverter. When relying on the default SimpleMessageConverter, this means that the resulting Spring Integration Message will have the JMS Message's body as its payload. A JMS TextMessage will produce a String-based payload, a JMS BytesMessage will produce a byte array payload, and a JMS ObjectMessage's Serializable instance will become the Spring Integration Message's payload. If instead you prefer to have the raw JMS Message as the Spring Integration Message's payload, then set 'extract-payload' to false.
<jms:inbound-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel" extract-payload="false"/> <integration:poller> <integration:interval-trigger interval="30" time-unit="SECONDS"/>

Spring-WS (2.0.0.M6)

68

JMS Support

</integration:poller> </jms:inbound-channel-adapter>

20.2. Message-Driven Channel Adapter
The "message-driven-channel-adapter" requires a reference to either an instance of a Spring MessageListener container (any subclass of AbstractMessageListenerContainer) or both ConnectionFactory and Destination (a 'destinationName' can be provided in place of the 'destination' reference). The following example defines a message-driven Channel Adapter with a Destination reference.
<jms:message-driven-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel"/>

Note
The Message-Driven adapter also accepts several properties that pertain to the MessageListener container. These values are only considered if you do not provide an actual 'container' reference. In that case, an instance of DefaultMessageListenerContainer will be created and configured based on these properties. For example, you can specify the "transaction-manager" reference, the "concurrent-consumers" value, and several other property references and values. Refer to the JavaDoc and Spring Integration's JMS Schema (spring-integration-jms-1.0.xsd) for more detail. The 'extract-payload' property has the same effect as described above, and once again its default value is 'true'. The poller sub-element is not applicable for a message-driven Channel Adapter, as it will be actively invoked. For most usage scenarios, the message-driven approach is better since the Messages will be passed along to the MessageChannel as soon as they are received from the underlying JMS co

推荐相关:

SpringCoud第二天

? ? 开源项目:http://git.oschina.net/zhou666/spring-cloud-7simple cloud-config-repo:配置文件存放的文件夹 cloud-simple-service:一个使用 mybatis 的数据库...


Spring_Hibernate_Tomcat_SVN基础知识梳理

Spring_Hibernate_Tomcat_SVN基础知识梳理_计算机软件及应用_IT/计算机_专业资料。本文主要介绍:Spring、Hibernate、Tomcat、SVN的基础知识...


Spring Festival(英文版春节介绍)

Spring Festival(英文版春节介绍)_英语学习_外语学习_牛股专区。Spring Festival 1. Tale of Nian We call the Spring Festival Nian, but did you know that ...


It’s spring教学设计

a. Show the pictures of rabbit and grass.The rabbit says “The spring is green.Look,the grass is green. b. Learn to say the new word and patten...


spring基本知识点

spring基本知识点_计算机软件及应用_IT/计算机_专业资料。Spring 基本知识点一、 编写一个 Spring 程序 1. 为什么要使用 Spring Spring 是一种轻量级框架,轻量级是...


Spring技术内幕

(this.proxyClassLoader); 上面我们看到了在 Spring 中通过 ProxyFactoryBean 实现 AOP 功能的第一步, 得到 AopProxy 代理对象的基本过程,下面我 们看看 AopProxy...


Spring技术介绍

Spring技术介绍_IT/计算机_专业资料。一、基本概念 Spring 的核心是轻量级(Lightweight)的容器(Container) ,它实现了 IoC 容器、非侵 入性(No intrusive)的框架,...


spring详细说明

spring详细说明_计算机软件及应用_IT/计算机_专业资料。对于Spring的详细解说,例如AOP,IOC等! 适合初学者Spring 课程设计 一、概述 Spring 是轻量级的 J2EE 应用...


尚学堂Spring笔记

一、锲子 二、环境 三、环境搭建 四、关键技术 Ioc 控制反转 AOP(面向切面编程) spring 对 AOP 的支持(采用 Annotation 的方式) spring 对 AOP 的支持(采用...


spring_security3.2 详细配置 培训笔记

Spring Security 3.x 出来一段时间了,跟 Acegi 是大丌同了,不 2.x 的版本也有一些小小的区别,网上有 一些文档,也有人翻译 Spring Security 3.x 的 guide,...

加微信领牛股 | 网站地图
All rights reserved Powered by 学霸学习网 www.tceic.com
copyright ©right 2010-2021。
文档资料库内容来自网络,如有侵犯请联系客服。zhit325@126.com