, is also available — for example, local-filename-generator-expression="#remoteFileName.toUpperCase() + headers.something"
.
The get
and mget
commands support the local-directory-expression
attribute.
It defines a SpEL expression to generate the name of local directories during the transfer.
The root object of the evaluation context is the request message but.
The remoteDirectory
variable, which is particularly useful for mget
, is also available — for example: local-directory-expression="'/tmp/local/' + #remoteDirectory.toUpperCase() + headers.something"
.
This attribute is mutually exclusive with the local-directory
attribute.
For all commands, the 'expression' property of the gateway provides the path on which the command acts.
For the mget
command, the expression might evaluate to '', meaning to retrieve all files, or 'somedirectory/', and so on.
The following example shows a gateway configured for an ls
command:
<int-ftp:outbound-gateway id="gateway1"
session-factory="ftpSessionFactory"
request-channel="inbound1"
command="ls"
command-options="-1"
expression="payload"
reply-channel="toSplitter"/>
The payload of the message sent to the toSplitter
channel is a list of String
objects that each contain the name of a file.
If the command-options
attribute was omitted, it holds FileInfo
objects.
It uses space-delimited options — for example, command-options="-1 -dirs -links"
.
Starting with version 4.2, the GET
, MGET
, PUT
and MPUT
commands support a FileExistsMode
property (mode
when using the namespace support).
This affects the behavior when the local file exists (GET
and MGET
) or the remote file exists (PUT
and MPUT
).
Supported modes are REPLACE
, APPEND
, FAIL
, and IGNORE
.
For backwards compatibility, the default mode for PUT
and MPUT
operations is REPLACE
.
For GET
and MGET
operations, the default is FAIL
.
Starting with version 5.0, the setWorkingDirExpression()
(working-dir-expression
in XML) option is provided on the FtpOutboundGateway
(<int-ftp:outbound-gateway>
in XML).
It lets you change the client working directory at runtime.
The expression is evaluated against the request message.
The previous working directory is restored after each gateway operation.
18.9.9. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the outbound gateway with Java configuration:
@SpringBootApplication
public class FtpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public SessionFactory<FTPFile> ftpSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost("localhost");
sf.setPort(port);
sf.setUsername("foo");
sf.setPassword("foo");
sf.setTestSession(true);
return new CachingSessionFactory<FTPFile>(sf);
}
@Bean
@ServiceActivator(inputChannel = "ftpChannel")
public MessageHandler handler() {
FtpOutboundGateway ftpOutboundGateway =
new FtpOutboundGateway(ftpSessionFactory(), "ls", "'my_remote_dir/'");
ftpOutboundGateway.setOutputChannelName("lsReplyChannel");
return ftpOutboundGateway;
}
}
18.9.10. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound gateway with the Java DSL:
@SpringBootApplication
public class FtpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public SessionFactory<FTPFile> ftpSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost("localhost");
sf.setPort(port);
sf.setUsername("foo");
sf.setPassword("foo");
sf.setTestSession(true);
return new CachingSessionFactory<FTPFile>(sf);
}
@Bean
public FtpOutboundGatewaySpec ftpOutboundGateway() {
return Ftp.outboundGateway(ftpSessionFactory(),
AbstractRemoteFileOutboundGateway.Command.MGET, "payload")
.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
.regexFileNameFilter("(subFtpSource|.*1.txt)")
.localDirectoryExpression("'localDirectory/' + #remoteDirectory")
.localFilenameExpression("#remoteFileName.replaceFirst('ftpSource', 'localTarget')");
}
@Bean
public IntegrationFlow ftpMGetFlow(AbstractRemoteFileOutboundGateway<FTPFile> ftpOutboundGateway) {
return f -> f
.handle(ftpOutboundGateway)
.channel(c -> c.queue("remoteFileOutputChannel"));
}
}
18.9.11. Outbound Gateway Partial Success (mget
and mput
)
When you perform operations on multiple files (by using mget
and mput
), an exception can occur some time after one or more files have been transferred.
In this case (starting with version 4.2), a PartialSuccessException
is thrown.
As well as the usual MessagingException
properties (failedMessage
and cause
), this exception has two additional
properties:
-
partialResults
: The successful transfer results. -
derivedInput
: The list of files generated from the request message (for example, local files to transfer for anmput
).
These attributes let you determine which files were successfully transferred and which were not.
In the case of a recursive mput
, the PartialSuccessException
may have nested PartialSuccessException
occurrences.
Consider the following directory structure:
root/
|- file1.txt
|- subdir/
| - file2.txt
| - file3.txt
|- zoo.txt
If the exception occurs on file3.txt
, the PartialSuccessException
thrown by the gateway has derivedInput
of file1.txt
, subdir
, and zoo.txt
and partialResults
of file1.txt
.
Its cause
is another PartialSuccessException
with derivedInput
of file2.txt
and file3.txt
and
partialResults
of file2.txt
.
18.10. FTP Session Caching
Starting with Spring Integration 3.0, sessions are no longer cached by default.
The cache-sessions attribute is no longer supported on endpoints.
You must use a CachingSessionFactory (shown in the next example) if you wish to cache sessions.
|
In versions prior to 3.0, the sessions were automatically cached by default.
A cache-sessions
attribute was available for disabling the auto caching, but that solution did not provide a way to configure other session caching attributes.
For example, you could not limit the number of sessions created.
To support that requirement and other configuration options, a CachingSessionFactory
was added.
It provides sessionCacheSize
and sessionWaitTimeout
properties.
The sessionCacheSize
property controls how many active sessions the factory maintains in its cache (the default is unbounded).
If the sessionCacheSize
threshold has been reached, any attempt to acquire another session blocks until either one of the cached sessions becomes available or until the wait time for a session expires (the default wait time is Integer.MAX_VALUE
).
The sessionWaitTimeout
property configures that value.
If you want your sessions to be cached, configure your default session factory as described earlier and then wrap it in an instance of CachingSessionFactory
, where you can provide those additional properties.
The following example shows how to do so:
<bean id="ftpSessionFactory" class="o.s.i.ftp.session.DefaultFtpSessionFactory">
<property name="host" value="localhost"/>
</bean>
<bean id="cachingSessionFactory" class="o.s.i.file.remote.session.CachingSessionFactory">
<constructor-arg ref="ftpSessionFactory"/>
<constructor-arg value="10"/>
<property name="sessionWaitTimeout" value="1000"/>
</bean>
The preceding example shows a CachingSessionFactory
created with the sessionCacheSize
set to 10
and the
sessionWaitTimeout
set to one second (its value is in milliseconds).
Starting with Spring Integration 3.0, the CachingConnectionFactory
provides a resetCache()
method.
When invoked, all idle sessions are immediately closed and in-use sessions are closed when they are returned to the cache.
New requests for sessions establish new sessions as necessary.
Starting with version 5.1, the CachingSessionFactory
has a new property testSession
.
When true, the session will be tested by sending a NOOP command to ensure it is still active; if not, it will be removed from the cache; a new session is created if no active sessions are in the cache.
18.11. Using RemoteFileTemplate
Starting with Spring Integration 3.0, a new abstraction is provided over the FtpSession
object.
The template provides methods to send, retrieve (as an InputStream
), remove, and rename files.
In addition an execute
method is provided allowing the caller to execute multiple operations on the session.
In all cases, the template takes care of reliably closing the session.
For more information, see the
Javadoc for RemoteFileTemplate
.
There is a subclass for FTP: FtpRemoteFileTemplate
.
Version 4.1 added added additional methods, including getClientInstance()
, which provides access to the underlying FTPClient
and thus gives you access to low-level APIs.
Not all FTP servers properly implement the STAT <path>
command.
Some return a positive result for a non-existent path.
The NLST
command reliably returns the name when the path is a file and it exists.
However, this does not support checking that an empty directory exists since NLST
always returns an empty list when the path is a directory.
Since the template does not know whether the path represents a directory, it has to perform additional checks when the path does not appear to exist (when using NLST
).
This adds overhead, requiring several requests to the server.
Starting with version 4.1.9, the FtpRemoteFileTemplate
provides the FtpRemoteFileTemplate.ExistsMode
property, which has the following options:
-
STAT
: Perform theSTAT
FTP command (FTPClient.getStatus(path)
) to check the path existence. This is the default and requires that your FTP server properly support theSTAT
command (with a path). -
NLST
: Perform theNLST
FTP command —FTPClient.listName(path)
. Use this if you are testing for a path that is a full path to a file. It does not work for empty directories. -
NLST_AND_DIRS
: Perform theNLST
command first and, if it returns no files, fall back to a technique that temporarily switches the working directory by usingFTPClient.changeWorkingDirectory(path)
. SeeFtpSession.exists()
for more information.
Since we know that the FileExistsMode.FAIL
case is always only looking for a file (and not a directory), we safely use NLST
mode for the FtpMessageHandler
and FtpOutboundGateway
components.
For any other cases, the FtpRemoteFileTemplate
can be extended to implement custom logic in the overridden exist()
method.
Starting with version 5.0, the new RemoteFileOperations.invoke(OperationsCallback<F, T> action)
method is available.
This method lets several RemoteFileOperations
calls be called in the scope of the same, thread-bounded, Session
.
This is useful when you need to perform several high-level operations of the RemoteFileTemplate
as one unit of work.
For example, AbstractRemoteFileOutboundGateway
uses it with the mput
command implementation, where we perform a put
operation for each file in the provided directory and recursively for its sub-directories.
See the Javadoc for more information.
18.12. Using MessageSessionCallback
Starting with Spring Integration 4.2, you can use a MessageSessionCallback<F, T>
implementation with the
<int-ftp:outbound-gateway/>
(FtpOutboundGateway
in Java) to perform any operations on the Session<FTPFile>
with
the requestMessage
context.
It can be used for any non-standard or low-level FTP operations and allows access from an integration flow definition and functional interface (Lambda) implementation injection, as the following example shows:
@Bean
@ServiceActivator(inputChannel = "ftpChannel")
public MessageHandler ftpOutboundGateway(SessionFactory<FTPFile> sessionFactory) {
return new FtpOutboundGateway(sessionFactory,
(session, requestMessage) -> session.list(requestMessage.getPayload()));
}
Another example might be to pre- or post-process the file data being sent or retrieved.
When using XML configuration, the <int-ftp:outbound-gateway/>
provides a session-callback
attribute to let you specify the MessageSessionCallback
bean name.
The session-callback is mutually exclusive with the command and expression attributes.
When configuring with Java, different constructors are available in the FtpOutboundGateway class.
|
19. Pivotal GemFire and Apache Geode Support
Spring Integration provides support for Pivotal GemFire and Apache Geode.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-gemfire</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-gemfire:5.1.5.RELEASE"
GemFire is a distributed data management platform that provides a key-value data grid along with advanced distributed system features, such as event processing, continuous querying, and remote function execution. This guide assumes some familiarity with the commercial Pivotal GemFire or Open Source Apache Geode.
Spring integration provides support for GemFire by implementing inbound adapters for entry and continuous query events, an outbound adapter to write entries to the cache, and message and metadata stores and GemfireLockRegistry
implementations.
Spring integration leverages the Spring Data for Pivotal GemFire project, providing a thin wrapper over its components.
Starting with version 5.1, the Spring Integration GemFire module uses the Spring Data for Apache Geode transitive dependency by default.
To switch to the commercial Pivotal GemFire-based Spring Data for Pivotal GemFire, exclude spring-data-geode
from dependencies and add spring-data-gemfire
, as the following Maven snippet shows:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-gemfire</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-geode</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-gemfire</artifactId>
</dependency>
To configure the 'int-gfe' namespace, include the following elements within the headers of your XML configuration file:
xmlns:int-gfe="http://www.springframework.org/schema/integration/gemfire"
xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire
https://www.springframework.org/schema/integration/gemfire/spring-integration-gemfire.xsd"
19.1. Inbound Channel Adapter
The inbound channel adapter produces messages on a channel when triggered by a GemFire EntryEvent
.
GemFire generates events whenever an entry is CREATED
, UPDATED
, DESTROYED
, or INVALIDATED
in the associated region.
The inbound channel adapter lets you filter on a subset of these events.
For example, you may want to produce messages only in response to an entry being created.
In addition, the inbound channel adapter can evaluate a SpEL expression if, for example, you want your message payload to contain an event property such as the new entry value.
The following example shows how to configure an inbound channel adapter with a SpEL language (in the expression
attribute):
<gfe:cache/>
<gfe:replicated-region id="region"/>
<int-gfe:inbound-channel-adapter id="inputChannel" region="region"
cache-events="CREATED" expression="newValue"/>
The preceding configuration creates a GemFire Cache
and Region
by using Spring GemFire’s 'gfe' namespace.
The inbound-channel-adapter
element requires a reference to the GemFire region on which the adapter listens for events.
Optional attributes include cache-events
, which can contain a comma-separated list of event types for which a message is produced on the input channel.
By default, CREATED
and UPDATED
are enabled.
If no channel
attribute is provided, the channel is created from the id
attribute.
This adapter also supports an error-channel
.
The GemFire EntryEvent
is the #root
object of the expression
evaluation.
The following example shows an expression that replaces a value for a key:
expression="new something.MyEvent(key, oldValue, newValue)"
If the expression
attribute is not provided, the message payload is the GemFire EntryEvent
itself.
| This adapter conforms to Spring Integration conventions. |
19.2. Continuous Query Inbound Channel Adapter
The continuous query inbound channel adapter produces messages on a channel when triggered by a GemFire continuous query or CqEvent
event.
In release 1.1
, Spring Data introduced continuous query support, including ContinuousQueryListenerContainer
, which provides a nice abstraction over the GemFire native API.
This adapter requires a reference to a ContinuousQueryListenerContainer
instance, creates a listener for a given query
, and executes the query.
The continuous query acts as an event source that fires whenever its result set changes state.
| GemFire queries are written in OQL and are scoped to the entire cache (not just one region). Additionally, continuous queries require a remote (that is, running in a separate process or remote host) cache server. See the GemFire documentation for more information on implementing continuous queries. |
The following configuration creates a GemFire client cache (recall that a remote cache server is required for this implementation and its address is configured as a child element of the pool), a client region, and a ContinuousQueryListenerContainer
that uses Spring Data:
<gfe:client-cache id="client-cache" pool-name="client-pool"/>
<gfe:pool id="client-pool" subscription-enabled="true" >
<!--configure server or locator here required to address the cache server -->
</gfe:pool>
<gfe:client-region id="test" cache-ref="client-cache" pool-name="client-pool"/>
<gfe:cq-listener-container id="queryListenerContainer" cache="client-cache"
pool-name="client-pool"/>
<int-gfe:cq-inbound-channel-adapter id="inputChannel"
cq-listener-container="queryListenerContainer"
query="select * from /test"/>
The continuous query inbound channel adapter requires a cq-listener-container
attribute, which must contain a reference to the ContinuousQueryListenerContainer
.
Optionally, it accepts an expression
attribute that uses SpEL to transform the CqEvent
or extract an individual property as needed.
The cq-inbound-channel-adapter
provides a query-events
attribute that contains a comma-separated list of event types for which a message is produced on the input channel.
The available event types are CREATED
, UPDATED
, DESTROYED
, REGION_DESTROYED
, and REGION_INVALIDATED
.
By default, CREATED
and UPDATED
are enabled.
Additional optional attributes include query-name
(which provides an optional query name), expression
(which works as described in the preceding section), and durable
(a boolean value indicating if the query is durable — it is false by default).
If you do not provide a channel
, the channel is created from the id
attribute.
This adapter also supports an error-channel
.
| This adapter conforms to Spring Integration conventions. |
19.3. Outbound Channel Adapter
The outbound channel adapter writes cache entries that are mapped from the message payload.
In its simplest form, it expects a payload of type java.util.Map
and puts the map entries into its configured region.
The following example shows how to configure an outbound channel adapter:
<int-gfe:outbound-channel-adapter id="cacheChannel" region="region"/>
Given the preceding configuration, an exception is thrown if the payload is not a Map
.
Additionally, you can configure the outbound channel adapter to create a map of cache entries by using SpEL.
The following example shows how to do so:
<int-gfe:outbound-channel-adapter id="cacheChannel" region="region">
<int-gfe:cache-entries>
<entry key="payload.toUpperCase()" value="payload.toLowerCase()"/>
<entry key="'thing1'" value="'thing2'"/>
</int-gfe:cache-entries>
</int-gfe:outbound-channel-adapter>
In the preceding configuration, the inner element (cache-entries
) is semantically equivalent to a Spring 'map' element.
The adapter interprets the key
and value
attributes as SpEL expressions with the message as the evaluation context.
Note that this can contain arbitrary cache entries (not only those derived from the message) and that literal values must be enclosed in single quotes.
In the preceding example, if the message sent to cacheChannel
has a String
payload with a value Hello
, two entries ([HELLO:hello, thing1:thing2]
) are written (either created or updated) in the cache region.
This adapter also supports the order
attribute, which may be useful if it is bound to a PublishSubscribeChannel
.
19.4. Gemfire Message Store
As described in EIP, a message store lets you persist messages.
This can be useful when dealing with components that have a capability to buffer messages (QueueChannel
, Aggregator
, Resequencer
, and others) if reliability is a concern.
In Spring Integration, the MessageStore
strategy interface also provides the foundation for the claim check pattern, which is described in EIP as well.
Spring Integration’s Gemfire module provides GemfireMessageStore
, which is an implementation of both the the MessageStore
strategy (mainly used by the QueueChannel
and ClaimCheck
patterns) and the MessageGroupStore
strategy (mainly used by the Aggregator
and Resequencer
patterns).
The following example configures the cache and region by using the spring-gemfire
namespace (not to be confused with the spring-integration-gemfire
namespace):
<bean id="gemfireMessageStore" class="o.s.i.gemfire.store.GemfireMessageStore">
<constructor-arg ref="myRegion"/>
</bean>
<gfe:cache/>
<gfe:replicated-region id="myRegion"/>
<int:channel id="somePersistentQueueChannel">
<int:queue message-store="gemfireMessageStore"/>
<int:channel>
<int:aggregator input-channel="inputChannel" output-channel="outputChannel"
message-store="gemfireMessageStore"/>
Often, it is desirable for the message store to be maintained in one or more remote cache servers in a client-server configuration.
In this case, you should configure a client cache, a client region, and a client pool and inject the region into the MessageStore
.
The following example shows how to do so:
<bean id="gemfireMessageStore"
class="org.springframework.integration.gemfire.store.GemfireMessageStore">
<constructor-arg ref="myRegion"/>
</bean>
<gfe:client-cache/>
<gfe:client-region id="myRegion" shortcut="PROXY" pool-name="messageStorePool"/>
<gfe:pool id="messageStorePool">
<gfe:server host="localhost" port="40404" />
</gfe:pool>
Note that the pool
element is configured with the address of a cache server (you can substitute a locator here).
The region is configured as a 'PROXY' so that no data is stored locally.
The region’s id
corresponds to a region with the same name in the cache server.
Starting with version 4.3.12, the GemfireMessageStore
supports the key prefix
option to allow distinguishing between instances of the store on the same GemFire region.
19.5. Gemfire Lock Registry
Starting with version 4.0, the GemfireLockRegistry
is available.
Certain components (for example, the aggregator and the resequencer) use a lock obtained from a LockRegistry
instance to ensure that only one thread is manipulating a group at any given time.
The DefaultLockRegistry
performs this function within a single component.
You can now configure an external lock registry on these components.
When you use a shared MessageGroupStore
with the GemfireLockRegistry
, it can provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.
One of the GemfireLockRegistry constructors requires a Region as an argument.
It is used to obtain a Lock from the getDistributedLock() method.
This operation requires GLOBAL scope for the Region .
Another constructor requires a Cache , and the Region is created with GLOBAL scope and with the name, LockRegistry .
|
19.6. Gemfire Metadata Store
Version 4.0 introduced a new Gemfire-based MetadataStore
(Metadata Store) implementation.
You can use the GemfireMetadataStore
to maintain metadata state across application restarts.
This new MetadataStore
implementation can be used with adapters such as:
To get these adapters to use the new GemfireMetadataStore
, declare a Spring bean with a bean name of metadataStore
.
The feed inbound channel adapter automatically picks up and use the declared GemfireMetadataStore
.
The GemfireMetadataStore also implements ConcurrentMetadataStore , letting it be reliably shared across multiple application instances, where only one instance can store or modify a key’s value.
These methods give various levels of concurrency guarantees based on the scope and data policy of the region.
They are implemented in the peer cache and client-server cache but are disallowed in peer regions that have NORMAL or EMPTY data policies.
|
Since version 5.0, the GemfireMetadataStore also implements ListenableMetadataStore , which lets you listen to cache events by providing MetadataStoreListener instances to the store, as the following example shows:
|
GemfireMetadataStore metadataStore = new GemfireMetadataStore(cache);
metadataStore.addListener(new MetadataStoreListenerAdapter() {
@Override
public void onAdd(String key, String value) {
...
}
});
20. HTTP Support
Spring Integration’s HTTP support allows for the running of HTTP requests and the processing of inbound HTTP requests.
The HTTP support consists of the following gateway implementations: HttpInboundEndpoint
and HttpRequestExecutingMessageHandler
.
See also WebFlux Support.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-http</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-http:5.1.5.RELEASE"
The javax.servlet:javax.servlet-api
dependency must be provided on the target Servlet container.
20.1. Http Inbound Components
To receive messages over HTTP, you need to use an HTTP inbound channel adapter or an HTTP inbound gateway.
To support the HTTP inbound adapters, they need to be deployed within a servlet container such as Apache Tomcat or Jetty.
The easiest way to do this is to use Spring’s
HttpRequestHandlerServlet
, by providing the following servlet definition in the web.xml
file:
<servlet>
<servlet-name>inboundGateway</servlet-name>
<servlet-class>o.s.web.context.support.HttpRequestHandlerServlet</servlet-class>
</servlet>
Notice that the servlet name matches the bean name.
For more information on using the HttpRequestHandlerServlet
, see
Remoting and web services using Spring,
which is part of the Spring Framework Reference documentation.
If you are running within a Spring MVC application, then the aforementioned explicit servlet definition is not necessary. In that case, the bean name for your gateway can be matched against the URL path as you would for a Spring MVC Controller bean. For more information, see Web MVC framework, which is part of the Spring Framework Reference documentation.
| For a sample application and the corresponding configuration, see the Spring Integration Samples repository. It contains the HTTP sample application, which demonstrates Spring Integration’s HTTP support. |
The following example bean defines an HTTP inbound endpoint:
<bean id="httpInbound"
class="org.springframework.integration.http.inbound.HttpRequestHandlingMessagingGateway">
<property name="requestChannel" ref="httpRequestChannel" />
<property name="replyChannel" ref="httpReplyChannel" />
</bean>
The HttpRequestHandlingMessagingGateway
accepts a list of HttpMessageConverter
instances or else relies on a default list.
The converters allow customization of the mapping from HttpServletRequest
to Message
.
The default converters encapsulate simple strategies, which (for example) create a String
message for a POST
request where the content type starts with text
.
See the Javadoc for full details.
An additional flag (mergeWithDefaultConverters
) can be set along with the list of custom HttpMessageConverter
to add the default converters after the custom converters.
By default, this flag is set to false
, meaning that the custom converters replace the default list.
The message conversion process uses the (optional) requestPayloadType
property and the incoming Content-Type
header.
Starting with version 4.3, if a request has no content type header, application/octet-stream
is assumed, as
recommended by RFC 2616
.
Previously, the body of such messages was ignored.
Spring Integration 2.0 implemented multipart file support.
If the request has been wrapped as a MultipartHttpServletRequest
, when you use the default converters, that request is converted to a Message
payload that is a MultiValueMap
containing values that may be byte arrays, strings, or instances of Spring’s MultipartFile
, depending on the content type of the individual parts.
The HTTP inbound endpoint locates a MultipartResolver in the context if one has a bean name of multipartResolver (the same name expected by Spring’s DispatcherServlet ).
If it does locate that bean, the support for multipart files is enabled on the inbound request mapper.
Otherwise, it fails when it tries to map a multipart file request to a Spring Integration Message .
For more on Spring’s support for MultipartResolver , see the Spring Reference Manual.
|
|
If you wish to proxy a
|
When you send a response to the client, you have a number of ways to customize the behavior of the gateway.
By default, the gateway acknowledges that the request was received by sending a 200
status code back.
It is possible to customize this response by providing a 'viewName' to be resolved by the Spring MVC ViewResolver
.
If the gateway should expect a reply to the Message
, you can set the expectReply
flag (constructor argument) to cause the gateway to wait for a reply Message
before creating an HTTP response.
The following example configures a gateway to serve as a Spring MVC Controller with a view name:
<bean id="httpInbound"
class="org.springframework.integration.http.inbound.HttpRequestHandlingController">
<constructor-arg value="true" /> <!-- indicates that a reply is expected -->
<property name="requestChannel" ref="httpRequestChannel" />
<property name="replyChannel" ref="httpReplyChannel" />
<property name="viewName" value="jsonView" />
<property name="supportedMethodNames" >
<list>
<value>GET</value>
<value>DELETE</value>
</list>
</property>
</bean>
Because of the constructor-arg
value of true
, it waits for a reply.
The preceding example also shows how to customize the HTTP methods accepted by the gateway, which are POST
and GET
by default.
The reply message is available in the model map. By default, the key for that map entry is 'reply', but you can override this default by setting the 'replyKey' property on the endpoint’s configuration.
20.2. HTTP Outbound Components
This section describes Spring Integration’s HTTP outbound components
20.2.1. Using HttpRequestExecutingMessageHandler
To configure the HttpRequestExecutingMessageHandler
, write a bean definition similar to the following:
<bean id="httpOutbound"
class="org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler">
<constructor-arg value="http://localhost:8080/example" />
<property name="outputChannel" ref="responseChannel" />
</bean>
This bean definition runs HTTP requests by delegating to a RestTemplate
.
That template, in turn, delegates to a list of HttpMessageConverter
instances to generate the HTTP request body from the Message
payload.
You can configure those converters as well as the ClientHttpRequestFactory
instance to use, as the following example shows:
<bean id="httpOutbound"
class="org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler">
<constructor-arg value="http://localhost:8080/example" />
<property name="outputChannel" ref="responseChannel" />
<property name="messageConverters" ref="messageConverterList" />
<property name="requestFactory" ref="customRequestFactory" />
</bean>
By default, the HTTP request is generated by using an instance of SimpleClientHttpRequestFactory
, which uses the JDK HttpURLConnection
.
Use of the Apache Commons HTTP Client is also supported through CommonsClientHttpRequestFactory
, which you can inject (as shown earlier).
| For the outbound gateway, the reply message produced by the gateway contains all the message headers that are present in the request message. |
20.2.2. Using Cookies
Basic cookie support is provided by the transfer-cookies
attribute on the outbound gateway.
When set to true
(the default is false
), a Set-Cookie
header received from the server in a response is converted to Cookie
in the reply message.
This header is then used on subsequent sends.
This enables simple stateful interactions, such as the following:
…→logonGateway→…→doWorkGateway→…→logoffGateway→…
If transfer-cookies
is false
, any Set-Cookie
header received remains as Set-Cookie
in the reply message and is dropped on subsequent sends.
|
Empty Response Bodies
HTTP is a request-response protocol.
However, the response may not have a body, only headers.
In this case, the |
|
expected-response-type
Further to the preceding note about empty response bodies, if a response does contain a body, you must provide an appropriate |
20.3. HTTP Namespace Support
Spring Integration provides an http
namespace and the corresponding schema definition.
To include it in your configuration, provide the following namespace declaration in your application context configuration file:
<?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:int="http://www.springframework.org/schema/integration"
xmlns:int-http="http://www.springframework.org/schema/integration/http"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/http
https://www.springframework.org/schema/integration/http/spring-integration-http.xsd">
...
</beans>
20.3.1. Inbound
The XML namespace provides two components for handling HTTP inbound requests: inbound-channel-adapter
and inbound-gateway
.
In order to process requests without returning a dedicated response, use the inbound-channel-adapter
.
The following example shows how to configure one:
<int-http:inbound-channel-adapter id="httpChannelAdapter" channel="requests"
supported-methods="PUT, DELETE"/>
To process requests that do expect a response, use an inbound-gateway
.
The following example shows how to configure one:
<int-http:inbound-gateway id="inboundGateway"
request-channel="requests"
reply-channel="responses"/>
20.3.2. Request Mapping Support
Spring Integration 3.0 improved the REST support by introducing the IntegrationRequestMappingHandlerMapping .
The implementation relies on the enhanced REST support provided by Spring Framework 3.1 or higher.
|
The parsing of the HTTP inbound gateway or the HTTP inbound channel adapter registers an integrationRequestMappingHandlerMapping
bean of type IntegrationRequestMappingHandlerMapping
, in case one is not yet registered.
This particular implementation of the HandlerMapping
delegates its logic to RequestMappingInfoHandlerMapping
.
The implementation provides functionality similar to the org.springframework.web.bind.annotation.RequestMapping
annotation in Spring MVC.
For more information, see Mapping Requests With @RequestMapping .
|
For this purpose, Spring Integration 3.0 introduces the <request-mapping>
element.
You can add this optional element to <http:inbound-channel-adapter>
and <http:inbound-gateway>
.
It works in conjunction with the path
and supported-methods
attributes.
The following example shows how to configure it on an inbound gateway:
<inbound-gateway id="inboundController"
request-channel="requests"
reply-channel="responses"
path="/foo/{fooId}"
supported-methods="GET"
view-name="foo"
error-code="oops">
<request-mapping headers="User-Agent"
params="myParam=myValue"
consumes="application/json"
produces="!text/plain"/>
</inbound-gateway>
Based on the preceding configuration, the namespace parser creates an instance of the IntegrationRequestMappingHandlerMapping
(if none exists) and an HttpRequestHandlingController
bean and associates with it an instance of RequestMapping
.
This RequestMapping
instance is, in turn, converted to the Spring MVC RequestMappingInfo
.
The <request-mapping>
element provides the following attributes:
-
headers
-
params
-
consumes
-
produces
With the path
and supported-methods
attributes of the <http:inbound-channel-adapter>
or the <http:inbound-gateway>
, <request-mapping>
attributes translate directly into the respective options provided by the org.springframework.web.bind.annotation.RequestMapping
annotation in Spring MVC.
The <request-mapping>
element lets you configure several Spring Integration HTTP inbound endpoints to the same path
(or even the same supported-methods
) and lets you provide different downstream message flows based on incoming HTTP requests.
Alternatively, you can also declare only one HTTP inbound endpoint and apply routing and filtering logic within the Spring Integration flow to achieve the same result.
This lets you get the Message
into the flow as early as possibly.
The following example shows how to do so:
<int-http:inbound-gateway request-channel="httpMethodRouter"
supported-methods="GET,DELETE"
path="/process/{entId}"
payload-expression="#pathVariables.entId"/>
<int:router input-channel="httpMethodRouter" expression="headers.http_requestMethod">
<int:mapping value="GET" channel="in1"/>
<int:mapping value="DELETE" channel="in2"/>
</int:router>
<int:service-activator input-channel="in1" ref="service" method="getEntity"/>
<int:service-activator input-channel="in2" ref="service" method="delete"/>
For more information regarding handler mappings, see the Spring Framework Web Servlet documentation or the Spring Framework Web Reactive documentation.
20.3.3. Cross-origin Resource Sharing (CORS) Support
Starting with version 4.2, you can configure the <http:inbound-channel-adapter>
and <http:inbound-gateway>
with
a <cross-origin>
element.
It represents the same options as Spring MVC’s @CrossOrigin
for @Controller
annotations and allows the configuration of cross-origin resource sharing (CORS) for Spring Integration HTTP endpoints:
-
origin
: List of allowed origins.means that all origins are allowed. These values are placed in the
Access-Control-Allow-Origin
header of both the pre-flight and actual responses. The default value is.
-
allowed-headers
: Indicates which request headers can be used during the actual request.means that all headers requested by the client are allowed. This property controls the value of the pre-flight response’s
Access-Control-Allow-Headers
header. The default value is.
-
exposed-headers
: List of response headers that the user-agent lets the client access. This property controls the value of the actual response’sAccess-Control-Expose-Headers
header. -
method
: The HTTP request methods to allow:GET
,POST
,HEAD
,OPTIONS
,PUT
,PATCH
,DELETE
,TRACE
. Methods specified here overrides those insupported-methods
. -
allow-credentials
: Set totrue
if the the browser should include any cookies associated to the domain of the request orfalse
if it should not. An empty string ('""') means undefined. Iftrue
, the pre-flight response includes theAccess-Control-Allow-Credentials=true
header. The default value istrue
. -
max-age
: Controls the cache duration for pre-flight responses. Setting this to a reasonable value can reduce the number of pre-flight request-response interactions required by the browser. This property controls the value of theAccess-Control-Max-Age
header in the pre-flight response. A value of-1
means undefined. The default value is 1800 seconds (30 minutes).
The CORS Java Configuration is represented by the org.springframework.integration.http.inbound.CrossOrigin
class,
instances of which can be injected into the HttpRequestHandlingEndpointSupport
beans.
20.3.4. Response Status Code
Starting with version 4.1, you can configure the <http:inbound-channel-adapter>
with a status-code-expression
to override the default 200 OK
status.
The expression must return an object that can be converted to an org.springframework.http.HttpStatus
enum value.
The evaluationContext
has a BeanResolver
and, starting with version 5.1, is supplied with the RequestEntity<?>
as root object.
An example might be to resolve, at runtime, some scoped bean that returns a status code value.
However, most likely, it is set to a fixed value such as status-code=expression="204"
(No Content), or status-code-expression="T(org.springframework.http.HttpStatus).NO_CONTENT"
.
By default, status-code-expression
is null, meaning that the normal '200 OK' response status is returned.
Using the RequestEntity<?>
as root object, the status code can be conditional e.g. on the request method, some header, URI content or even request body.
The following example shows how to set the status code to ACCEPTED
:
<http:inbound-channel-adapter id="inboundController"
channel="requests" view-name="foo" error-code="oops"
status-code-expression="T(org.springframework.http.HttpStatus).ACCEPTED">
<request-mapping headers="BAR"/>
</http:inbound-channel-adapter>
The <http:inbound-gateway>
resolves the 'status code' from the http_statusCode
header of the reply Message
.
Starting with version 4.2, the default response status code when no reply is received within the reply-timeout
is 500 Internal Server Error
.
There are two ways to modify this behavior:
-
Add a
reply-timeout-status-code-expression
. This has the same semantics as thestatus-code-expression
on the inbound adapter. -
Add an
error-channel
and return an appropriate message with an HTTP status code header, as the following example shows:<int:chain input-channel="errors"> <int:header-enricher> <int:header name="http_statusCode" value="504" /> </int:header-enricher> <int:transformer expression="payload.failedMessage" /> </int:chain>
The payload of the ErrorMessage
is a MessageTimeoutException
.
It must be transformed to something that can be converted by the gateway, such as a String
.
A good candidate is the exception’s message property, which is the value used when you use the expression
technique.
If the error flow times out after a main flow timeout, 500 Internal Server Error
is returned, or, if the
reply-timeout-status-code-expression
is present, it is evaluated.
Previously, the default status code for a timeout was 200 OK .
To restore that behavior, set reply-timeout-status-code-expression="200" .
|
20.3.5. URI Template Variables and Expressions
By using the path
attribute in conjunction with the payload-expression
attribute and the header
element, you have a high degree of flexibility for mapping inbound request data.
In the following example configuration, an inbound channel adapter is configured to accept requests using the following URI:
/first-name/{firstName}/last-name/{lastName}
When you use the payload-expression
attribute, the Mark
URI template variable maps to be the Message
payload, while the Fisher
URI template variable maps to the lname
message header, as defined in the following example:
<int-http:inbound-channel-adapter id="inboundAdapterWithExpressions"
path="/first-name/{firstName}/last-name/{lastName}"
channel="requests"
payload-expression="#pathVariables.firstName">
<int-http:header name="lname" expression="#pathVariables.lastName"/>
</int-http:inbound-channel-adapter>
For more information about URI template variables, see uri template patterns in the Spring Reference Manual.
Since Spring Integration 3.0, in addition to the existing #pathVariables
and #requestParams
variables being available in payload and header expressions, we added other useful expression variables:
-
#requestParams
: TheMultiValueMap
from theServletRequest
parameterMap
. -
#pathVariables
: TheMap
from URI Template placeholders and their values. -
#matrixVariables
: TheMap
ofMultiValueMap
according to the Spring MVC Specification. Note that#matrixVariables
requires Spring MVC 3.2 or higher. -
#requestAttributes
: Theorg.springframework.web.context.request.RequestAttributes
associated with the current request. -
#requestHeaders
: Theorg.springframework.http.HttpHeaders
object from the current request. -
#cookies
: TheMap<String, Cookie>
ofjavax.servlet.http.Cookie
instances from the current request.
Note that all these values (and others) can be accessed within expressions in the downstream message flow through the ThreadLocal
org.springframework.web.context.request.RequestAttributes
variable, if that message flow is single-threaded and lives within the request thread.
The following example configures a transformer that uses an expression
attribute:
<int-:transformer
expression="T(org.springframework.web.context.request.RequestContextHolder).
requestAttributes.request.queryString"/>
20.3.6. Outbound
To configure the outbound gateway, you can use the namespace support. The following code snippet shows the available configuration options for an outbound HTTP gateway:
<int-http:outbound-gateway id="example"
request-channel="requests"
url="http://localhost/test"
http-method="POST"
extract-request-payload="false"
expected-response-type="java.lang.String"
charset="UTF-8"
request-factory="requestFactory"
reply-timeout="1234"
reply-channel="replies"/>
Most importantly, notice that the 'http-method' and 'expected-response-type' attributes are provided.
Those are two of the most commonly configured values.
The default http-method
is POST
, and the default response type is null.
With a null response type, the payload of the reply Message
contains the ResponseEntity
, as long as its HTTP status is a success (non-successful status codes throw exceptions).
If you expect a different type, such as a String
, provide that as a fully-qualified class name (java.lang.String
in the preceding example).
See also the note about empty response bodies in HTTP Outbound Components.
Beginning with Spring Integration 2.1, the request-timeout attribute of the HTTP outbound gateway was renamed to reply-timeout to better reflect its intent.
|
|
Since Spring Integration 2.2, Java serialization over HTTP is no longer enabled by default.
Previously, when setting the However, because this could cause incompatibility with existing applications, it was decided to no longer automatically add this converter to the HTTP endpoints.
If you wish to use Java serialization, you can add the |
Beginning with Spring Integration 2.2, you can also determine the HTTP method dynamically by using SpEL and the http-method-expression
attribute.
Note that this attribute is mutually exclusive with http-method
.
You can also use the expected-response-type-expression
attribute instead of expected-response-type
and provide any valid SpEL expression that determines the type of the response.
The following configuration example uses expected-response-type-expression
:
<int-http:outbound-gateway id="example"
request-channel="requests"
url="http://localhost/test"
http-method-expression="headers.httpMethod"
extract-request-payload="false"
expected-response-type-expression="payload"
charset="UTF-8"
request-factory="requestFactory"
reply-timeout="1234"
reply-channel="replies"/>
If your outbound adapter is to be used in a unidirectional way, you can use an outbound-channel-adapter
instead.
This means that a successful response executes without sending any messages to a reply channel.
In the case of any non-successful response status code, it throws an exception.
The configuration looks very similar to the gateway, as the following example shows:
<int-http:outbound-channel-adapter id="example"
url="http://localhost/example"
http-method="GET"
channel="requests"
charset="UTF-8"
extract-payload="false"
expected-response-type="java.lang.String"
request-factory="someRequestFactory"
order="3"
auto-startup="false"/>
|
To specify the URL, you can use either the 'url' attribute or the 'url-expression' attribute.
The 'url' attribute takes a simple string (with placeholders for URI variables, as described below).
The 'url-expression' is a SpEL expression, with the In previous releases, some users used the place holders to replace the entire URL with a URI variable. Changes in Spring 3.1 can cause some issues with escaped characters, such as '?'. For this reason, we recommend that, if you wish to generate the URL entirely at runtime, you use the 'url-expression' attribute. |
20.3.7. Mapping URI Variables
If your URL contains URI variables, you can map them by using the uri-variable
element.
This element is available for the HTTP outbound gateway and the HTTP outbound channel adapter.
The following example maps the zipCode
URI variable to an expression:
<int-http:outbound-gateway id="trafficGateway"
url="https://local.yahooapis.com/trafficData?appid=YdnDemo&zip={zipCode}"
request-channel="trafficChannel"
http-method="GET"
expected-response-type="java.lang.String">
<int-http:uri-variable name="zipCode" expression="payload.getZip()"/>
</int-http:outbound-gateway>
The uri-variable
element defines two attributes: name
and expression
.
The name
attribute identifies the name of the URI variable, while the expression
attribute is used to set the actual value.
By using the expression
attribute, you can leverage the full power of the Spring Expression Language (SpEL), which gives you full dynamic access to the message payload and the message headers.
For example, in the preceding configuration, the getZip()
method is invoked on the payload object of the Message
and the result of that method is used as the value of the URI variable named 'zipCode'.
Since Spring Integration 3.0, HTTP outbound endpoints support the uri-variables-expression
attribute to specify an expression
that should be evaluated, resulting in a Map
of all URI variable placeholders within the URL template.
It provides a mechanism whereby you can use different variable expressions, based on the outbound message.
This attribute is mutually exclusive with the <uri-variable/>
element.
The following example shows how to use the uri-variables-expression
attribute:
<int-http:outbound-gateway
url="https://foo.host/{foo}/bars/{bar}"
request-channel="trafficChannel"
http-method="GET"
uri-variables-expression="@uriVariablesBean.populate(payload)"
expected-response-type="java.lang.String"/>
uriVariablesBean
might be defined as follows:
public class UriVariablesBean {
private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
public Map<String, ?> populate(Object payload) {
Map<String, Object> variables = new HashMap<String, Object>();
if (payload instanceOf String.class)) {
variables.put("foo", "foo"));
}
else {
variables.put("foo", EXPRESSION_PARSER.parseExpression("headers.bar"));
}
return variables;
}
}
The uri-variables-expression must evaluate to a Map .
The values of the Map must be instances of String or Expression .
This Map is provided to an ExpressionEvalMap for further resolution of URI variable placeholders by using those expressions in the context of the outbound Message .
|
IMPORTANT
The uriVariablesExpression
property provides a very powerful mechanism for evaluating URI variables.
We anticipate that people mostly use simple expressions, such as the preceding example.
However, you can also configure something such as "@uriVariablesBean.populate(#root)"
with an expression in the returned map being variables.put("thing1", EXPRESSION_PARSER.parseExpression(message.getHeaders().get("thing2", String.class)));
, where the expression is dynamically provided in the message header named thing2
.
Since the header may come from an untrusted source, the HTTP outbound endpoints use SimpleEvaluationContext
when evaluating these expressions.
SimpleEvaluationContext
uses only a subset of SpEL features.
If you trust your message sources and wish to use the restricted SpEL constructs, set the trustedSpel
property of the outbound endpoint to true
.
You can achieve scenarios that need to supply a dynamic set of URI variables on a per-message basis by using a custom url-expression
and some utilities for building and encoding URL parameters.
The following example shows how to do so:
url-expression="T(org.springframework.web.util.UriComponentsBuilder)
.fromHttpUrl('http://HOST:PORT/PATH')
.queryParams(payload)
.build()
.toUri()"
The queryParams()
method expects a MultiValueMap<String, String>
as an argument, so you can build a real set of URL query parameters in advance, before performing the request.
The whole queryString
can also be presented as a uri-variable
, as the following example shows:
<int-http:outbound-gateway id="proxyGateway" request-channel="testChannel"
url="http://testServer/test?{queryString}">
<int-http:uri-variable name="queryString" expression="'a=A&b=B'"/>
</int-http:outbound-gateway>
In this case, you must manually provide the URL encoding.
For example, you can use the org.apache.http.client.utils.URLEncodedUtils#format()
for this purpose.
As mentioned earlier, a manually built MultiValueMap<String, String>
can be converted to the the List<NameValuePair>
format()
method argument by using the following Java Streams snippet:
List<NameValuePair> nameValuePairs =
params.entrySet()
.stream()
.flatMap(e -> e
.getValue()
.stream()
.map(v -> new BasicNameValuePair(e.getKey(), v)))
.collect(Collectors.toList());
20.3.8. Controlling URI Encoding
By default, the URL string is encoded (see UriComponentsBuilder
) to the URI object before sending the request.
In some scenarios with a non-standard URI (such as the RabbitMQ REST API), it is undesirable to perform the encoding.
The <http:outbound-gateway/>
and <http:outbound-channel-adapter/>
provide an encode-uri
attribute.
To disable encoding the URL, set this attribute to false
(by default, it is true
).
If you wish to partially encode some of the URL, use an expression
within a <uri-variable/>
, as the following example shows:
<http:outbound-gateway url="http://somehost/%2f/fooApps?bar={param}" encode-uri="false">
<http:uri-variable name="param"
expression="T(org.apache.commons.httpclient.util.URIUtil)
.encodeWithinQuery('Hello World!')"/>
</http:outbound-gateway>
20.4. Configuring HTTP Endpoints with Java
The following example shows how to configure an inbound gateway with Java:
@Bean
public HttpRequestHandlingMessagingGateway inbound() {
HttpRequestHandlingMessagingGateway gateway =
new HttpRequestHandlingMessagingGateway(true);
gateway.setRequestMapping(mapping());
gateway.setRequestPayloadType(String.class);
gateway.setRequestChannelName("httpRequest");
return gateway;
}
@Bean
public RequestMapping mapping() {
RequestMapping requestMapping = new RequestMapping();
requestMapping.setPathPatterns("/foo");
requestMapping.setMethods(HttpMethod.POST);
return requestMapping;
}
The following example shows how to configure an inbound gateway with the Java DSL:
@Bean
public IntegrationFlow inbound() {
return IntegrationFlows.from(Http.inboundGateway("/foo")
.requestMapping(m -> m.methods(HttpMethod.POST))
.requestPayloadType(String.class))
.channel("httpRequest")
.get();
}
The following example shows how to configure an outbound gateway with Java:
@ServiceActivator(inputChannel = "httpOutRequest")
@Bean
public HttpRequestExecutingMessageHandler outbound() {
HttpRequestExecutingMessageHandler handler =
new HttpRequestExecutingMessageHandler("http://localhost:8080/foo");
handler.setHttpMethod(HttpMethod.POST);
handler.setExpectedResponseType(String.class);
return handler;
}
The following example shows how to configure an outbound gateway with the Java DSL:
@Bean
public IntegrationFlow outbound() {
return IntegrationFlows.from("httpOutRequest")
.handle(Http.outboundGateway("http://localhost:8080/foo")
.httpMethod(HttpMethod.POST)
.expectedResponseType(String.class))
.get();
}
20.5. Timeout Handling
In the context of HTTP components, there are two timing areas that have to be considered:
-
Timeouts when interacting with Spring Integration Channels
-
Timeouts when interacting with a remote HTTP server
The components interact with message channels, for which timeouts can be specified. For example, an HTTP Inbound Gateway forwards messages received from connected HTTP Clients to a message channel (which uses a request timeout) and consequently the HTTP Inbound Gateway receives a reply message from the reply channel (which uses a reply timeout) that is used to generate the HTTP Response. The following illustration offers a visual explanation:
For outbound endpoints, we need to consider how timing works while interacting with the remote server. The following image shows this scenario:
You may want to configure the HTTP related timeout behavior, when making active HTTP requests by using the HTTP outbound gateway or the HTTP outbound channel adapter.
In those instances, these two components use Spring’s
RestTemplate
support to execute HTTP requests.
To configure timeouts for the HTTP outbound gateway and the HTTP outbound channel adapter, you can either reference a RestTemplate
bean directly (by using the rest-template
attribute) or you can provide a reference to a ClientHttpRequestFactory
bean (by using the request-factory
attribute).
Spring provides the following implementations of the ClientHttpRequestFactory
interface:
-
SimpleClientHttpRequestFactory
: Uses standard J2SE facilities for making HTTP Requests -
HttpComponentsClientHttpRequestFactory
: Uses Apache HttpComponents HttpClient (since Spring 3.1)
If you do not explicitly configure the request-factory
or rest-template
attribute, a default RestTemplate
(which uses a SimpleClientHttpRequestFactory
) is instantiated.
|
With some JVM implementations, the handling of timeouts by the For example, from the Java™ Platform, Standard Edition 6 API Specification on Some non-standard implementation of this method may ignore the specified timeout. To see the connect timeout set, please call getConnectTimeout(). If you have specific needs, you should test your timeouts.
Consider using the |
| When you use the Apache HttpComponents HttpClient with a pooling connection manager, you should be aware that, by default, the connection manager creates no more than two concurrent connections per given route and no more than 20 connections in total. For many real-world applications, these limits may prove to be too constraining. See the Apache documentation for information about configuring this important component. |
The following example configures an HTTP outbound gateway by using a SimpleClientHttpRequestFactory
that is configured with connect and read timeouts of 5 seconds, respectively:
<int-http:outbound-gateway url="https://www.google.com/ig/api?weather={city}"
http-method="GET"
expected-response-type="java.lang.String"
request-factory="requestFactory"
request-channel="requestChannel"
reply-channel="replyChannel">
<int-http:uri-variable name="city" expression="payload"/>
</int-http:outbound-gateway>
<bean id="requestFactory"
class="org.springframework.http.client.SimpleClientHttpRequestFactory">
<property name="connectTimeout" value="5000"/>
<property name="readTimeout" value="5000"/>
</bean>
HTTP Outbound Gateway
For the HTTP Outbound Gateway, the XML Schema defines only the reply-timeout.
The reply-timeout maps to the sendTimeout property of the org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler class.
More precisely, the property is set on the extended AbstractReplyProducingMessageHandler
class, which ultimately sets the property on the MessagingTemplate
.
The value of the sendTimeout property defaults to "-1" and will be applied to the connected MessageChannel
.
This means, that depending on the implementation, the Message Channel’s send method may block indefinitely.
Furthermore, the sendTimeout property is only used, when the actual MessageChannel implementation has a blocking send (such as 'full' bounded QueueChannel).
20.5.1. HTTP Inbound Gateway
For the HTTP inbound gateway, the XML Schema defines the request-timeout
attribute, which is used to set the requestTimeout
property on the HttpRequestHandlingMessagingGateway
class (on the extended MessagingGatewaySupport
class).
You can also use the reply-timeout
attribute to map to the replyTimeout
property on the same class.
The default for both timeout properties is 1000ms
(one thousand milliseconds or one second).
Ultimately, the request-timeout
property is used to set the sendTimeout
on the MessagingTemplate
instance.
The replyTimeout
property, on the other hand, is used to set the receiveTimeout
property on the MessagingTemplate
instance.
| To simulate connection timeouts, you can connect to a non-routable IP address, such as <IP_ADDRESS>. |
20.6. HTTP Proxy configuration
If you are behind a proxy and need to configure proxy settings for HTTP outbound adapters or gateways, you can apply one of two approaches. In most cases, you can rely on the standard Java system properties that control the proxy settings. Otherwise, you can explicitly configure a Spring bean for the HTTP client request factory instance.
20.6.1. Standard Java Proxy configuration
You can set three system properties to configure the proxy settings that are used by the HTTP protocol handler:
-
http.proxyHost
: The host name of the proxy server. -
http.proxyPort
: The port number (the default is80
). -
http.nonProxyHosts
: A list of hosts that should be reached directly, bypassing the proxy. This is a list of patterns separated by|
. The patterns may start or end with a*
for wildcards. Any host that matches one of these patterns is reached through a direct connection instead of through a proxy.
For HTTPS, the following properties are available:
-
https.proxyHost
: The host name of the proxy server. -
https.proxyPort
: The port number, the default value being 80.
For more information, see download.oracle.com/javase/6/docs/technotes/guides/net/proxies.html
20.6.2. Spring’s SimpleClientHttpRequestFactory
If you need more explicit control over the proxy configuration, you can use Spring’s SimpleClientHttpRequestFactory
and configure its 'proxy' property, as the following example shows:
<bean id="requestFactory"
class="org.springframework.http.client.SimpleClientHttpRequestFactory">
<property name="proxy">
<bean id="proxy" class="java.net.Proxy">
<constructor-arg>
<util:constant static-field="java.net.Proxy.Type.HTTP"/>
</constructor-arg>
<constructor-arg>
<bean class="java.net.InetSocketAddress">
<constructor-arg value="<IP_ADDRESS>"/>
<constructor-arg value="8080"/>
</bean>
</constructor-arg>
</bean>
</property>
</bean>
20.7. HTTP Header Mappings
Spring Integration provides support for HTTP header mapping for both HTTP Request and HTTP Responses.
By default, all standard HTTP headers are mapped from the message to HTTP request or response headers without further configuration.
However, if you do need further customization, you can provide additional configuration by taking advantage of the namespace support.
You can provide a comma-separated list of header names, and you can include simple patterns with the '*' character acting as a wildcard.
Provide such values overrides the default behavior.
Basically, it assumes you are in complete control at that point.
However, if you do want to include all of the standard HTTP headers, you can use the shortcut patterns: HTTP_REQUEST_HEADERS
and HTTP_RESPONSE_HEADERS
.
The following listing shows two examples (the first of which uses a wildcard):
<int-http:outbound-gateway id="httpGateway"
url="http://localhost/test2"
mapped-request-headers="thing1, thing2"
mapped-response-headers="X-*, HTTP_RESPONSE_HEADERS"
channel="someChannel"/>
<int-http:outbound-channel-adapter id="httpAdapter"
url="http://localhost/test2"
mapped-request-headers="thing1, thing2, HTTP_REQUEST_HEADERS"
channel="someChannel"/>
The adapters and gateways use the DefaultHttpHeaderMapper
, which now provides two static factory methods for inbound and outbound adapters so that the proper direction can be applied (mapping HTTP requests and responses either in or out, as appropriate).
If you need further customization, you can also configure a DefaultHttpHeaderMapper
independently and inject it into the adapter through the header-mapper
attribute.
Before version 5.0, the DefaultHttpHeaderMapper
the default prefix for user-defined, non-standard HTTP headers was X-
.
Version 5.0 changed the default prefix to an empty string.
According to RFC-6648, the use of such prefixes is now discouraged.
You can still customize this option by setting the DefaultHttpHeaderMapper.setUserDefinedHeaderPrefix()
property.
The following example configures a header mapper for an HTTP gateway:
<int-http:outbound-gateway id="httpGateway"
url="http://localhost/test2"
header-mapper="headerMapper"
channel="someChannel"/>
<bean id="headerMapper" class="o.s.i.http.support.DefaultHttpHeaderMapper">
<property name="inboundHeaderNames" value="thing1*, *thing2, thing3"/>
<property name="outboundHeaderNames" value="a*b, d"/>
</bean>
If you need to do something other than what the DefaultHttpHeaderMapper
supports, you can implement the HeaderMapper
strategy interface directly and provide a reference to your implementation.
20.8. Integration Graph Controller
Starting with version 4.3, the HTTP module provides an @EnableIntegrationGraphController
configuration class annotation and an <int-http:graph-controller/>
XML element to expose the IntegrationGraphServer
as a REST service.
See Integration Graph for more information.
20.9. HTTP Samples
This section wraps up our coverage of Spring Integration’s HTTP support with a few examples.
20.9.1. Multipart HTTP Request — RestTemplate (Client) and Http Inbound Gateway (Server)
This example shows how simple it is to send a multipart HTTP request with Spring’s RestTemplate
and receive it with a Spring Integration HTTP inbound adapter.
We create a MultiValueMap
and populate it with multipart data.
The RestTemplate
takes care of the rest (no pun intended) by converting it to a MultipartHttpServletRequest
.
This particular client sends a multipart HTTP Request that contains the name of the company and an image file (the company logo).
The following listing shows the example:
RestTemplate template = new RestTemplate();
String uri = "http://localhost:8080/multipart-http/inboundAdapter.htm";
Resource s2logo =
new ClassPathResource("org/springframework/samples/multipart/spring09_logo.png");
MultiValueMap map = new LinkedMultiValueMap();
map.add("company", "SpringSource");
map.add("company-logo", s2logo);
HttpHeaders headers = new HttpHeaders();
headers.setContentType(new MediaType("multipart", "form-data"));
HttpEntity request = new HttpEntity(map, headers);
ResponseEntity<?> httpResponse = template.exchange(uri, HttpMethod.POST, request, null);
That is all we need for the client.
On the server side, we have the following configuration:
<int-http:inbound-channel-adapter id="httpInboundAdapter"
channel="receiveChannel"
path="/inboundAdapter.htm"
supported-methods="GET, POST"/>
<int:channel id="receiveChannel"/>
<int:service-activator input-channel="receiveChannel">
<bean class="org.springframework.integration.samples.multipart.MultipartReceiver"/>
</int:service-activator>
<bean id="multipartResolver"
class="org.springframework.web.multipart.commons.CommonsMultipartResolver"/>
The 'httpInboundAdapter' receives the request and converts it to a Message
with a payload that is a LinkedMultiValueMap
.
We then parse that in the 'multipartReceiver' service-activator, as the following example shows:
public void receive(LinkedMultiValueMap<String, Object> multipartRequest){
System.out.println("### Successfully received multipart request ###");
for (String elementName : multipartRequest.keySet()) {
if (elementName.equals("company")){
System.out.println("\t" + elementName + " - " +
((String[]) multipartRequest.getFirst("company"))[0]);
}
else if (elementName.equals("company-logo")){
System.out.println("\t" + elementName + " - as UploadedMultipartFile: " +
((UploadedMultipartFile) multipartRequest
.getFirst("company-logo")).getOriginalFilename());
}
}
}
You should see the following output:
### Successfully received multipart request ###
company - SpringSource
company-logo - as UploadedMultipartFile: spring09_logo.png
21. JDBC Support
Spring Integration provides channel adapters for receiving and sending messages by using database queries. Through those adapters, Spring Integration supports not only plain JDBC SQL queries but also stored procedure and stored function calls.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jdbc</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-jdbc:5.1.5.RELEASE"
By default, the following JDBC components are available:
The Spring Integration JDBC Module also provides a JDBC Message Store.
21.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 depend on the row-mapping strategy.
The default strategy is a generic mapper that returns a Map
for each row in the query result.
Optionally, you can change this by adding a reference to a RowMapper
instance (see the Spring JDBC documentation for more detailed information about row mapping).
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 a JdbcTemplate
instance or a DataSource
.
As well as the SELECT
statement to generate the messages, the adapter also has an UPDATE
statement that marks the records as processed so that they do not show up in the next poll.
The update can be parameterized by the list of IDs from the original select.
By default, this is done through a naming convention (a column in the input result set called id
is translated into a list in the parameter map for the update called id
).
The following example defines an inbound channel adapter with an update query and a DataSource
reference.
<int-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 (:id)" />
The parameters in the update query are specified with a colon (: ) prefix to the name of a parameter (which, in the preceding example, is an expression to be applied to each of the rows in the polled result set).
This is a standard feature of the named parameter JDBC support in Spring JDBC, combined with a convention (projection onto the polled result list) adopted in Spring Integration.
The underlying Spring JDBC features limit the available expressions (for example, most special characters other than a period are disallowed), but since the target is usually a list of objects (possibly a list of one) that are addressable by bean paths this is not unduly restrictive.
|
To change the parameter generation strategy, you can inject a SqlParameterSourceFactory
into the adapter to override the default behavior (the adapter has a sql-parameter-source-factory
attribute).
Spring Integration provides ExpressionEvaluatingSqlParameterSourceFactory
, which creates a SpEL-based parameter source, with the results of the query as the #root
object.
(If update-per-row
is true, the root object is the row).
If the same parameter name appears multiple times in the update query, it is evaluated only once, and its result is cached.
You can also use a parameter source for the select query. In this case, since there is no “result” object to evaluate against, a single parameter source is used each time (rather than using a parameter source factory). Starting with version 4.0, you can use Spring to create a SpEL based parameter source, as the following example shows:
<int-jdbc:inbound-channel-adapter query="select * from item where status=:status"
channel="target" data-source="dataSource"
select-sql-parameter-source="parameterSource" />
<bean id="parameterSource" factory-bean="parameterSourceFactory"
factory-method="createParameterSourceNoCache">
<constructor-arg value="" />
</bean>
<bean id="parameterSourceFactory"
class="o.s.integration.jdbc.ExpressionEvaluatingSqlParameterSourceFactory">
<property name="parameterExpressions">
<map>
<entry key="status" value="@statusBean.which()" />
</map>
</property>
</bean>
<bean id="statusBean" class="foo.StatusDetermination" />
The value
in each parameter expression can be any valid SpEL expression.
The #root
object for the expression evaluation is the constructor argument defined on the parameterSource
bean.
It is static for all evaluations (in the preceding example, an empty String
).
Starting with version 5.0, you ca supply ExpressionEvaluatingSqlParameterSourceFactory
with sqlParameterTypes
to specify the target SQL type for the particular parameter.
The following example provides SQL types for the parameters being used in the query:
<int-jdbc:inbound-channel-adapter query="select * from item where status=:status"
channel="target" data-source="dataSource"
select-sql-parameter-source="parameterSource" />
<bean id="parameterSource" factory-bean="parameterSourceFactory"
factory-method="createParameterSourceNoCache">
<constructor-arg value="" />
</bean>
<bean id="parameterSourceFactory"
class="o.s.integration.jdbc.ExpressionEvaluatingSqlParameterSourceFactory">
<property name="sqlParameterTypes">
<map>
<entry key="status" value="#{ T(java.sql.Types).BINARY}" />
</map>
</property>
</bean>
Use the createParameterSourceNoCache factory method.
Otherwise, the parameter source caches the result of the evaluation.
Also note that, because caching is disabled, if the same parameter name appears in the select query multiple times, it is re-evaluated for each occurrence.
|
21.1.1. Polling and Transactions
The inbound adapter accepts a regular Spring Integration poller as a child element. Consequently, the frequency of the polling can be controlled (among other uses). An important feature of the poller for JDBC usage is the option to wrap the poll operation in a transaction, as the following example shows:
<int-jdbc:inbound-channel-adapter query="..."
channel="target" data-source="dataSource" update="...">
<int:poller fixed-rate="1000">
<int:transactional/>
</int:poller>
</int-jdbc:inbound-channel-adapter>
| If you do not explicitly specify a poller, a default value is used. As is normal with Spring Integration, it can be defined as a top-level bean). |
In the preceding example, the database is polled every 1000 milliseconds (or once a second), and the update and select queries are both executed in the same transaction. The transaction manager configuration is not shown. However, as long as it is aware of the data source, 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. That way, if any of them fail, the transaction rolls back and the input data is reverted to its original state.
21.1.2. max-rows
Versus max-messages-per-poll
The JDBC inbound channel adapter defines an attribute called max-rows
.
When you specify the adapter’s poller, you can also define a property called max-messages-per-poll
.
While these two attributes look similar, their meaning is quite different.
max-messages-per-poll
specifies the number of times the query is executed per polling interval, whereas max-rows
specifies the number of rows returned for each execution.
Under normal circumstances, you would likely not want to set the poller’s max-messages-per-poll
property when you use the JDBC inbound channel adapter.
Its default value is 1
, which means that the JDBC inbound channel adapter’s receive()
method is executed exactly once for each poll interval.
Setting the max-messages-per-poll
attribute to a larger value means that the query is executed that many times back to back.
For more information regarding the max-messages-per-poll
attribute, see Configuring An Inbound Channel Adapter.
In contrast, the max-rows
attribute, if greater than 0
, specifies the maximum number of rows to be used from the query result set created by the receive()
method.
If the attribute is set to 0
, all rows are included in the resulting message.
The attribute defaults to 0
.
It is recommended to use result set limiting via vendor-specific query options, for example MySQL LIMIT or SQL Server TOP or Oracle’s ROWNUM .
See the particular vendor documentation for more information.
|
21.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. By default, the message payload and headers are available as input parameters to the query, as the following example shows:
<int-jdbc:outbound-channel-adapter
query="insert into foos (id, status, name) values (:headers[id], 0, :payload[something])"
data-source="dataSource"
channel="input"/>
In the preceding example, messages arriving on the channel labelled input
have a payload of a map with a key of something
, so the []
operator dereferences that value from the map.
The headers are also accessed as a map.
The parameters in the preceding query are bean property expressions on the incoming message (not SpEL expressions).
This behavior is part of the SqlParameterSource , which is the default source created by the outbound adapter.
You can inject a different SqlParameterSourceFactory to get different behavior.
|
The outbound adapter requires a reference to either a DataSource
or a JdbcTemplate
.
You can also inject a SqlParameterSourceFactory
to control the binding of each incoming message to a query.
If the input channel is a direct channel, the outbound adapter runs its query in the same thread and, therefore, the same transaction (if there is one) as the sender of the message.
21.2.1. Passing Parameters by Using SpEL Expressions
A common requirement for most JDBC channel adapters is to pass parameters as part of SQL queries or stored procedures or functions.
As mentioned earlier, these parameters are by default bean property expressions, not SpEL expressions.
However, if you need to pass SpEL expression as parameters, you must explicitly inject a SqlParameterSourceFactory
.
The following example uses a ExpressionEvaluatingSqlParameterSourceFactory
to achieve that requirement:
<jdbc:outbound-channel-adapter data-source="dataSource" channel="input"
query="insert into MESSAGES (MESSAGE_ID,PAYLOAD,CREATED_DATE) \
values (:id, :payload, :createdDate)"
sql-parameter-source-factory="spelSource"/>
<bean id="spelSource"
class="o.s.integration.jdbc.ExpressionEvaluatingSqlParameterSourceFactory">
<property name="parameterExpressions">
<map>
<entry key="id" value="headers['id'].toString()"/>
<entry key="createdDate" value="new java.util.Date()"/>
<entry key="payload" value="payload"/>
</map>
</property>
</bean>
For further information, see Defining Parameter Sources.
21.2.2. Using the PreparedStatement
Callback
Sometimes, the flexibility and loose-coupling of SqlParameterSourceFactory
does not do what we need for the target PreparedStatement
or we need to do some low-level JDBC work.
The Spring JDBC module provides APIs to configure the execution environment (such as ConnectionCallback
or PreparedStatementCreator
) and manipulate parameter values (such as SqlParameterSource
).
It can even access APIs for low-level operations, such as StatementCallback
.
Starting with Spring Integration 4.2, MessagePreparedStatementSetter
allows the specification of parameters on the PreparedStatement
manually, in the requestMessage
context.
This class plays exactly the same role as PreparedStatementSetter
in the standard Spring JDBC API.
Actually, it is invoked directly from an inline PreparedStatementSetter
implementation when the JdbcMessageHandler
invokes execute
on the JdbcTemplate
.
This functional interface option is mutually exclusive with sqlParameterSourceFactory
and can be used as a more powerful alternative to populate parameters of the PreparedStatement
from the requestMessage
.
For example, it is useful when we need to store File
data to the DataBase BLOB
column in a streaming manner.
The following example shows how to do so:
@Bean
@ServiceActivator(inputChannel = "storeFileChannel")
public MessageHandler jdbcMessageHandler(DataSource dataSource) {
JdbcMessageHandler jdbcMessageHandler = new JdbcMessageHandler(dataSource,
"INSERT INTO imagedb (image_name, content, description) VALUES (?, ?, ?)");
jdbcMessageHandler.setPreparedStatementSetter((ps, m) -> {
ps.setString(1, m.getHeaders().get(FileHeaders.FILENAME));
try (FileInputStream inputStream = new FileInputStream((File) m.getPayload()); ) {
ps.setBlob(2, inputStream);
}
catch (Exception e) {
throw new MessageHandlingException(m, e);
}
ps.setClob(3, new StringReader(m.getHeaders().get("description", String.class)));
});
return jdbcMessageHandler;
}
From the XML configuration perspective, the prepared-statement-setter
attribute is available on the <int-jdbc:outbound-channel-adapter>
component.
It lets you specify a MessagePreparedStatementSetter
bean reference.
21.2.3. Batch Update
Starting with version 5.1, the JdbcMessageHandler
performs a JdbcOperations.batchUpdate()
if the payload of the request message is an Iterable
instance.
Each element of the Iterable
is wrapped to a Message
with the headers from the request message.
In the case of regular SqlParameterSourceFactory
-based configuration these messages are used to build an SqlParameterSource[]
for an argument used in the mentioned JdbcOperations.batchUpdate()
function.
When a MessagePreparedStatementSetter
configuration is applied, a BatchPreparedStatementSetter
variant is used to iterate over those messages for each item and the provided MessagePreparedStatementSetter
is called against them.
The batch update is not supported when keysGenerated
mode is selected.
21.3. Outbound Gateway
The outbound gateway is like a combination of the outbound and inbound adapters: Its role is to handle a message and use it to execute a SQL query and then respond with the result by sending it to a reply channel. By default, the message payload and headers are available as input parameters to the query, as the following example shows:
<int-jdbc:outbound-gateway
update="insert into mythings (id, status, name) values (:headers[id], 0, :payload[thing])"
request-channel="input" reply-channel="output" data-source="dataSource" />
The result of the preceding example is to insert a record into the mythings
table and return a message that indicates the number of rows affected (the payload is a map: {UPDATED=1}
) to the output channel .
If the update query is an insert with auto-generated keys, you can populate the reply message with the generated keys by adding keys-generated="true"
to the preceding example (this is not the default because it is not supported by some database platforms).
The following example shows the changed configuration:
<int-jdbc:outbound-gateway
update="insert into mythings (status, name) values (0, :payload[thing])"
request-channel="input" reply-channel="output" data-source="dataSource"
keys-generated="true"/>
Instead of the update count or the generated keys, you can also provide a select query to execute and generate a reply message from the result (such as the inbound adapter), as the following example shows:
<int-jdbc:outbound-gateway
update="insert into foos (id, status, name) values (:headers[id], 0, :payload[foo])"
query="select * from foos where id=:headers[$id]"
request-channel="input" reply-channel="output" data-source="dataSource"/>
Since Spring Integration 2.2, the update SQL query is no longer mandatory.
You can now provide only a select query, by using either the query
attribute or the query
element.
This is extremely useful if you need to actively retrieve data by using, for example, a generic gateway or a payload enricher.
The reply message is then generated from the result (similar to how the inbound adapter works) and passed to the reply channel.
The following example show to use the query
attribute:
<int-jdbc:outbound-gateway
query="select * from foos where id=:headers[id]"
request-channel="input"
reply-channel="output"
data-source="dataSource"/>
|
By default, the component for the |
As with the channel adapters, you can also provide SqlParameterSourceFactory
instances for request and reply.
The default is the same as for the outbound adapter, so the request message is available as the root of an expression.
If keys-generated="true"
, the root of the expression is the generated keys (a map if there is only one or a list of maps if multi-valued).
The outbound gateway requires a reference to either a DataSource
or a JdbcTemplate
.
It can also have a SqlParameterSourceFactory
injected to control the binding of the incoming message to the query.
Starting with the version 4.2, the request-prepared-statement-setter
attribute is available on the <int-jdbc:outbound-gateway>
as an alternative to request-sql-parameter-source-factory
.
It lets you specify a MessagePreparedStatementSetter
bean reference, which implements more sophisticated PreparedStatement
preparation before its execution.
See Outbound Channel Adapter for more information about MessagePreparedStatementSetter
.
21.4. JDBC Message Store
Spring Integration provides two JDBC specific message store implementations.
The JdbcMessageStore
is suitable for use with aggregators and the claim check pattern.
The JdbcChannelMessageStore
implementation provides a more targeted and scalable implementation specifically for message channel.
Note that you can use a JdbcMessageStore
to back a message channel, JdbcChannelMessageStore
is optimized for that purpose.
Starting with versions 5.0.11, 5.1.2, the indexes for the JdbcChannelMessageStore have been optimized.
If you have large message groups in such a store, you may wish to alter the indexes.
Furthermore, the index for PriorityChannel is commented out because it is not needed unless you are using such channels backed by JDBC.
|
When using the OracleChannelMessageStoreQueryProvider , the priority channel index must be added because it is included in a hint in the query.
|
21.4.1. Initializing the Database
Before starting to use JDBC message store components, you should provision a target database with the appropriate objects.
Spring Integration ships with some sample scripts that can be used to initialize a database.
In the spring-integration-jdbc
JAR file, you can find scripts in the org.springframework.integration.jdbc
package.
It provides an example create and an example drop script 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 and as specifications of the the required table and column names.
You may find that you need to enhance them for production use (for, example, by adding index declarations).
21.4.2. The Generic JDBC 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 such as an aggregator) backed by a database.
Both interfaces are implemented by the JdbcMessageStore
, and there is support for configuring store instances in XML, as the following example shows:
<int-jdbc:message-store id="messageStore" data-source="dataSource"/>
You can specify a JdbcTemplate
instead of a DataSource
.
The following example shows some other optional attributes:
<int-jdbc:message-store id="messageStore" data-source="dataSource"
lob-handler="lobHandler" table-prefix="MY_INT_"/>
In the preceding example, we have specified a LobHandler
for dealing with messages as large objects (which is often necessary for Oracle) and a prefix for the table names in the queries generated by the store.
The table name prefix defaults to INT_
.
21.4.3. Backing Message Channels
If you intend to backing message channels with JDBC, we recommend using the JdbcChannelMessageStore
implementation.
It works only in conjunction with Message Channels.
Supported Databases
The JdbcChannelMessageStore
uses database-specific SQL queries to retrieve messages from the database.
Therefore, you must set the ChannelMessageStoreQueryProvider
property on the JdbcChannelMessageStore
.
This channelMessageStoreQueryProvider
provides the SQL queries for the particular database you specify.
Spring Integration provides support for the following relational databases:
-
PostgreSQL
-
HSQLDB
-
MySQL
-
Oracle
-
Derby
-
H2
-
SqlServer
-
Sybase
-
DB2
If your database is not listed, you can extend the AbstractChannelMessageStoreQueryProvider
class and provide your own custom queries.
Version 4.0 added the MESSAGE_SEQUENCE
column to the table to ensure first-in-first-out (FIFO) queueing even when messages are stored in the same millisecond.
Custom Message Insertion
Since version 5.0, by overloading the ChannelMessageStorePreparedStatementSetter
class, you can provide a custom implementation for message insertion in the JdbcChannelMessageStore
.
You can use it to set different columns or change the table structure or serialization strategy.
For example, instead of default serialization to byte[]
, you can store its structure as a JSON string.
The following example uses the default implementation of setValues
to store common columns and overrides the behavior to store the message payload as a varchar
:
public class JsonPreparedStatementSetter extends ChannelMessageStorePreparedStatementSetter {
public JsonPreparedStatementSetter() {
super();
}
@Override
public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage,
Object groupId, String region, boolean priorityEnabled) throws SQLException {
// Populate common columns
super.setValues(preparedStatement, requestMessage, groupId, region, priorityEnabled);
// Store message payload as varchar
preparedStatement.setString(6, requestMessage.getPayload().toString());
}
}
|
Generally, we do not recommend using a relational database for queuing. Instead, if possible, consider using either JMS- or AMQP-backed channels instead. For further reference, see the following resources: |
Concurrent Polling
When polling a message channel, you have the option to configure the associated Poller
with a TaskExecutor
reference.
|
Keep in mind, though, that if you use a JDBC backed message channel and you plan to poll the channel and consequently the message store transactionally with multiple threads, you should ensure that you use a relational database that supports Multiversion Concurrency Control (MVCC). Otherwise, locking may be an issue and the performance, when using multiple threads, may not materialize as expected. For example, Apache Derby is problematic in that regard. To achieve better JDBC queue throughput and avoid issues when different threads may poll the same
|
Priority Channel
Starting with version 4.0, JdbcChannelMessageStore
implements PriorityCapableChannelMessageStore
and provides the priorityEnabled
option, letting it be used as a message-store
reference for priority-queue
instances.
For this purpose, the INT_CHANNEL_MESSAGE
table has a MESSAGE_PRIORITY
column to store the value of PRIORITY
message headers.
In addition, a new MESSAGE_SEQUENCE
column lets us achieve a robust first-in-first-out (FIFO) polling mechanism, even when multiple messages are stored with the same priority in the same millisecond.
Messages are polled (selected) from the database with order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE
.
We do not recommend using the same JdbcChannelMessageStore bean for priority and non-priority queue channels, because the priorityEnabled option applies to the entire store and proper FIFO queue semantics are not retained for the queue channel.
However, the same INT_CHANNEL_MESSAGE table (and even region ) can be used for both JdbcChannelMessageStore types.
To configure that scenario, you can extend one message store bean from the other, as the following example shows:
|
<bean id="channelStore" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
<property name="dataSource" ref="dataSource"/>
<property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
</bean>
<int:channel id="queueChannel">
<int:queue message-store="channelStore"/>
</int:channel>
<bean id="priorityStore" parent="channelStore">
<property name="priorityEnabled" value="true"/>
</bean>
<int:channel id="priorityChannel">
<int:priority-queue message-store="priorityStore"/>
</int:channel>
21.4.4. 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 protection against name clashes and to give control over the database meta-data configuration, the message store lets the tables be partitioned in two ways.
One way is to use separate table names, by changing the prefix (as described earlier).
The other way is to specify a region
name for partitioning data within a single table.
An important use case for the second approach is when the MessageStore
is managing persistent queues that back a Spring Integration Message Channel.
The message data for a persistent channel is keyed in the store on the channel name.
Consequently, if the channel names are not globally unique, the channels can pick up data that is not intended for them.
To avoid this danger, you can use the message store region
to keep data separate for different physical channels that have the same logical name.
21.5. Stored Procedures
In certain situations, plain JDBC support is not sufficient. Maybe you deal with legacy relational database schemas or you have complex data processing needs, but, ultimately, you have to use stored procedures or stored functions. Since Spring Integration 2.1, we provide three components to execute stored procedures or stored functions:
-
Stored Procedures Inbound Channel Adapter
-
Stored Procedures Outbound Channel Adapter
-
Stored Procedures Outbound Gateway
21.5.1. Supported Databases
In order to enable calls to stored procedures and stored functions, the stored procedure components use the org.springframework.jdbc.core.simple.SimpleJdbcCall
class.
Consequently, the following databases are fully supported for executing stored procedures:
-
Apache Derby
-
DB2
-
MySQL
-
Microsoft SQL Server
-
Oracle
-
PostgreSQL
-
Sybase
If you want to execute stored functions instead, the following databases are fully supported:
-
MySQL
-
Microsoft SQL Server
-
Oracle
-
PostgreSQL
|
Even though your particular database may not be fully supported, chances are that you can use the stored procedure Spring Integration components quite successfully anyway, provided your RDBMS supports stored procedures or stored functions. As a matter of fact, some of the provided integration tests use the H2 database. Nevertheless, it is very important to thoroughly test those usage scenarios. |
21.5.2. Configuration
The stored procedure components provide full XML Namespace support, and configuring the components is similar as for the general purpose JDBC components discussed earlier.
21.5.3. Common Configuration Attributes
All stored procedure components share certain configuration parameters:
-
auto-startup
: Lifecycle attribute signaling whether this component should be started during application context startup. It defaults totrue
. Optional. -
data-source
: Reference to ajavax.sql.DataSource
, which is used to access the database. Required. -
id
: Identifies the underlying Spring bean definition, which is an instance of eitherEventDrivenConsumer
orPollingConsumer
, depending on whether the outbound channel adapter’schannel
attribute references aSubscribableChannel
or aPollableChannel
. Optional. -
ignore-column-meta-data
: For fully supported databases, the underlyingSimpleJdbcCall
class can automatically retrieve the parameter information for the stored procedure or stored function from the JDBC metadata.However, if the database does not support metadata lookups or if you need to provide customized parameter definitions, this flag can be set to
true
. It defaults tofalse
. Optional. -
is-function
: Iftrue
, a SQL Function is called. In that case, thestored-procedure-name
orstored-procedure-name-expression
attributes define the name of the called function. It defaults tofalse
. Optional. -
stored-procedure-name
: This attribute specifies the name of the stored procedure. If theis-function
attribute is set totrue
, this attribute specifies the function name instead. Either this property orstored-procedure-name-expression
must be specified. -
stored-procedure-name-expression
: This attribute specifies the name of the stored procedure by using a SpEL expression. By using SpEL, you have access to the full message (if available), including its headers and payload. You can use this attribute to invoke different stored procedures at runtime. For example, you can provide stored procedure names that you would like to execute as a message header. The expression must resolve to aString
.If the
is-function
attribute is set totrue
, this attribute specifies a stored function. Either this property orstored-procedure-name
must be specified. -
jdbc-call-operations-cache-size
: Defines the maximum number of cachedSimpleJdbcCallOperations
instances. Basically, for each stored procedure name, a newSimpleJdbcCallOperations
instance is created that, in return, is cached.Spring Integration 2.2 added the stored-procedure-name-expression
attribute and thejdbc-call-operations-cache-size
attribute.The default cache size is
10
. A value of0
disables caching. Negative values are not permitted.If you enable JMX, statistical information about the
jdbc-call-operations-cache
is exposed as an MBean. See MBean Exporter for more information. -
sql-parameter-source-factory
: (Not available for the stored procedure inbound channel adapter.) Reference to aSqlParameterSourceFactory
. By default, bean properties of the passed inMessage
payload are used as a source for the stored procedure’s input parameters by using aBeanPropertySqlParameterSourceFactory
.This may suffice for basic use cases. For more sophisticated options, consider passing in one or more
ProcedureParameter
values. See Defining Parameter Sources. Optional. -
use-payload-as-parameter-source
: (Not available for the stored procedure inbound channel adapter.) If set totrue
, the payload of theMessage
is used as a source for providing parameters. If set tofalse
, however, the entireMessage
is available as a source for parameters.If no procedure parameters are passed in, this property defaults to
true
. This means that, by using a defaultBeanPropertySqlParameterSourceFactory
, the bean properties of the payload are used as a source for parameter values for the stored procedure or stored function.However, if procedure parameters are passed in, this property (by default) evaluates to
false
.ProcedureParameter
lets SpEL Expressions be provided. Therefore, it is highly beneficial to have access to the entireMessage
. The property is set on the underlyingStoredProcExecutor
. Optional.
21.5.4. Common Configuration Sub-Elements
The stored procedure components share a common set of child elements that you can use to define and pass parameters to stored procedures or stored functions. The following elements are available:
-
parameter
-
returning-resultset
-
sql-parameter-definition
-
poller
-
parameter
: Provides a mechanism to provide stored procedure parameters. Parameters can be either static or provided by using a SpEL Expressions.<int-jdbc:parameter name="" (1) type="" (2) value=""/> (3) <int-jdbc:parameter name="" expression=""/> (4)
+ <1> The name of the parameter to be passed into the Stored Procedure or Stored Function. Required. <2> This attribute specifies the type of the value. If nothing is provided, this attribute defaults to
java.lang.String
. This attribute is used only when thevalue
attribute is used. Optional. <3> The value of the parameter. You must provide either this attribute or theexpression
attribute. Optional. <4> Instead of thevalue
attribute, you can specify a SpEL expression for passing the value of the parameter. If you specify theexpression
, thevalue
attribute is not allowed. Optional.Optional.
-
returning-resultset
: Stored procedures may return multiple result sets. By setting one or morereturning-resultset
elements, you can specifyRowMappers
to convert each returnedResultSet
to meaningful objects. Optional.<int-jdbc:returning-resultset name="" row-mapper="" />
-
sql-parameter-definition
: If you use a database that is fully supported, you typically do not have to specify the stored procedure parameter definitions. Instead, those parameters can be automatically derived from the JDBC metadata. However, if you use databases that are not fully supported, you must set those parameters explicitly by using thesql-parameter-definition
element.You can also choose to turn off any processing of parameter metadata information obtained through JDBC by using the
ignore-column-meta-data
attribute.<int-jdbc:sql-parameter-definition name="" (1) direction="IN" (2) type="STRING" (3) scale="5" (4) type-name="FOO_STRUCT" (5) return-type="fooSqlReturnType"/> (6)
1 Specifies the name of the SQL parameter. Required. 2 Specifies the direction of the SQL parameter definition. Defaults to IN
. Valid values are:IN
,OUT
, andINOUT
. If your procedure is returning result sets, use thereturning-resultset
element. Optional.3 The SQL type used for this SQL parameter definition. Translates into an integer value, as defined by java.sql.Types
. Alternatively, you can provide the integer value as well. If this attribute is not explicitly set, it defaults to 'VARCHAR'. Optional.4 The scale of the SQL parameter. Only used for numeric and decimal parameters. Optional. 5 The typeName
for types that are user-named, such as:STRUCT
,DISTINCT
,JAVA_OBJECT
, and named array types. This attribute is mutually exclusive with thescale
attribute. Optional.6 The reference to a custom value handler for complex types. An implementation of SqlReturnType
. This attribute is mutually exclusive with thescale
attribute and is only applicable for OUT and INOUT parameters. Optional. -
poller
: Lets you configure a message poller if this endpoint is aPollingConsumer
. Optional.
21.5.5. Defining Parameter Sources
Parameter sources govern the techniques of retrieving and mapping the Spring Integration message properties to the relevant stored procedure input parameters.
The stored procedure components follow certain rules.
By default, the bean properties of the Message
payload are used as a source for the stored procedure’s input parameters.
In that case, a BeanPropertySqlParameterSourceFactory
is used.
This may suffice for basic use cases.
The next example illustrates that default behavior.
For the “automatic” lookup of bean properties by using the BeanPropertySqlParameterSourceFactory to work, your bean properties must be defined in lower case.
This is due to the fact that in org.springframework.jdbc.core.metadata.CallMetaDataContext (the Java method is matchInParameterValuesWithCallParameters() ), the retrieved stored procedure parameter declarations are converted to lower case.
As a result, if you have camel-case bean properties (such as lastName ), the lookup fails.
In that case, provide an explicit ProcedureParameter .
|
Suppose we have a payload that consists of a simple bean with the following three properties: id
, name
, and description
.
Furthermore, we have a simplistic Stored Procedure called INSERT_COFFEE
that accepts three input parameters: id
, name
, and description
.
We also use a fully supported database.
In that case, the following configuration for a stored procedure outbound adapter suffices:
<int-jdbc:stored-proc-outbound-channel-adapter data-source="dataSource"
channel="insertCoffeeProcedureRequestChannel"
stored-procedure-name="INSERT_COFFEE"/>
For more sophisticated options, consider passing in one or more ProcedureParameter
values.
If you do provide ProcedureParameter
values explicitly, by default, an ExpressionEvaluatingSqlParameterSourceFactory
is used for parameter processing, to enable the full power of SpEL expressions.
If you need even more control over how parameters are retrieved, consider passing in a custom implementation of SqlParameterSourceFactory
by using the sql-parameter-source-factory
attribute.
21.5.6. Stored Procedure Inbound Channel Adapter
The following listing calls out the attributes that matter for a stored procedure inbound channel adapter:
<int-jdbc:stored-proc-inbound-channel-adapter
channel="" (1)
stored-procedure-name=""
data-source=""
auto-startup="true"
id=""
ignore-column-meta-data="false"
is-function="false"
skip-undeclared-results="" (2)
return-value-required="false" (3)
<int:poller/>
<int-jdbc:sql-parameter-definition name="" direction="IN"
type="STRING"
scale=""/>
<int-jdbc:parameter name="" type="" value=""/>
<int-jdbc:parameter name="" expression=""/>
<int-jdbc:returning-resultset name="" row-mapper="" />
</int-jdbc:stored-proc-inbound-channel-adapter>
| 1 | Channel to which polled messages are sent.
If the stored procedure or function does not return any data, the payload of the Message is null.
Required. |
| 2 | If this attribute is set to true , all results from a stored procedure call that do not have a corresponding SqlOutParameter declaration are bypassed.
For example, stored procedures can return an update count value, even though your stored procedure declared only a single result parameter.
The exact behavior depends on the database implementation.
The value is set on the underlying JdbcTemplate .
The value defaults to true .
Optional. |
| 3 | Indicates whether this procedure’s return value should be included. Since Spring Integration 3.0. Optional. |
21.5.7. Stored Procedure Outbound Channel Adapter
The following listing calls out the attributes that matter for a stored procedure outbound channel adapter:
<int-jdbc:stored-proc-outbound-channel-adapter channel="" (1)
stored-procedure-name=""
data-source=""
auto-startup="true"
id=""
ignore-column-meta-data="false"
order="" (2)
sql-parameter-source-factory=""
use-payload-as-parameter-source="">
<int:poller fixed-rate=""/>
<int-jdbc:sql-parameter-definition name=""/>
<int-jdbc:parameter name=""/>
</int-jdbc:stored-proc-outbound-channel-adapter>
| 1 | The receiving message channel of this endpoint. Required. |
| 2 | Specifies the order for invocation when this endpoint is connected as a subscriber to a channel.
This is particularly relevant when that channel is using a failover dispatching strategy.
It has no effect when this endpoint is itself a polling consumer for a channel with a queue.
Optional. |
21.5.8. Stored Procedure Outbound Gateway
The following listing calls out the attributes that matter for a stored procedure outbound channel adapter:
<int-jdbc:stored-proc-outbound-gateway request-channel="" (1)
stored-procedure-name=""
data-source=""
auto-startup="true"
id=""
ignore-column-meta-data="false"
is-function="false"
order=""
reply-channel="" (2)
reply-timeout="" (3)
return-value-required="false" (4)
skip-undeclared-results="" (5)
sql-parameter-source-factory=""
use-payload-as-parameter-source="">
<int-jdbc:sql-parameter-definition name="" direction="IN"
type=""
scale="10"/>
<int-jdbc:sql-parameter-definition name=""/>
<int-jdbc:parameter name="" type="" value=""/>
<int-jdbc:parameter name="" expression=""/>
<int-jdbc:returning-resultset name="" row-mapper="" />
| 1 | The receiving message channel of this endpoint. Required. |
| 2 | Message channel to which replies should be sent after receiving the database response. Optional. |
| 3 | Lets you specify how long this gateway waits for the reply message to be sent successfully before throwing an exception.
Keep in mind that, when sending to a DirectChannel , the invocation occurs in the sender’s thread.
Consequently, the failing of the send operation may be caused by other components further downstream.
By default, the gateway waits indefinitely.
The value is specified in milliseconds.
Optional. |
| 4 | Indicates whether this procedure’s return value should be included. Optional. |
| 5 | If the skip-undeclared-results attribute is set to true , all results from a stored procedure call that do not have a corresponding SqlOutParameter declaration are bypassed.
For example, stored procedures may return an update count value, even though your stored procedure only declared a single result parameter.
The exact behavior depends on the database.
The value is set on the underlying JdbcTemplate .
The value defaults to true .
Optional. |
21.5.9. Examples
This section contains two examples that call Apache Derby stored procedures.
The first procedure calls a stored procedure that returns a ResultSet
.
By using a RowMapper
, the data is converted into a domain object, which then becomes the Spring Integration message payload.
In the second sample, we call a stored procedure that uses output parameters to return data instead.
|
Have a look at the Spring Integration Samples project. The project contains the Apache Derby example referenced here, as well as instructions on how to run it. The Spring Integration Samples project also provides an example of using Oracle stored procedures. |
In the first example, we call a stored procedure named FIND_ALL_COFFEE_BEVERAGES
that does not define any input parameters but that returns a ResultSet
.
In Apache Derby, stored procedures are implemented in Java. The following listing shows the method signature:
public static void findAllCoffeeBeverages(ResultSet[] coffeeBeverages)
throws SQLException {
...
}
The following listing shows the corresponding SQL:
CREATE PROCEDURE FIND_ALL_COFFEE_BEVERAGES() \
PARAMETER STYLE JAVA LANGUAGE JAVA MODIFIES SQL DATA DYNAMIC RESULT SETS 1 \
EXTERNAL NAME 'o.s.i.jdbc.storedproc.derby.DerbyStoredProcedures.findAllCoffeeBeverages';
In Spring Integration, you can now call this stored procedure by using, for example, a stored-proc-outbound-gateway
, as the following example shows:
<int-jdbc:stored-proc-outbound-gateway id="outbound-gateway-storedproc-find-all"
data-source="dataSource"
request-channel="findAllProcedureRequestChannel"
expect-single-result="true"
stored-procedure-name="FIND_ALL_COFFEE_BEVERAGES">
<int-jdbc:returning-resultset name="coffeeBeverages"
row-mapper="org.springframework.integration.support.CoffeBeverageMapper"/>
</int-jdbc:stored-proc-outbound-gateway>
In the second example, we call a stored procedure named FIND_COFFEE
that has one input parameter.
Instead of returning a ResultSet
, it uses an output parameter.
The following example shows the method signature:
public static void findCoffee(int coffeeId, String[] coffeeDescription)
throws SQLException {
...
}
The following listing shows the corresponding SQL:
CREATE PROCEDURE FIND_COFFEE(IN ID INTEGER, OUT COFFEE_DESCRIPTION VARCHAR(200)) \
PARAMETER STYLE JAVA LANGUAGE JAVA EXTERNAL NAME \
'org.springframework.integration.jdbc.storedproc.derby.DerbyStoredProcedures.findCoffee';
In Spring Integration, you can now call this Stored Procedure by using, for example, a stored-proc-outbound-gateway
, as the following example shows:
<int-jdbc:stored-proc-outbound-gateway id="outbound-gateway-storedproc-find-coffee"
data-source="dataSource"
request-channel="findCoffeeProcedureRequestChannel"
skip-undeclared-results="true"
stored-procedure-name="FIND_COFFEE"
expect-single-result="true">
<int-jdbc:parameter name="ID" expression="payload" />
</int-jdbc:stored-proc-outbound-gateway>
21.6. JDBC Lock Registry
Version 4.3 introduced the JdbcLockRegistry
.
Certain components (for example, aggregator and resequencer) use a lock obtained from a LockRegistry
instance to ensure that only one thread manipulates a group at a time.
The DefaultLockRegistry
performs this function within a single component.
You can now configure an external lock registry on these components.
When used with a shared MessageGroupStore
, you can use the JdbcLockRegistry
to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.
When a lock is released by a local thread, another local thread can generally acquire the lock immediately. If a lock is released by a thread that uses a different registry instance, it can take up to 100ms to acquire the lock.
The JdbcLockRegistry
is based on the LockRepository
abstraction, which has a DefaultLockRepository
implementation.
The database schema scripts are located in the org.springframework.integration.jdbc
package, which is divided for the particular RDBMS vendors.
For example, the following listing shows the H2 DDL for the lock table:
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);
The INT_
can be changed according to the target database design requirements.
Therefore, you must use prefix
property on the DefaultLockRepository
bean definition.
Sometimes, one application has moved to such a state that it cannot release the distributed lock and remove the particular record in the database.
For this purpose, such dead locks can be expired by the other application on the next locking invocation.
The timeToLive
(TTL) option on the DefaultLockRepository
is provided for this purpose.
You may also want to specify CLIENT_ID
for the locks stored for a given DefaultLockRepository
instance.
If so, you can specify the id
to be associated with the DefaultLockRepository
as a constructor parameter.
21.7. JDBC Metadata Store
Version 5.0 introduced the JDBC MetadataStore
(see Metadata Store) implementation.
You can use the JdbcMetadataStore
to maintain the metadata state across application restarts.
This MetadataStore
implementation can be used with adapters such as the following:
To configure these adapters to use the JdbcMetadataStore
, declare a Spring bean by using a bean name of metadataStore
.
The Feed inbound channel adapter and the feed inbound channel adapter both automatically pick up and use the declared JdbcMetadataStore
, as the following example shows:
@Bean
public MetadataStore metadataStore(DataSource dataSource) {
return new JdbcMetadataStore(dataSource);
}
The org.springframework.integration.jdbc
package has Database schema scripts for several RDMBS vendors.
For example, the following listing shows the H2 DDL for the metadata table:
CREATE TABLE INT_METADATA_STORE (
METADATA_KEY VARCHAR(255) NOT NULL,
METADATA_VALUE VARCHAR(4000),
REGION VARCHAR(100) NOT NULL,
constraint METADATA_STORE primary key (METADATA_KEY, REGION)
);
You can change the INT_
prefix to match the target database design requirements.
You can also configure JdbcMetadataStore
to use the custom prefix.
The JdbcMetadataStore
implements ConcurrentMetadataStore
, letting it be reliably shared across multiple application instances, where only one instance can store or modify a key’s value.
All of these operations are atomic, thanks to transaction guarantees.
Transaction management must use JdbcMetadataStore
.
Inbound channel adapters can be supplied with a reference to the TransactionManager
in the poller configuration.
Unlike non-transactional MetadataStore
implementations, with JdbcMetadataStore
, the entry appears in the target table only after the transaction commits.
When a rollback occurs, no entries are added to the INT_METADATA_STORE
table.
Since version 5.0.7, you can configure the JdbcMetadataStore
with the RDBMS vendor-specific lockHint
option for lock-based queries on metadata store entries.
By default, it is FOR UPDATE
and can be configured with an empty string if the target database does not support row locking functionality.
Consult with your vendor for particular and possible hints in the SELECT
expression for locking rows before updates.
22. JPA Support
Spring Integration’s JPA (Java Persistence API) module provides components for performing various database operations using JPA.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jpa</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-jpa:5.1.5.RELEASE"
The JPA API must be included via some vendor-specific implementation, e.g. Hibernate ORM Framework.
The following components are provided:
These components can be used to perform select
, create
, update
, and delete
operations on the target databases by sending and receiving messages to them.
The JPA inbound channel adapter lets you poll and retrieve (select
) data from the database by using JPA, whereas the JPA outbound channel adapter lets you create, update, and delete entities.
You can use outbound gateways for JPA to persist entities to the database, letting you continue the flow and execute further components downstream. Similarly, you can use an outbound gateway to retrieve entities from the database.
For example, you may use the outbound gateway, which receives a Message
with a userId
as payload on its request channel, to query the database, retrieve the user entity, and pass it downstream for further processing.
Recognizing these semantic differences, Spring Integration provides two separate JPA outbound gateways:
-
Retrieving outbound gateway
-
Updating outbound gateway
22.1. Functionality
All JPA components perform their respective JPA operations by using one of the following:
-
Entity classes
-
Java Persistence Query Language (JPQL) for update, select and delete (JPQL does not support inserts)
-
Native Query
-
Named Query
The following sections describe each of these components in more detail.
22.2. Supported Persistence Providers
The Spring Integration JPA support has been tested against the following persistence providers:
-
Hibernate
-
EclipseLink
When using a persistence provider, you should ensure that the provider is compatible with JPA 2.1.
22.3. Java Implementation
Each of the provided components uses the o.s.i.jpa.core.JpaExecutor
class, which, in turn, uses an implementation of the o.s.i.jpa.core.JpaOperations
interface.
JpaOperations
operates like a typical Data Access Object (DAO) and provides methods such as find, persist, executeUpdate, and so on.
For most use cases, the default implementation (o.s.i.jpa.core.DefaultJpaOperations
) should be sufficient.
However, you can specify your own implementation if you require custom behavior.
To initialize a JpaExecutor
, you must use one of the constructors that accept one of:
-
EntityManagerFactory
-
EntityManager
-
JpaOperations
The following example shows how to initialize a JpaExecutor
with an entityManagerFactory
and use it in an outbound gateway:
@Bean
public JpaExecutor jpaExecutor() {
JpaExecutor executor = new JpaExecutor(this.entityManagerFactory);
executor.setJpaParameters(Collections.singletonList(new JpaParameter("firstName", null, "#this")));
executor.setUsePayloadAsParameterSource(true);
executor.setExpectSingleResult(true);
return executor;
}
@ServiceActivator(inputChannel = "getEntityChannel")
@Bean
public MessageHandler retrievingJpaGateway() {
JpaOutboundGateway gateway = new JpaOutboundGateway(jpaExecutor());
gateway.setGatewayType(OutboundGatewayType.RETRIEVING);
gateway.setOutputChannelName("resultsChannel");
return gateway;
}
22.4. Namespace Support
When using XML namespace support, the underlying parser classes instantiate the relevant Java classes for you. Thus, you typically need not deal with the inner workings of the JPA adapter. This section documents the XML namespace support provided by Spring Integration and shows you how to use the XML Namespace support to configure the JPA components.
22.4.1. Common XML Namespace Configuration Attributes
Certain configuration parameters are shared by all JPA components:
auto-startup
-
Lifecycle attribute that signals whether this component should be started during application context startup. Defaults to
true
. Optional. id
-
Identifies the underlying Spring bean definition, which is an instance of either
EventDrivenConsumer
orPollingConsumer
. Optional. entity-manager-factory
-
The reference to the JPA entity manager factory that the adapter uses to create the
EntityManager
. You must provide this attribute, theentity-manager
attribute, or thejpa-operations
attribute. entity-manager
-
The reference to the JPA Entity Manager that the component uses. You must provide this attribute, the
entity-manager-factory
attribute, or thejpa-operations
attribute.Usually, your Spring application context defines only a JPA entity manager factory, and the EntityManager
is injected by using the@PersistenceContext
annotation. This approach does not apply for the Spring Integration JPA components. Usually, injecting the JPA entity manager factory is best, but, when you want to inject anEntityManager
explicitly, you have to define aSharedEntityManagerBean
. For more information, see the relevant Javadoc.The following example shows how to explicitly include an entity manager factory:
<bean id="entityManager" class="org.springframework.orm.jpa.support.SharedEntityManagerBean"> <property name="entityManagerFactory" ref="entityManagerFactoryBean" /> </bean>
jpa-operations
-
A reference to a bean that implements the
JpaOperations
interface. In rare cases, it might be advisable to provide your own implementation of theJpaOperations
interface instead of relying on the default implementation (org.springframework.integration.jpa.core.DefaultJpaOperations
). If you use thejpa-operations
attribute, you must not provide the JPA entity manager or JPA entity manager factory, becauseJpaOperations
wraps the necessary datasource. entity-class
-
The fully qualified name of the entity class. The exact semantics of this attribute vary, depending on whether we are performing a persist or update operation or whether we are retrieving objects from the database.
When retrieving data, you can specify the
entity-class
attribute to indicate that you would like to retrieve objects of this type from the database. In that case, you must not define any of the query attributes (jpa-query
,native-query
, ornamed-query
).When persisting data, the
entity-class
attribute indicates the type of object to persist. If not specified (for persist operations) the entity class is automatically retrieved from the message’s payload. jpa-query
-
Defines the JPA query (Java Persistence Query Language) to be used.
native-query
-
Defines the native SQL query to be used.
named-query
-
Refers to a named query. A named query can be defined in either Native SQL or JPAQL, but the underlying JPA persistence provider handles that distinction internally.
22.4.2. Providing JPA Query Parameters
To provide parameters, you can use the parameter
XML element.
It has a mechanism that lets you provide parameters for queries that are based on either the Java Persistence Query Language (JPQL) or native SQL queries.
You can also provide parameters for named queries.
- Expression-based Parameters
-
The following example shows how to set an expression-based parameter:
<int-jpa:parameter expression="payload.name" name="firstName"/>
- Value-based Parameters
-
The following example shows how to set an value-based parameter:
<int-jpa:parameter name="name" type="java.lang.String" value="myName"/>
- Positional Parameters
-
The following example shows how to set an expression-based parameter:
<int-jpa:parameter expression="payload.name"/> <int-jpa:parameter type="java.lang.Integer" value="21"/>
22.4.3. Transaction Handling
All JPA operations (such as INSERT
, UPDATE
, and DELETE
) require a transaction to be active whenever they are performed.
For inbound channel adapters, you need do nothing special.
It works similarly to the way we configure transaction managers with pollers that are used with other inbound channel adapters.
The following XML example configures a transaction manager that uses a poller with an inbound channel adapter:
<int-jpa:inbound-channel-adapter
channel="inboundChannelAdapterOne"
entity-manager="em"
auto-startup="true"
jpa-query="select s from Student s"
expect-single-result="true"
delete-after-poll="true">
<int:poller fixed-rate="2000" >
<int:transactional propagation="REQUIRED"
transaction-manager="transactionManager"/>
</int:poller>
</int-jpa:inbound-channel-adapter>
However, you may need to specifically start a transaction when using an outbound channel adapter or gateway.
If a DirectChannel
is an input channel for the outbound adapter or gateway and if the transaction is active in the current thread of execution, the JPA operation is performed in the same transaction context.
You can also configure this JPA operation to run as a new transaction, as the following example shows:
<int-jpa:outbound-gateway
request-channel="namedQueryRequestChannel"
reply-channel="namedQueryResponseChannel"
named-query="updateStudentByRollNumber"
entity-manager="em"
gateway-type="UPDATING">
<int-jpa:parameter name="lastName" expression="payload"/>
<int-jpa:parameter name="rollNumber" expression="headers['rollNumber']"/>
<int-jpa:transactional propagation="REQUIRES_NEW"
transaction-manager="transactionManager"/>
</int-jpa:outbound-gateway>
In the preceding example, the transactional element of the outbound gateway or adapter specifies the transaction attributes.
It is optional to define this child element if you have DirectChannel
as an input channel to the adapter and you want the adapter to execute the operations in the same transaction context as the caller.
If, however, you use an ExecutorChannel
, you must have the transactional
element, because the invoking client’s transaction context is not propagated.
Unlike the transactional element of the poller, which is defined in Spring Integration’s namespace, the transactional element for the outbound gateway or adapter is defined in the JPA namespace.
|
22.5. Inbound Channel Adapter
An inbound channel adapter is used to execute a select query over the database using JPA QL and return the result.
The message payload is either a single entity or a List
of entities.
The following XML configures an inbound-channel-adapter
:
<int-jpa:inbound-channel-adapter channel="inboundChannelAdapterOne" (1)
entity-manager="em" (2)
auto-startup="true" (3)
query="select s from Student s" (4)
expect-single-result="true" (5)
max-results="" (6)
max-results-expression="" (7)
delete-after-poll="true" (8)
flush-after-delete="true"> (9)
<int:poller fixed-rate="2000" >
<int:transactional propagation="REQUIRED" transaction-manager="transactionManager"/>
</int:poller>
</int-jpa:inbound-channel-adapter>
| 1 | The channel over which the inbound-channel-adapter puts the messages (with the payload) after executing the JPA QL in the query attribute. |
| 2 | The EntityManager instance used to perform the required JPA operations. |
| 3 | Attribute signaling whether the component should automatically start when the application context starts.
The value defaults to true . |
| 4 | The JPA QL whose result are sent out as the payload of the message |
| 5 | This attribute tells whether the JPQL query gives a single entity in the result or a List of entities.
If the value is set to true , the single entity is sent as the payload of the message.