If, however, multiple results are returned after setting this to true , a MessagingException is thrown.
The value defaults to false . |
| 6 | This non-zero, non-negative integer value tells the adapter not to select more than the given number of rows on execution of the select operation.
By default, if this attribute is not set, all possible records are selected by the query.
This attribute is mutually exclusive with max-results-expression .
Optional. |
| 7 | An expression that is evaluated to find the maximum number of results in a result set.
Mutually exclusive with max-results .
Optional. |
| 8 | Set this value to true if you want to delete the rows received after execution of the query.
You must ensure that the component operates as part of a transaction.
Otherwise, you may encounter an exception such as: java.lang.IllegalArgumentException: Removing a detached instance … |
| 9 | Set this value to true if you want to flush the persistence context immediately after deleting received entities and if you do not want to rely on the flushMode of the EntityManager .
The value defaults to false . |
22.5.1. Configuration Parameter Reference
The following listing shows all the values that can be set for an inbound-channel-adapter
:
<int-jpa:inbound-channel-adapter
auto-startup="true" (1)
channel="" (2)
delete-after-poll="false" (3)
delete-per-row="false" (4)
entity-class="" (5)
entity-manager="" (6)
entity-manager-factory="" (7)
expect-single-result="false" (8)
id=""
jpa-operations="" (9)
jpa-query="" (10)
named-query="" (11)
native-query="" (12)
parameter-source="" (13)
send-timeout=""> (14)
<int:poller ref="myPoller"/>
</int-jpa:inbound-channel-adapter>
| 1 | This lifecycle attribute signals whether this component should automatically start when the application context starts.
This attribute defaults to true .
Optional. |
| 2 | The channel to which the adapter sends a message with the payload from performing the desired JPA operation. |
| 3 | A boolean flag that indicates whether to delete the selected records after they have been polled by the adapter.
By default, the value is false (that is, the records are not deleted).
You must ensure that the component operates as part of a transaction.
Otherwise, you may encounter an exception, such as: java.lang.IllegalArgumentException: Removing a detached instance … .
Optional. |
| 4 | A boolean flag that indicates whether the records can be deleted in bulk or must be deleted one record at a time.
By default the value is false (that is, the records can be bulk-deleted).
Optional. |
| 5 | The fully qualified name of the entity class to be queried from the database. The adapter automatically builds a JPA Query based on the entity class name. Optional. |
| 6 | An instance of javax.persistence.EntityManager used to perform the JPA operations.
Optional. |
| 7 | An instance of javax.persistence.EntityManagerFactory used to obtain an instance of javax.persistence.EntityManager that performs the JPA operations.
Optional. |
| 8 | A boolean flag indicating whether the select operation is expected to return a single result or a List of results.
If this flag is set to true , the single entity selected is sent as the payload of the message.
If multiple entities are returned, an exception is thrown.
If false , the List of entities is sent as the payload of the message.
The value defaults to false .
Optional. |
| 9 | An implementation of org.springframework.integration.jpa.core.JpaOperations used to perform the JPA operations.
We recommend not providing an implementation of your own but using the default org.springframework.integration.jpa.core.DefaultJpaOperations implementation.
You can use any of the entity-manager , entity-manager-factory , or jpa-operations attributes.
Optional. |
| 10 | The JPA QL to be executed by this adapter. Optional. |
| 11 | The named query that needs to be executed by this adapter. Optional. |
| 12 | The native query executed by this adapter.
You can use any of the jpa-query , named-query , entity-class , or native-query attributes.
Optional. |
| 13 | An implementation of o.s.i.jpa.support.parametersource.ParameterSource used to resolve the values of the parameters in the query.
Ignored if the entity-class attribute has a value.
Optional. |
| 14 | Maximum amount of time (in milliseconds) to wait when sending a message to the channel. Optional. |
22.5.2. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with Java:
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@Bean
public JpaExecutor jpaExecutor() {
JpaExecutor executor = new JpaExecutor(this.entityManagerFactory);
jpaExecutor.setJpaQuery("from Student");
return executor;
}
@Bean
@InboundChannelAdapter(channel = "jpaInputChannel",
poller = @Poller(fixedDelay = "${poller.interval}"))
public MessageSource<?> jpaInbound() {
return new JpaPollingChannelAdapter(jpaExecutor());
}
@Bean
@ServiceActivator(inputChannel = "jpaInputChannel")
public MessageHandler handler() {
return message -> System.out.println(message.getPayload());
}
}
22.5.3. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the inbound adapter with the Java DSL:
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@Bean
public IntegrationFlow pollingAdapterFlow() {
return IntegrationFlows
.from(Jpa.inboundAdapter(this.entityManagerFactory)
.entityClass(StudentDomain.class)
.maxResults(1)
.expectSingleResult(true),
e -> e.poller(p -> p.trigger(new OnlyOnceTrigger())))
.channel(c -> c.queue("pollingResults"))
.get();
}
}
22.6. Outbound Channel Adapter
The JPA outbound channel adapter lets you accept messages over a request channel. The payload can either be used as the entity to be persisted or used with the headers in the parameter expressions for a JPQL query. The following sections cover the possible ways of performing these operations.
22.6.1. Using an Entity Class
The following XML configures the outbound channel adapter to persist an entity to the database:
<int-jpa:outbound-channel-adapter channel="entityTypeChannel" (1)
entity-class="org.springframework.integration.jpa.test.entity.Student" (2)
persist-mode="PERSIST" (3)
entity-manager="em"/ > (4)
| 1 | The channel over which a valid JPA entity is sent to the JPA outbound channel adapter. |
| 2 | The fully qualified name of the entity class accepted by the adapter to be persisted in the database. You can actually leave off this attribute in most cases as the adapter can determine the entity class automatically from the Spring Integration message payload. |
| 3 | The operation to be done by the adapter.
The valid values are PERSIST , MERGE , and DELETE .
The default value is MERGE . |
| 4 | The JPA entity manager to be used. |
These four attributes of the outbound-channel-adapter
configure it to accept entities over the input channel and process them to PERSIST
, MERGE
, or DELETE
the entities from the underlying data source.
As of Spring Integration 3.0, payloads to PERSIST or MERGE can also be of type java.lang.Iterable .
In that case, each object returned by the Iterable is treated as an entity and persisted or merged using the underlying EntityManager .
Null values returned by the iterator are ignored.
|
22.6.2. Using JPA Query Language (JPA QL)
The previous section showed how to perform a PERSIST
action by using an entity.
This section shows how to use an outbound channel adapter with JPA QL.
The following XML configures the outbound channel adapter to persist an entity to the database:
<int-jpa:outbound-channel-adapter channel="jpaQlChannel" (1)
jpa-query="update Student s set s.firstName = :firstName where s.rollNumber = :rollNumber" (2)
entity-manager="em"> (3)
<int-jpa:parameter name="firstName" expression="payload['firstName']"/> (4)
<int-jpa:parameter name="rollNumber" expression="payload['rollNumber']"/>
</int-jpa:outbound-channel-adapter>
| 1 | The input channel over which the message is sent to the outbound channel adapter. |
| 2 | The JPA QL to execute.
This query may contain parameters that are evaluated by using the parameter element. |
| 3 | The entity manager used by the adapter to perform the JPA operations. |
| 4 | The elements (one for each parameter) used to define the value of the parameter names for the JPA QL specified in the query attribute. |
The parameter
element accepts an attribute whose name
corresponds to the named parameter specified in the provided JPA QL (point 2 in the preceding example).
The value of the parameter can either be static or be derived by using an expression.
The static value and the expression to derive the value are specified using the value
and expression
attributes, respectively.
These attributes are mutually exclusive.
If the value
attribute is specified, you can provide an optional type
attribute.
The value of this attribute is the fully qualified name of the class whose value is represented by the value
attribute.
By default, the type is assumed to be a java.lang.String
.
The following example shows how to define a JPA parameter:
<int-jpa:outbound-channel-adapter ...
>
<int-jpa:parameter name="level" value="2" type="java.lang.Integer"/>
<int-jpa:parameter name="name" expression="payload['name']"/>
</int-jpa:outbound-channel-adapter>
As the preceding example shows, you can use multiple parameter
elements within an outbound channel adapter element and define some parameters by using expressions and others with static values.
However, take care not to specify the same parameter name multiple times.
You should provide one parameter
element for each named parameter specified in the JPA query.
For example, we specify two parameters: level
and name
.
The level
attribute is a static value of type java.lang.Integer
, while the name
attribute is derived from the payload of the message.
Though specifying select is valid for JPA QL, it makes no sense to do so.
Outbound channel adapters do not return any result.
If you want to select some values, consider using the outbound gateway instead.
|
22.6.3. Using Native Queries
This section describes how to use native queries to perform operations with the JPA outbound channel adapter. Using native queries is similar to using JPA QL, except that the queries are native database queries. By using native queries, we lose database vendor independence, which we get using JPA QL.
One of the things we can achieve by using native queries is to perform database inserts, which is not possible with JPA QL. (To perform inserts, we send JPA entities to the channel adapter, as described earlier). Below is a small xml fragment that demonstrates the use of native query to insert values in a table.
| Named parameters may not be supported by your JPA provider in conjunction with native SQL queries. While they work fine with Hibernate, OpenJPA and EclipseLink do not support them. See issues.apache.org/jira/browse/OPENJPA-111. Section 3.8.12 of the JPA 2.0 spec states: “Only positional parameter binding and positional access to result items may be portably used for native queries.” |
The following example configures an outbound-channel-adapter with a native query:
<int-jpa:outbound-channel-adapter channel="nativeQlChannel"
native-query="insert into STUDENT_TABLE(FIRST_NAME,LAST_UPDATED) values (:lastName,:lastUpdated)" (1)
entity-manager="em">
<int-jpa:parameter name="lastName" expression="payload['updatedLastName']"/>
<int-jpa:parameter name="lastUpdated" expression="new java.util.Date()"/>
</int-jpa:outbound-channel-adapter>
| 1 | The native query executed by this outbound channel adapter. |
Note that the other attributes (such as channel
and entity-manager
) and the parameter
element have the same semantics as they do for JPA QL.
22.6.4. Using Named Queries
Using named queries is similar to using JPA QL or a native query, except that we specify a named query instead of a query.
First, we cover how to define a JPA named query.
Then we cover how to declare an outbound channel adapter to work with a named query.
If we have an entity called Student
, we can use annotations on the Student
class to define two named queries: selectStudent
and updateStudent
.
The following example shows how to do so:
@Entity
@Table(name="Student")
@NamedQueries({
@NamedQuery(name="selectStudent",
query="select s from Student s where s.lastName = 'Last One'"),
@NamedQuery(name="updateStudent",
query="update Student s set s.lastName = :lastName,
lastUpdated = :lastUpdated where s.id in (select max(a.id) from Student a)")
})
public class Student {
...
}
Alternatively, you can use orm.xml to define named queries as the following example shows:
<entity-mappings ...>
...
<named-query name="selectStudent">
<query>select s from Student s where s.lastName = 'Last One'</query>
</named-query>
</entity-mappings>
Now that we have shown how to define named queries by using annotations or by using orm.xml
, we now show a small XML fragment that defines an outbound-channel-adapter
by using a named query, as the following example shows:
<int-jpa:outbound-channel-adapter channel="namedQueryChannel"
named-query="updateStudent" (1)
entity-manager="em">
<int-jpa:parameter name="lastName" expression="payload['updatedLastName']"/>
<int-jpa:parameter name="lastUpdated" expression="new java.util.Date()"/>
</int-jpa:outbound-channel-adapter>
| 1 | The named query that we want the adapter to execute when it receives a message over the channel. |
22.6.5. Configuration Parameter Reference
The following listing shows all the attributes that you can set on an outbound channel adapter:
<int-jpa:outbound-channel-adapter
auto-startup="true" (1)
channel="" (2)
entity-class="" (3)
entity-manager="" (4)
entity-manager-factory="" (5)
id=""
jpa-operations="" (6)
jpa-query="" (7)
named-query="" (8)
native-query="" (9)
order="" (10)
parameter-source-factory="" (11)
persist-mode="MERGE" (12)
flush="true" (13)
flush-size="10" (14)
clear-on-flush="true" (15)
use-payload-as-parameter-source="true" (16)
<int:poller/>
<int-jpa:transactional/> (17)
<int-jpa:parameter/> (18)
</int-jpa:outbound-channel-adapter>
| 1 | Lifecycle attribute signaling whether this component should start during application context startup.
It defaults to true .
Optional. |
| 2 | The channel from which the outbound adapter receives messages for performing the desired operation. |
| 3 | The fully qualified name of the entity class for the JPA Operation.
The entity-class , query , and named-query attributes are mutually exclusive.
Optional. |
| 4 | An instance of javax.persistence.EntityManager used to perform the JPA operations.
Optional. |
| 5 | An instance of javax.persistence.EntityManagerFactory used to obtain an instance of javax.persistence.EntityManager , which performs the JPA operations.
Optional. |
| 6 | An implementation of org.springframework.integration.jpa.core.JpaOperations used to perform the JPA operations.
We recommend not providing an implementation of your own but using the default org.springframework.integration.jpa.core.DefaultJpaOperations implementation.
You can use any one of the entity-manager , entity-manager-factory , or jpa-operations attributes.
Optional. |
| 7 | The JPA QL to be executed by this adapter. Optional. |
| 8 | The named query that needs to be executed by this adapter. Optional. |
| 9 | The native query to be executed by this adapter.
You can use any one of the jpa-query , named-query , or native-query attributes.
Optional. |
| 10 | The order for this consumer when multiple consumers are registered, thereby managing load-balancing and failover.
It defaults to Ordered.LOWEST_PRECEDENCE .
Optional. |
| 11 | An instance of o.s.i.jpa.support.parametersource.ParameterSourceFactory used to get an instance of o.s.i.jpa.support.parametersource.ParameterSource , which is used to resolve the values of the parameters in the query.
Ignored if you perform operations by using a JPA entity.
The parameter sub-elements are mutually exclusive with the parameter-source-factory attribute and must be configured on the provided ParameterSourceFactory .
Optional. |
| 12 | Accepts one of the following: PERSIST , MERGE , or DELETE .
Indicates the operation that the adapter needs to perform.
Relevant only if you use an entity for JPA operations.
Ignored if you provide JPA QL, a named query, or a native query.
It defaults to MERGE .
Optional.
As of Spring Integration 3.0, payloads to persist or merge can also be of type java.lang.Iterable .
In that case, each object returned by the Iterable is treated as an entity and persisted or merged by using the underlying EntityManager .
Null values returned by the iterator are ignored. |
| 13 | Set this value to true if you want to flush the persistence context immediately after persist, merge, or delete operations and do not want to rely on the flushMode of the EntityManager .
It defaults to false .
Applies only if you did not specify the flush-size attribute.
If this attribute is set to true , flush-size is implicitly set to 1 , if no other value configured it. |
| 14 | Set this attribute to a value greater than '0' if you want to flush the persistence context immediately after persist, merge or delete operations and do not want to rely on the the flushMode of the EntityManager .
The default value is set to 0 , which means "'no flush'".
This attribute is geared towards messages with Iterable payloads.
For instance, if flush-size is set to 3 , then entityManager.flush() is called after every third entity.
Furthermore, entityManager.flush() is called once more after the entire loop.
If the 'flush-size' attribute is specified with a value greater than '0', you need not configure the flush attribute. |
| 15 | Set this value to 'true' if you want to clear the persistence context immediately after each flush operation.
The attribute’s value is applied only if the flush attribute is set to true or if the flush-size attribute is set to a value greater than 0 . |
| 16 | If set to true , the payload of the message is used as a source for parameters.
If set to false , however, the entire Message is available as a source for parameters.
Optional. |
| 17 | Defines the transaction management attributes and the reference to the transaction manager to be used by the JPA adapter. Optional. |
| 18 | One or more parameter attributes — one for each parameter used in the query.
The value or expression is evaluated to compute the value of the parameter.
Optional. |
22.6.6. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the outbound adapter with Java:
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
@IntegrationComponentScan
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@MessagingGateway
interface JpaGateway {
@Gateway(requestChannel = "jpaPersistChannel")
@Transactional
void persistStudent(StudentDomain payload);
}
@Bean
public JpaExecutor jpaExecutor() {
JpaExecutor executor = new JpaExecutor(this.entityManagerFactory);
jpaExecutor.setEntityClass(StudentDomain.class);
jpaExecutor.setPersistMode(PersistMode.PERSIST);
return executor;
}
@Bean
@ServiceActivator(channel = "jpaPersistChannel")
public MessageHandler jpaOutbound() {
JpaOutboundGateway adapter = new JpaOutboundGateway(jpaExecutor());
adapter.setProducesReply(false);
return adapter;
}
}
22.6.7. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound adapter with the Java DSL:
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@Bean
public IntegrationFlow outboundAdapterFlow() {
return f -> f
.handle(Jpa.outboundAdapter(this.entityManagerFactory)
.entityClass(StudentDomain.class)
.persistMode(PersistMode.PERSIST),
e -> e.transactional());
}
}
22.7. Outbound Gateways
The JPA inbound channel adapter lets you poll a database to retrieve one or more JPA entities. The retrieved data is consequently used to start a Spring Integration flow that uses the retrieved data as message payload.
Additionally, you can use JPA outbound channel adapters at the end of your flow in order to persist data, essentially terminating the flow at the end of the persistence operation.
However, how can you execute JPA persistence operations in the middle of a flow? For example, you may have business data that you are processing in your Spring Integration message flow and that you would like to persist, yet you still need to use other components further downstream. Alternatively, instead of polling the database using a poller, you need to execute JPQL queries and actively retrieve data, which is then processed in subsequent components within your flow.
This is where JPA Outbound Gateways come into play. They give you the ability to persist data as well as retrieving data. To facilitate these uses, Spring Integration provides two types of JPA outbound gateways:
-
Updating outbound gateway
-
Retrieving outbound gateway
Whenever the outbound gateway is used to perform an action that saves, updates, or solely deletes some records in the database, you need to use an updating outbound gateway.
If, for example, you use an entity
to persist it, a merged and persisted entity is returned as a result.
In other cases, the number of records affected (updated or deleted) is returned instead.
When retrieving (selecting) data from the database, we use a retrieving outbound gateway. With a retrieving outbound gateway, we can use JPQL, Named Queries (native or JPQL-based), or Native Queries (SQL) for selecting the data and retrieving the results.
An updating outbound gateway is functionally similar to an outbound channel adapter, except that an updating outbound gateway sends a result to the gateway’s reply channel after performing the JPA operation.
A retrieving outbound gateway is similar to an inbound channel adapter.
| We recommend you first read the Outbound Channel Adapter section and the Inbound Channel Adapter sections earlier in this chapter, as most of the common concepts are explained there. |
This similarity was the main factor to use the central JpaExecutor
class to unify common functionality as much as possible.
Common for all JPA outbound gateways and similar to the outbound-channel-adapter
, we can use for performing various JPA operations:
-
Entity classes
-
JPA Query Language (JPQL)
-
Native query
-
Named query
For configuration examples see JPA Outbound Gateway Samples.
22.7.1. Common Configuration Parameters
JPA Outbound Gateways always have access to the Spring Integration Message
as input.
Consequently, the following parameters is available:
parameter-source-factory
-
An instance of
o.s.i.jpa.support.parametersource.ParameterSourceFactory
used to get an instance ofo.s.i.jpa.support.parametersource.ParameterSource
. TheParameterSource
is used to resolve the values of the parameters provided in the query. If you perform operations by using a JPA entity, theparameter-source-factory
attribute is ignored. Theparameter
sub-elements are mutually exclusive with theparameter-source-factory
and they have to be configured on the providedParameterSourceFactory
. Optional. use-payload-as-parameter-source
-
If set to
true
, the payload of theMessage
is used as a source for parameters. If set tofalse
, the entireMessage
is available as a source for parameters. If no JPA Parameters are passed in, this property defaults totrue
. This means that, if you use a defaultBeanPropertyParameterSourceFactory
, the bean properties of the payload are used as a source for parameter values for the JPA query. However, if JPA Parameters are passed in, this property, by default, evaluates tofalse
. The reason is that JPA Parameters let you provide SpEL Expressions. Therefore, it is highly beneficial to have access to the entireMessage
, including the headers. Optional.
22.7.2. Updating Outbound Gateway
The following listing shows all the attributes that you can set on an updating-outbound-gateway and describes the key attributes:
<int-jpa:updating-outbound-gateway request-channel="" (1)
auto-startup="true"
entity-class=""
entity-manager=""
entity-manager-factory=""
id=""
jpa-operations=""
jpa-query=""
named-query=""
native-query=""
order=""
parameter-source-factory=""
persist-mode="MERGE"
reply-channel="" (2)
reply-timeout="" (3)
use-payload-as-parameter-source="true">
<int:poller/>
<int-jpa:transactional/>
<int-jpa:parameter name="" type="" value=""/>
<int-jpa:parameter name="" expression=""/>
</int-jpa:updating-outbound-gateway>
| 1 | The channel from which the outbound gateway receives messages for performing the desired operation.
This attribute is similar to the channel attribute of the outbound-channel-adapter .
Optional. |
| 2 | The channel to which the gateway send the response after performing the required JPA operation.
If this attribute is not defined, the request message must have a replyChannel header.
Optional. |
| 3 | Specifies the time the gateway waits to send the result to the reply channel.
Only applies when the reply channel itself might block the send operation (for example, a bounded QueueChannel that is currently full).
By default, the gateway waits indefinitely.
The value is specified in milliseconds.
Optional. |
The remaining attributes are described earlier in this chapter. See Configuration Parameter Reference and Configuration Parameter Reference.
22.7.3. Configuring with Java Configuration
The following Spring Boot application shows an example of how configure the outbound adapter with Java:
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
@IntegrationComponentScan
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@MessagingGateway
interface JpaGateway {
@Gateway(requestChannel = "jpaUpdateChannel")
@Transactional
void updateStudent(StudentDomain payload);
}
@Bean
@ServiceActivator(channel = "jpaUpdateChannel")
public MessageHandler jpaOutbound() {
JpaOutboundGateway adapter =
new JpaOutboundGateway(new JpaExecutor(this.entityManagerFactory));
adapter.setOutputChannelName("updateResults");
return adapter;
}
}
22.7.4. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound adapter using the Java DSL:
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@Bean
public IntegrationFlow updatingGatewayFlow() {
return f -> f
.handle(Jpa.updatingGateway(this.entityManagerFactory),
e -> e.transactional(true))
.channel(c -> c.queue("updateResults"));
}
}
22.7.5. Retrieving Outbound Gateway
The following example shows all the attributes that you can set on a retrieving outbound gateway and describes the key attributes:
<int-jpa:retrieving-outbound-gateway request-channel=""
auto-startup="true"
delete-after-poll="false"
delete-in-batch="false"
entity-class=""
id-expression="" (1)
entity-manager=""
entity-manager-factory=""
expect-single-result="false" (2)
id=""
jpa-operations=""
jpa-query=""
max-results="" (3)
max-results-expression="" (4)
first-result="" (5)
first-result-expression="" (6)
named-query=""
native-query=""
order=""
parameter-source-factory=""
reply-channel=""
reply-timeout=""
use-payload-as-parameter-source="true">
<int:poller></int:poller>
<int-jpa:transactional/>
<int-jpa:parameter name="" type="" value=""/>
<int-jpa:parameter name="" expression=""/>
</int-jpa:retrieving-outbound-gateway>
| 1 | (Since Spring Integration 4.0) The SpEL expression that determines the primaryKey value for EntityManager.find(Class entityClass, Object primaryKey) method against the requestMessage as the root object of evaluation context.
The entityClass argument is determined from the entity-class attribute, if present.
Otherwise, it is determined from the payload class.
All other attributes are disallowed if you use id-expression .
Optional. |
| 2 | A boolean flag indicating whether the select operation is expected to return a single result or a List of results.
If this flag is set to true , a single entity is sent as the payload of the message.
If multiple entities are returned, an exception is thrown.
If false , the List of entities is sent as the payload of the message.
It defaults to false .
Optional. |
| 3 | This non-zero, non-negative integer value tells the adapter not to select more than the specified number of rows on execution of the select operation.
By default, if this attribute is not set, all the possible records are selected by given query.
This attribute is mutually exclusive with max-results-expression .
Optional. |
| 4 | An expression that can be used to find the maximum number of results in a result set.
It is mutually exclusive with max-results .
Optional. |
| 5 | This non-zero, non-negative integer value tells the adapter the first record from which results are to be retrieved.
This attribute is mutually exclusive with first-result-expression .
Version 3.0 introduced this attribute.
Optional. |
| 6 | This expression is evaluated against the message, to find the position of the first record in the result set.
This attribute is mutually exclusive to first-result .
Version 3.0 introduced this attribute.
Optional. |
The remaining attributes are described earlier in this chapter. See Configuration Parameter Reference and Configuration Parameter Reference.
22.7.6. Configuring with Java Configuration
The following Spring Boot application shows an example of how configure the outbound adapter with Java:
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@Bean
public JpaExecutor jpaExecutor() {
JpaExecutor executor = new JpaExecutor(this.entityManagerFactory);
jpaExecutor.setJpaQuery("from Student s where s.id = :id");
executor.setJpaParameters(Collections.singletonList(new JpaParameter("id", null, "payload")));
jpaExecutor.setExpectSingleResult(true);
return executor;
}
@Bean
@ServiceActivator(channel = "jpaRetrievingChannel")
public MessageHandler jpaOutbound() {
JpaOutboundGateway adapter = new JpaOutboundGateway(jpaExecutor());
adapter.setOutputChannelName("retrieveResults");
adapter.setGatewayType(OutboundGatewayType.RETRIEVING);
return adapter;
}
}
22.7.7. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound adapter with the Java DSL:
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@Bean
public IntegrationFlow retrievingGatewayFlow() {
return f -> f
.handle(Jpa.retrievingGateway(this.entityManagerFactory)
.jpaQuery("from Student s where s.id = :id")
.expectSingleResult(true)
.parameterExpression("id", "payload"))
.channel(c -> c.queue("retrieveResults"));
}
}
|
When you choose to delete entities upon retrieval and you have retrieved a collection of entities, by default, entities are deleted on a per-entity basis. This may cause performance issues. Alternatively, you can set attribute JSR 317: Java™ Persistence 2.0 states in chapter 4.10, “Bulk Update and Delete Operations” that: “A delete operation only applies to entities of the specified class and its subclasses. It does not cascade to related entities.” For more information, see JSR 317: Java™ Persistence 2.0 |
22.7.8. JPA Outbound Gateway Samples
This section contains various examples of using the updating outbound gateway and the retrieving outbound gateway:
Update by Using an Entity Class
In the following example, an updating outbound gateway is persisted by using the org.springframework.integration.jpa.test.entity.Student
entity class as a JPA defining parameter:
<int-jpa:updating-outbound-gateway request-channel="entityRequestChannel" (1)
reply-channel="entityResponseChannel" (2)
entity-class="org.springframework.integration.jpa.test.entity.Student"
entity-manager="em"/>
| 1 | This is the request channel for the outbound gateway.
It is similar to the channel attribute of the outbound-channel-adapter . |
| 2 | This is where a gateway differs from an outbound adapter.
This is the channel over which the reply from the JPA operation is received.
If, however, you are not interested in the reply received and want only to perform the operation, using a JPA outbound-channel-adapter is the appropriate choice.
In this example, where we use an entity class, the reply is the entity object that was created or merged as a result of the JPA operation. |
Update using JPQL
The following example updates an entity by using the Java Persistence Query Language (JPQL), which mandates using an updating outbound gateway:
<int-jpa:updating-outbound-gateway request-channel="jpaqlRequestChannel"
reply-channel="jpaqlResponseChannel"
jpa-query="update Student s set s.lastName = :lastName where s.rollNumber = :rollNumber" (1)
entity-manager="em">
<int-jpa:parameter name="lastName" expression="payload"/>
<int-jpa:parameter name="rollNumber" expression="headers['rollNumber']"/>
</int-jpa:updating-outbound-gateway>
| 1 | The JPQL query that the gateway executes.
Since we used updating outbound gateway, only update and delete JPQL queries would be sensible choices. |
When you send a message with a String
payload that also contains a header called rollNumber
with a long
value, the last name of the student with the specified roll number is updated to the value in the message payload.
When using an updating gateway, the return value is always an integer value, which denotes the number of records affected by execution of the JPA QL.
Retrieving an Entity using JPQL
The following example uses a retrieving outbound gateway and JPQL to retrieve (select) one or more entities from the database:
<int-jpa:retrieving-outbound-gateway request-channel="retrievingGatewayReqChannel"
reply-channel="retrievingGatewayReplyChannel"
jpa-query="select s from Student s where s.firstName = :firstName and s.lastName = :lastName"
entity-manager="em">
<int-jpa:parameter name="firstName" expression="payload"/>
<int-jpa:parameter name="lastName" expression="headers['lastName']"/>
</int-jpa:outbound-gateway>
Retrieving an Entity by Using id-expression
The following example uses a retrieving outbound gateway with id-expression
to retrieve (find) one and only one entity from the database:
The primaryKey
is the result of id-expression
evaluation.
The entityClass
is a class of Message payload
.
<int-jpa:retrieving-outbound-gateway
request-channel="retrievingGatewayReqChannel"
reply-channel="retrievingGatewayReplyChannel"
id-expression="payload.id"
entity-manager="em"/>
Update using a Named Query
Using a named query is basically the same as using a JPQL query directly.
The difference is that the named-query
attribute is used instead, as the following example shows:
<int-jpa:updating-outbound-gateway request-channel="namedQueryRequestChannel"
reply-channel="namedQueryResponseChannel"
named-query="updateStudentByRollNumber"
entity-manager="em">
<int-jpa:parameter name="lastName" expression="payload"/>
<int-jpa:parameter name="rollNumber" expression="headers['rollNumber']"/>
</int-jpa:outbound-gateway>
| You can find a complete sample application that uses Spring Integration’s JPA adapter here. |
23. JMS Support
Spring Integration provides channel adapters for receiving and sending JMS messages.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jms</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-jms:5.1.5.RELEASE"
The javax.jms:javax.jms-api
must be added explicitly via some JMS vendor-specific implementation, e.g. Apache ActiveMQ.
There are actually two JMS-based inbound Channel Adapters.
The first uses Spring’s JmsTemplate
to receive based on a polling period.
The second is “message-driven” and relies on a Spring MessageListener
container.
The outbound channel adapter uses the JmsTemplate
to convert and send a JMS message on demand.
By using JmsTemplate
and the MessageListener
container, Spring Integration relies on Spring’s JMS support.
This is important to understand, since most of the attributes exposed on these adapters configure the underlying JmsTemplate
and MessageListener
container.
For more details about JmsTemplate
and the MessageListener
container, see the Spring JMS documentation.
Whereas the JMS channel adapters are intended for unidirectional messaging (send-only or receive-only), Spring Integration also provides inbound and outbound JMS Gateways for request and reply operations.
The inbound gateway relies on one of Spring’s MessageListener
container implementations for message-driven reception.
It is also capable of sending a return value to the reply-to
destination, as provided by the received message.
The outbound gateway sends a JMS message to a request-destination
(or request-destination-name
or request-destination-expression
) and then receives a reply message.
You can explicitly configure the reply-destination
reference (or reply-destination-name
or reply-destination-expression
).
Otherwise, the outbound gateway uses a JMS TemporaryQueue.
Prior to Spring Integration 2.2, if necessary, a TemporaryQueue
was created (and removed) for each request or reply.
Beginning with Spring Integration 2.2, you can configure the outbound gateway to use a MessageListener
container to receive replies instead of directly using a new (or cached) Consumer
to receive the reply for each request.
When so configured, and no explicit reply destination is provided, a single TemporaryQueue
is used for each gateway instead of one for each request.
23.1. Inbound Channel Adapter
The inbound channel adapter requires a reference to either a single JmsTemplate
instance or both a ConnectionFactory
and a Destination
(you can provide a 'destinationName' in place of the 'destination' reference).
The following example defines an inbound channel adapter with a Destination
reference:
<int-jms:inbound-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel">
<int:poller fixed-rate="30000"/>
</int-jms:inbound-channel-adapter>
Notice from the preceding configuration that the inbound-channel-adapter is a polling consumer.
That means that it invokes receive() when triggered.
You should use this should only in situations where polling is done relatively infrequently and timeliness is not important.
For all other situations (a vast majority of JMS-based use-cases), the message-driven-channel-adapter (described later) is a better option.
|
By default, all of the JMS adapters that require a reference to the ConnectionFactory automatically look for a bean named jmsConnectionFactory .
That is why you do not see a connection-factory attribute in many of the examples.
However, if your JMS ConnectionFactory has a different bean name, you need to provide that attribute.
|
If extract-payload
is set to true
(the default), the received JMS Message is passed through the MessageConverter
.
When relying on the default SimpleMessageConverter
, this means that the resulting Spring Integration Message has the JMS message’s body as its payload.
A JMS TextMessage
produces a string-based payload, a JMS BytesMessage
produces a byte array payload, and the serializable instance of a JMS ObjectMessage
becomes the Spring Integration message’s payload.
If you prefer to have the raw JMS message as the Spring Integration message’s payload, set extract-payload
to false
, as the following example shows:
<int-jms:inbound-channel-adapter id="jmsIn"
destination="inQueue"
channel="exampleChannel"
extract-payload="false"/>
<int:poller fixed-rate="30000"/>
</int-jms:inbound-channel-adapter>
Starting with version 5.0.8, a default value of the receive-timeout
is -1
(no wait) for the org.springframework.jms.connection.CachingConnectionFactory
and cacheConsumers
, otherwise it is 1 second.
23.1.1. Transactions
Starting with version 4.0, the inbound channel adapter supports the session-transacted
attribute.
In earlier versions, you had to inject a JmsTemplate
with sessionTransacted
set to true
.
(The adapter did let you set the acknowledge
attribute to transacted
, but this was incorrect and did not work).
Note, however, that setting session-transacted
to true
has little value, because the transaction is committed
immediately after the receive()
operation and before the message is sent to the channel
.
If you want the entire flow to be transactional (for example, if there is a downstream outbound channel adapter), you must use a transactional
poller with a JmsTransactionManager
.
Alternatively, consider using a jms-message-driven-channel-adapter
with acknowledge
set to transacted
(the default).
23.2. Message-driven Channel Adapter
The message-driven-channel-adapter
requires a reference to either an instance of a Spring MessageListener
container (any subclass of AbstractMessageListenerContainer
) or both ConnectionFactory
and Destination
(a 'destinationName' can be provided in place of the 'destination' reference).
The following example defines a message-driven channel adapter with a Destination
reference:
<int-jms:message-driven-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel"/>
|
The message-driven adapter also accepts several properties that pertain to the If you have a custom listener container implementation (usually a subclass of |
|
You can’t use the Spring JMS namespace element It is recommended to configure a regular |
Starting with version 4.2, the default acknowledge mode is transacted , unless you provide an external
container.
In that case, you should configure the container as needed.
We recommend using transacted with the DefaultMessageListenerContainer to avoid message loss.
|
The 'extract-payload' property has the same effect, and its default value is 'true'.
The poller
element is not applicable for a message-driven channel adapter, as it is actively invoked.
For most scenarios, the message-driven approach is better, since the messages are passed along to the MessageChannel
as soon as they are received from the underlying JMS consumer.
Finally, the <message-driven-channel-adapter>
element also accepts the 'error-channel' attribute.
This provides the same basic functionality, as described in Enter the GatewayProxyFactoryBean
.
The following example shows how to set an error channel on a message-driven channel adapter:
<int-jms:message-driven-channel-adapter id="jmsIn" destination="inQueue"
channel="exampleChannel"
error-channel="exampleErrorChannel"/>
When comparing the preceding example to the generic gateway configuration or the JMS 'inbound-gateway' that we discuss later, the key difference is that we are in a one-way flow, since this is a 'channel-adapter', not a gateway.
Therefore, the flow downstream from the 'error-channel' should also be one-way.
For example, it could send to a logging handler or it could connect to a different JMS <outbound-channel-adapter>
element.
When consuming from topics, set the pub-sub-domain
attribute to true.
Set subscription-durable
to true
for a durable subscription or subscription-shared
for a shared subscription (which requires a JMS 2.0 broker and has been available since version 4.2).
Use subscription-name
to name the subscription.
Starting with version 5.1, when the endpoint is stopped while the application remains running, the underlying listener container is shut down, closing its shared connection and consumers.
Previously, the connection and consumers remained open.
To revert to the previous behavior, set the shutdownContainerOnStop
on the JmsMessageDrivenEndpoint
to false
.
23.2.1. Inbound Conversion Errors
Starting with version 4.2, the 'error-channel' is used for the conversion errors, too.
Previously, if a JMS <message-driven-channel-adapter/>
or <inbound-gateway/>
could not deliver a message due to a conversion error, an exception would be thrown back to the container.
If the container is configured to use transactions, the message is rolled back and redelivered repeatedly.
The conversion process occurs before and during message construction so that such errors are not sent to the 'error-channel'.
Now such conversion exceptions result in an ErrorMessage
being sent to the 'error-channel', with the exception as the payload
.
If you wish the transaction to roll back and you have an 'error-channel' defined, the integration flow on the 'error-channel' must re-throw the exception (or another exception).
If the error flow does not throw an exception, the transaction is committed and the message is removed.
If no 'error-channel' is defined, the exception is thrown back to the container, as before.
23.3. Outbound Channel Adapter
The JmsSendingMessageHandler
implements the MessageHandler
interface and is capable of converting Spring Integration Messages
to JMS messages and then sending to a JMS destination.
It requires either a jmsTemplate
reference or both jmsConnectionFactory
and destination
references (destinationName
may be provided in place of destination
).
As with the inbound channel adapter, the easiest way to configure this adapter is with the namespace support.
The following configuration produces an adapter that receives Spring Integration messages from the exampleChannel
, converts those into JMS messages, and sends them to the JMS destination reference whose bean name is outQueue
:
<int-jms:outbound-channel-adapter id="jmsOut" destination="outQueue" channel="exampleChannel"/>
As with the inbound channel adapters, there is an 'extract-payload' property.
However, the meaning is reversed for the outbound adapter.
Rather than applying to the JMS message, the boolean property applies to the Spring Integration message payload.
In other words, the decision is whether to pass the Spring Integration message itself as the JMS message body or to pass the Spring Integration message payload as the JMS message body.
The default value is 'true'.

If, on the other hand, you want to send the actual Spring Integration message to another system over JMS, set it to 'false'.
Regardless of the boolean value for payload extraction, the Spring Integration MessageHeaders map to JMS properties, as long as you rely on the default converter or provide a reference to another instance of HeaderMappingMessageConverter .
(The same holds true for 'inbound' adapters, except that, in those cases, the JMS properties map to Spring Integration MessageHeaders ).
|
Starting with version 5.1, the <int-jms:outbound-channel-adapter>
(JmsSendingMessageHandler
) can be configured with the deliveryModeExpression
and timeToLiveExpression
properties to evaluate an appropriate QoS values for JMS message to send at runtime against request Spring Message
.
The new setMapInboundDeliveryMode(true)
and setMapInboundExpiration(true)
options of the DefaultJmsHeaderMapper
may facilitate as a source of the information for the dynamic deliveryMode
and timeToLive
from message headers:
<int-jms:outbound-channel-adapter delivery-mode-expression="headers.jms_deliveryMode"
time-to-live-expression="headers.jms_expiration - T(System).currentTimeMillis()"/>
23.3.1. Transactions
Starting with version 4.0, the outbound channel adapter supports the session-transacted
attribute.
In earlier versions, you had to inject a JmsTemplate
with sessionTransacted
set to true
.
The attribute now sets the property on the built-in default JmsTemplate
.
If a transaction exists (perhaps from an upstream message-driven-channel-adapter
), the send operation is performed within the same transaction.
Otherwise, a new transaction is started.
23.4. Inbound Gateway
Spring Integration’s message-driven JMS inbound-gateway delegates to a MessageListener
container, supports dynamically adjusting concurrent consumers, and can also handle replies.
The inbound gateway requires references to a ConnectionFactory
and a request Destination
(or 'requestDestinationName').
The following example defines a JMS inbound-gateway
that receives from the JMS queue referenced by the bean id, inQueue
, and sends to the Spring Integration channel named exampleChannel
:
<int-jms:inbound-gateway id="jmsInGateway"
request-destination="inQueue"
request-channel="exampleChannel"/>
Since the gateways provide request-reply behavior instead of unidirectional send or receive behavior, they also have two distinct properties for “payload extraction” (as discussed earlier for the channel adapters' 'extract-payload' setting). For an inbound gateway, the 'extract-request-payload' property determines whether the received JMS Message body is extracted. If 'false', the JMS message itself becomes the Spring Integration message payload. The default is 'true'.
Similarly, for an inbound-gateway, the 'extract-reply-payload' property applies to the Spring Integration message that is to be converted into a reply JMS Message.
If you want to pass the whole Spring Integration message (as the body of a JMS ObjectMessage), set value this to 'false'.
By default, it is also 'true' that the Spring Integration message payload is converted into a JMS Message (for example, a
String
payload becomes a JMS TextMessage).
As with anything else, gateway invocation might result in error. By default, a producer is not notified of the errors that might have occurred on the consumer side and times out waiting for the reply. However, there might be times when you want to communicate an error condition back to the consumer (in other words, you might want to treat the exception as a valid reply by mapping it to a message). To accomplish this, JMS inbound gateway provides support for a message channel to which errors can be sent for processing, potentially resulting in a reply message payload that conforms to some contract that defines what a caller may expect as an “error” reply. You can use the error-channel attribute to configure such a channel, as the following example shows:
<int-jms:inbound-gateway request-destination="requestQueue"
request-channel="jmsInputChannel"
error-channel="errorTransformationChannel"/>
<int:transformer input-channel="exceptionTransformationChannel"
ref="exceptionTransformer" method="createErrorResponse"/>
You might notice that this example looks very similar to that included within Enter the GatewayProxyFactoryBean
.
The same idea applies here: The exceptionTransformer
could be a POJO that creates error-response objects, you could reference the nullChannel
to suppress the errors, or you could leave 'error-channel' out to let the exception propagate.
When consuming from topics, set the pub-sub-domain
attribute to true.
Set subscription-durable
to true
for a durable subscription or subscription-shared
for a shared subscription (requires a JMS 2.0 broker and has been available since version 4.2).
Use subscription-name
to name the subscription.
Starting with version 4.2, the default acknowledge mode is transacted , unless an external container is provided.
In that case, you should configure the container as needed.
We recommend that you use transacted with the DefaultMessageListenerContainer to avoid message loss.
|
Starting with version 5.1, when the endpoint is stopped while the application remains running, the underlying listener container is shut down, closing its shared connection and consumers.
Previously, the connection and consumers remained open.
To revert to the previous behavior, set the shutdownContainerOnStop
on the JmsInboundGateway
to false
.
23.5. Outbound Gateway
The outbound gateway creates JMS messages from Spring Integration messages and sends them to a 'request-destination'.
It thens handle the JMS reply message either by using a selector to receive from the 'reply-destination' that you configure or, if no 'reply-destination' is provided, by creating JMS TemporaryQueue
instances.
|
Using a If you specify a reply destination, you are advised to not use cached consumers.
Alternatively, consider using a |
The following example shows how to configure an outbound gateway:
<int-jms:outbound-gateway id="jmsOutGateway"
request-destination="outQueue"
request-channel="outboundJmsRequests"
reply-channel="jmsReplies"/>
The 'outbound-gateway' payload extraction properties are inversely related to those of the 'inbound-gateway' (see the earlier discussion). That means that the 'extract-request-payload' property value applies to the Spring Integration message being converted into a JMS message to be sent as a request. The 'extract-reply-payload' property value applies to the JMS message received as a reply and is then converted into a Spring Integration message to be subsequently sent to the 'reply-channel', as shown in the preceding configuration example.
23.5.1. Using a <reply-listener/>
Spring Integration 2.2 introduced an alternative technique for handling replies.
If you add a <reply-listener/>
child element to the gateway instead of creating a consumer for each reply, a MessageListener
container is used to receive the replies and hand them over to the requesting thread.
This provides a number of performance benefits as well as alleviating the cached consumer memory utilization problem described in the earlier caution.
When using a <reply-listener/>
with an outbound gateway that has no reply-destination
, instead of creating a TemporaryQueue
for each request, a single TemporaryQueue
is used.
(The gateway creates an additional TemporaryQueue
, as necessary, if the connection to the broker is lost and recovered).
When using a correlation-key
, multiple gateways can share the same reply destination, because the listener container uses a selector that is unique to each gateway.
|
If you specify a reply listener and specify a reply destination (or reply destination name) but provide no correlation key, the gateway logs a warning and falls back to pre-version 2.2 behavior. This is because there is no way to configure a selector in this case. Thus, there is no way to avoid a reply going to a different gateway that might be configured with the same reply destination. Note that, in this situation, a new consumer is used for each request, and consumers can build up in memory as described in the caution above; therefore cached consumers should not be used in this case. |
The following example shows a reply listener with default attributes:
<int-jms:outbound-gateway id="jmsOutGateway"
request-destination="outQueue"
request-channel="outboundJmsRequests"
reply-channel="jmsReplies">
<int-jms:reply-listener />
</int-jms-outbound-gateway>
The listener is very lightweight, and we anticipate that, in most cases, you need only a single consumer.
However, you can add attributes such as concurrent-consumers
, max-concurrent-consumers
, and others.
See the schema for a complete list of supported attributes, together with the Spring JMS documentation for their meanings.
23.5.2. Idle Reply Listeners
Starting with version 4.2, you can start the reply listener as needed (and stop it after an idle time) instead of running for the duration of the gateway’s lifecycle. This can be useful if you have many gateways in the application context where they are mostly idle. One such situation is a context with many (inactive) partitioned Spring Batch jobs using Spring Integration and JMS for partition distribution. If all the reply listeners are active, the JMS broker has an active consumer for each gateway. By enabling the idle timeout, each consumer exists only while the corresponding batch job is running (and for a short time after it finishes).
See idle-reply-listener-timeout
in Attribute Reference.
23.5.3. Gateway Reply Correlation
This section describes the mechanisms used for reply correlation (ensuring the originating gateway receives replies to only its requests), depending on how the gateway is configured. See Attribute Reference for complete description of the attributes discussed here.
The following list describes the various scenarios (the numbers are for identification — order does not matter):
-
No
reply-destination*
properties and no<reply-listener>
A
TemporaryQueue
is created for each request and deleted when the request is complete (successfully or otherwise).correlation-key
is irrelevant. -
A
reply-destination*
property is provided and neither a<reply-listener/>
nor acorrelation-key
is providedThe
JMSCorrelationID
equal to the outgoing message IS is used as a message selector for the consumer:messageSelector = "JMSCorrelationID = '" + messageId + "'"
The responding system is expected to return the inbound
JMSMessageID
in the replyJMSCorrelationID
. This is a common pattern and is implemented by the Spring Integration inbound gateway as well as Spring’sMessageListenerAdapter
for message-driven POJOs.When you use this configuration, you should not use a topic for replies. The reply may be lost. -
A
reply-destination*
property is provided, no<reply-listener/>
is provided, andcorrelation-key="JMSCorrelationID"
The gateway generates a unique correlation IS and inserts it in the
JMSCorrelationID
header. The message selector is:messageSelector = "JMSCorrelationID = '" + uniqueId + "'"
The responding system is expected to return the inbound
JMSCorrelationID
in the replyJMSCorrelationID
. This is a common pattern and is implemented by the Spring Integration inbound gateway as well as Spring’sMessageListenerAdapter
for message-driven POJOs. -
A
reply-destination*
property is provided, no<reply-listener/>
is provided, andcorrelation-key="myCorrelationHeader"
The gateway generates a unique correlation ID and inserts it in the
myCorrelationHeader
message property. Thecorrelation-key
can be any user-defined value. The message selector is:messageSelector = "myCorrelationHeader = '" + uniqueId + "'"
The responding system is expected to return the inbound
myCorrelationHeader
in the replymyCorrelationHeader
. -
A
reply-destination*
property is provided, no<reply-listener/>
is provided, andcorrelation-key="JMSCorrelationID*"
:: (Note the*
in the correlation key.)The gateway uses the value in the
jms_correlationId
header (if present) from the request message and inserts it in theJMSCorrelationID
header. The message selector is:messageSelector = "JMSCorrelationID = '" + headers['jms_correlationId'] + "'"
The user must ensure this value is unique.
If the header does not exist, the gateway behaves as in
3
.The responding system is expected to return the inbound
JMSCorrelationID
in the replyJMSCorrelationID
. This is a common pattern and is implemented by the Spring Integration inbound gateway as well as Spring’sMessageListenerAdapter
for message-driven POJOs. -
No
reply-destination*
properties is provided, and a<reply-listener>
is providedA temporary queue is created and used for all replies from this gateway instance. No correlation data is needed in the message, but the outgoing
JMSMessageID
is used internally in the gateway to direct the reply to the correct requesting thread. -
A
reply-destination*
property is provided, a<reply-listener>
is provided, and nocorrelation-key
is providedNot allowed.
The
<reply-listener/>
configuration is ignored, and the gateway behaves as in2
. A warning log message is written to indicate this situation. -
A
reply-destination*
property is provided, a<reply-listener>
is provided, andcorrelation-key="JMSCorrelationID"
The gateway has a unique correlation ID and inserts it, together with an incrementing value in the
JMSCorrelationID
header (gatewayId + "_" + ++seq
). The message selector is:messageSelector = "JMSCorrelationID LIKE '" + gatewayId%'"
The responding system is expected to return the inbound
JMSCorrelationID
in the replyJMSCorrelationID
. This is a common pattern and is implemented by the Spring Integration inbound gateway as well as Spring’sMessageListenerAdapter
for message-driven POJOs. Since each gateway has a unique ID, each instance gets only its own replies. The complete correlation data is used to route the reply to the correct requesting thread. -
A
reply-destination*
property is provided a<reply-listener/>
is provided, andcorrelation-key="myCorrelationHeader"
The gateway has a unique correlation ID and inserts it, together with an incrementing value in the
myCorrelationHeader
property (gatewayId + "_" + ++seq
). Thecorrelation-key
can be any user-defined value. The message selector is:messageSelector = "myCorrelationHeader LIKE '" + gatewayId%'"
The responding system is expected to return the inbound
myCorrelationHeader
in the replymyCorrelationHeader
. Since each gateway has a unique ID, each instance only gets its own replies. The complete correlation data is used to route the reply to the correct requesting thread. -
A
reply-destination*
property is provided, a<reply-listener/>
is provided, andcorrelation-key="JMSCorrelationID*"
*(Note the
*
in the correlation key)Not allowed.
User-supplied correlation IDs are not permitted with a reply listener. The gateway does not initialize with this configuration.
23.5.4. Async Gateway
Starting with version 4.3, you can now specify async="true"
(or setAsync(true)
in Java) when configuring the outbound
gateway.
By default, when a request is sent to the gateway, the requesting thread is suspended until the reply is received.
The flow then continues on that thread.
If async
is true
, the requesting thread is released immediately after the send completes, and the reply is returned
(and the flow continues) on the listener container thread.
This can be useful when the gateway is invoked on a poller thread.
The thread is released and is available for other tasks within the framework.
async
requires a <reply-listener/>
(or setUseReplyContainer(true)
when using Java configuration).
It also requires a correlationKey
(usually JMSCorrelationID
) to be specified.
If either of these conditions are not met, async
is ignored.
23.5.5. Attribute Reference
The following listing shows all the available attributes for an outbound-gateway
:
<int-jms:outbound-gateway
connection-factory="connectionFactory" (1)
correlation-key="" (2)
delivery-persistent="" (3)
destination-resolver="" (4)
explicit-qos-enabled="" (5)
extract-reply-payload="true" (6)
extract-request-payload="true" (7)
header-mapper="" (8)
message-converter="" (9)
priority="" (10)
receive-timeout="" (11)
reply-channel="" (12)
reply-destination="" (13)
reply-destination-expression="" (14)
reply-destination-name="" (15)
reply-pub-sub-domain="" (16)
reply-timeout="" (17)
request-channel="" (18)
request-destination="" (19)
request-destination-expression="" (20)
request-destination-name="" (21)
request-pub-sub-domain="" (22)
time-to-live="" (23)
requires-reply="" (24)
idle-reply-listener-timeout="" (25)
async=""> (26)
<int-jms:reply-listener /> (27)
</int-jms:outbound-gateway>
| 1 | Reference to a javax.jms.ConnectionFactory .
The default jmsConnectionFactory . |
||
| 2 | The name of a property that contains correlation data to correlate responses with replies.
If omitted, the gateway expects the responding system to return the value of the outbound JMSMessageID header in the JMSCorrelationID header.
If specified, the gateway generates a correlation ID and populates the specified property with it.
The responding system must echo back that value in the same property.
It can be set to JMSCorrelationID , in which case the standard header is used instead of a String property to hold the correlation data.
When you use a <reply-container/> , you must specify the correlation-key if you provide an explicit reply-destination .
Starting with version 4.0.1, this attribute also supports the value JMSCorrelationID* , which means that if the outbound message already has a JMSCorrelationID (mapped from the jms_correlationId ) header, it is used instead of generating a new one.
Note, the JMSCorrelationID* key is not allowed when you use a <reply-container/> , because the container needs to set up a message selector during initialization.
|
||
| 3 | A boolean value indicating whether the delivery mode should be DeliveryMode.PERSISTENT (true ) or DeliveryMode.NON_PERSISTENT (false ).
This setting takes effect only if explicit-qos-enabled is true . |
||
| 4 | A DestinationResolver .
The default is a DynamicDestinationResolver , which maps the destination name to a queue or topic of that name. |
||
| 5 | When set to true , it enables the use of quality of service attributes: priority , delivery-mode , and time-to-live . |
||
| 6 | When set to true (the default), the payload of the Spring Integration reply message is created from the JMS Reply message’s body (by using the MessageConverter ).
When set to false , the entire JMS message becomes the payload of the Spring Integration message. |
||
| 7 | When set to true (the default), the payload of the Spring Integration message is converted to a JMSMessage (by using the MessageConverter ).
When set to false , the entire Spring Integration Message is converted to the JMSMessage .
In both cases, the Spring Integration message headers are mapped to JMS headers and properties by using the HeaderMapper . |
||
| 8 | A HeaderMapper used to map Spring Integration message headers to and from JMS message headers and properties. |
||
| 9 | A reference to a MessageConverter for converting between JMS messages and the Spring Integration message payloads (or messages if extract-request-payload is false ).
The default is a SimpleMessageConverter . |
||
| 10 | The default priority of request messages.
Overridden by the message priority header, if present.
Its range is 0 to 9 .
This setting takes effect only if explicit-qos-enabled is true . |
||
| 11 | The time (in milliseconds) to wait for a reply.
The default is 5000 (five seconds). |
||
| 12 | The channel to which the reply message is sent. | ||
| 13 | A reference to a Destination , which is set as the JMSReplyTo header.
At most, only one of reply-destination , reply-destination-expression , or reply-destination-name is allowed.
If none is provided, a TemporaryQueue is used for replies to this gateway. |
||
| 14 | A SpEL expression evaluating to a Destination , which will be set as the JMSReplyTo header.
The expression can result in a Destination object or a String .
It is used by the DestinationResolver to resolve the actual Destination .
At most, only one of reply-destination , reply-destination-expression , or reply-destination-name is allowed.
If none is provided, a TemporaryQueue is used for replies to this gateway. |
||
| 15 | The name of the destination that is set as the JMSReplyTo header.
It is used by the DestinationResolver to resolve the actual Destination .
At most, only one of reply-destination , reply-destination-expression , or reply-destination-name is allowed.
If none is provided, a TemporaryQueue is used for replies to this gateway. |
||
| 16 | When set to true , it indicates that any reply Destination resolved by the DestinationResolver should be a Topic rather then a Queue . |
||
| 17 | The time the gateway waits when sending the reply message to the reply-channel .
This only has an effect if the reply-channel can block — such as a QueueChannel with a capacity limit that is currently full.
The default is infinity. |
||
| 18 | The channel on which this gateway receives request messages. | ||
| 19 | A reference to a Destination to which request messages are sent.
One of reply-destination , reply-destination-expression , or reply-destination-name is required.
You can use only one of those three attributes. |
||
| 20 | A SpEL expression evaluating to a Destination to which request messages are sent.
The expression can result in a Destination object or a String .
It is used by the DestinationResolver to resolve the actual Destination .
Oneof reply-destination , reply-destination-expression , or reply-destination-name is required.
You can use only one of those three attributes. |
||
| 21 | The name of the destination to which request messages are sent.
It is used by the DestinationResolver to resolve the actual Destination .
One of reply-destination , reply-destination-expression , or reply-destination-name is required.
You can use only one of those three attributes. |
||
| 22 | When set to true , it indicates that any request Destination resolved by the DestinationResolver should be a Topic rather then a Queue . |
||
| 23 | Specifies the message time to live.
This setting takes effect only if explicit-qos-enabled is true . |
||
| 24 | Specifies whether this outbound gateway must return a non-null value.
By default, this value is true , and a MessageTimeoutException is thrown when the underlying service does not return a value after the receive-timeout .
Note that, if the service is never expected to return a reply, it would be better to use a <int-jms:outbound-channel-adapter/> instead of a <int-jms:outbound-gateway/> with requires-reply="false" .
With the latter, the sending thread is blocked, waiting for a reply for the receive-timeout period. |
||
| 25 | When you use a <reply-listener /> , its lifecycle (start and stop) matches that of the gateway by default.
When this value is greater than 0 , the container is started on demand (when a request is sent).
The container continues to run until at least this time elapses with no requests being received (and until no replies are outstanding).
The container is started again on the next request.
The stop time is a minimum and may actually be up to 1.5x this value. |
||
| 26 | See Async Gateway. | ||
| 27 | When this element is included, replies are received by an asynchronous MessageListenerContainer rather than creating a consumer for each reply.
This can be more efficient in many cases. |
23.6. Mapping Message Headers to and from JMS Message
JMS messages can contain meta-information such as JMS API headers and simple properties.
You can map those to and from Spring Integration message headers by using JmsHeaderMapper
.
The JMS API headers are passed to the appropriate setter methods (such as setJMSReplyTo
), whereas other headers are copied to the general properties of the JMS Message.
JMS outbound gateway is bootstrapped with the default implementation of JmsHeaderMapper
, which will map standard JMS API Headers as well as primitive or String
message headers.
You could also provide a custom header mapper by using the header-mapper
attribute of inbound and outbound gateways.
Since version 4.0, the JMSPriority header is mapped to the standard priority header for inbound messages.
(previously, the priority header was only used for outbound messages).
To revert to the previous behavior (that is, to not map the inbound priority), set the mapInboundPriority property of DefaultJmsHeaderMapper to false .
|
Since version 4.3, the DefaultJmsHeaderMapper maps the standard correlationId header as a message
property by invoking its toString() method (correlationId is often a UUID , which is not supported by JMS).
On the inbound side, it is mapped as a String .
This is independent of the jms_correlationId header, which is mapped to and from the JMSCorrelationID header.
The JMSCorrelationID is generally used to correlate requests and replies, whereas the correlationId is often used to combine related messages into a group (such as with an aggregator or a resequencer).
|
Starting with version 5.1, the DefaultJmsHeaderMapper
can be configured for mapping inbound JMSDeliveryMode
and JMSExpiration
properties:
====
[source,java]
----
@Bean
public DefaultJmsHeaderMapper jmsHeaderMapper() {
DefaultJmsHeaderMapper mapper = new DefaultJmsHeaderMapper();
mapper.setMapInboundDeliveryMode(true)
mapper.setMapInboundExpiration(true)
return mapper;
}
----
====
These JMS properties are mapped to the JmsHeaders.DELIVERY_MODE
and JmsHeaders.EXPIRATION
Spring Message headers respectively.
23.7. Message Conversion, Marshalling, and Unmarshalling
If you need to convert the message, all JMS adapters and gateways let you provide a MessageConverter
by setting the message-converter
attribute.
To do so, provide the bean name of an instance of MessageConverter
that is available within the same ApplicationContext.
Also, to provide some consistency with marshaller and unmarshaller interfaces, Spring provides MarshallingMessageConverter
, which you can configure with your own custom marshallers and unmarshallers.
The following example shows how to do so
<int-jms:inbound-gateway request-destination="requestQueue"
request-channel="inbound-gateway-channel"
message-converter="marshallingMessageConverter"/>
<bean id="marshallingMessageConverter"
class="org.springframework.jms.support.converter.MarshallingMessageConverter">
<constructor-arg>
<bean class="org.bar.SampleMarshaller"/>
</constructor-arg>
<constructor-arg>
<bean class="org.bar.SampleUnmarshaller"/>
</constructor-arg>
</bean>
When you provide your own MessageConverter instance, it is still wrapped within the HeaderMappingMessageConverter .
This means that the 'extract-request-payload' and 'extract-reply-payload' properties can affect the actual objects passed to your converter.
The HeaderMappingMessageConverter itself delegates to a target MessageConverter while also mapping the Spring Integration MessageHeaders to JMS message properties and back again.
|
23.8. JMS-backed Message Channels
The channel adapters and gateways featured earlier are all intended for applications that integrate with other external systems.
The inbound options assume that some other system is sending JMS messages to the JMS destination, and the outbound options assume that some other system is receiving from the destination.
The other system may or may not be a Spring Integration application.
Of course, when sending a Spring Integration message instance as the body of the JMS message itself (with 'extract-payload' value set to false
), it is assumed that the other system is based on Spring Integration.
However, that is by no means a requirement.
That flexibility is one of the benefits of using a message-based integration option with the abstraction of “channels”( or destinations in the case of JMS).
Sometimes, both the producer and consumer for a given JMS Destination are intended to be part of the same application, running within the same process. You can accomplish this by using a pair of inbound and outbound channel adapters. The problem with that approach is that you need two adapters, even though, conceptually, the goal is to have a single message channel. A better option is supported as of Spring Integration version 2.0. Now it is possible to define a single “channel” when using the JMS namespace, as the following example shows:
<int-jms:channel id="jmsChannel" queue="exampleQueue"/>
The channel in the preceding example behaves much like a normal <channel/>
element from the main Spring Integration namespace.
It can be referenced by both the input-channel
and output-channel
attributes of any endpoint.
The difference is that this channel is backed by a JMS Queue instance named exampleQueue
.
This means that asynchronous messaging is possible between the producing and consuming endpoints.
However, unlike the simpler asynchronous message channels created by adding a <queue/>
element within a non-JMS <channel/>
element, the messages are not stored in an in-memory queue.
Instead, those messages are passed within a JMS message body, and the full power of the underlying JMS provider is then available for that channel.
Probably the most common rationale for using this alternative is to take advantage of the persistence made available by the store-and-forward approach of JMS messaging.
If configured properly, the JMS-backed message channel also supports transactions.
In other words, a producer would not actually write to a transactional JMS-backed channel if its send operation is part of a transaction that rolls back.
Likewise, a consumer would not physically remove a JMS message from the channel if the reception of that message is part of a transaction that rolls back.
Note that the producer and consumer transactions are separate in such a scenario.
This is significantly different than the propagation of a transactional context across a simple, synchronous <channel/>
element that has no <queue/>
child element.
Since the preceding example above references a JMS Queue instance, it acts as a point-to-point channel. If, on the other hand, you need publish-subscribe behavior, you can use a separate element and reference a JMS Topic instead. The following example shows how to do so:
<int-jms:publish-subscribe-channel id="jmsChannel" topic="exampleTopic"/>
For either type of JMS-backed channel, the name of the destination may be provided instead of a reference, as the following example shows:
<int-jms:channel id="jmsQueueChannel" queue-name="exampleQueueName"/>
<jms:publish-subscribe-channel id="jmsTopicChannel" topic-name="exampleTopicName"/>
In the preceding examples, the destination names are resolved by Spring’s default DynamicDestinationResolver
implementation, but you could provide any implementation of the DestinationResolver
interface.
Also, the JMS ConnectionFactory
is a required property of the channel, but, by default, the expected bean name would be jmsConnectionFactory
.
The following example provides both a custom instance for resolution of the JMS destination names and a different name for the ConnectionFactory
:
<int-jms:channel id="jmsChannel" queue-name="exampleQueueName"
destination-resolver="customDestinationResolver"
connection-factory="customConnectionFactory"/>
For the <publish-subscribe-channel />
, set the durable
attribute to true
for a durable subscription or subscription-shared
for a shared subscription (requires a JMS 2.0 broker and has been available since version 4.2).
Use subscription
to name the subscription.
23.9. Using JMS Message Selectors
With JMS message selectors, you can filter JMS Messages based on JMS headers as well as JMS properties.
For example, if you want to listen to messages whose custom JMS header property, myHeaderProperty
, equals something
, you can specify the following expression:
myHeaderProperty = 'something'
Message selector expressions are a subset of the SQL-92 conditional expression syntax and are defined as part of the Java Message Service specification (Version 1.1, April 12, 2002). Specifically, see chapter "3.8, Message Selection". It contains a detailed explanation of the expressions syntax.
You can specify the JMS message selector
attribute by using XML namespace configuration for the following Spring Integration JMS components:
-
JMS Channel
-
JMS Publish Subscribe Channel
-
JMS Inbound Channel Adapter
-
JMS Inbound Gateway
-
JMS Message-driven Channel Adapter
| You cannot reference message body values by using JMS Message selectors. |
23.10. JMS Samples
To experiment with these JMS adapters, check out the JMS samples available in the Spring Integration Samples Git repository at https://github.com/SpringSource/spring-integration-samples/tree/master/basic/jms.
That repository includes two samples. One provides inbound and outbound channel adapters, and the other provides inbound and outbound gateways. They are configured to run with an embedded ActiveMQ process, but you can modify the common.xml Spring application context file of each sample to support either a different JMS provider or a standalone ActiveMQ process.
In other words, you can split the configuration so that the inbound and outbound adapters run in separate JVMs.
If you have ActiveMQ installed, modify the brokerURL
property within the common.xml file to use tcp://localhost:61616
(instead of vm://localhost
).
Both of the samples accept input from stdin and echo back to stdout.
Look at the configuration to see how these messages are routed over JMS.
24. Mail Support
This section describes how to work with mail messages in Spring Integration.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mail</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-mail:5.1.5.RELEASE"
The javax.mail:javax.mail-api
must be included via vendor-specific implementation.
24.1. Mail-sending Channel Adapter
Spring Integration provides support for outbound email with the MailSendingMessageHandler
.
It delegates to a configured instance of Spring’s JavaMailSender
, as the following example shows:
JavaMailSender mailSender = context.getBean("mailSender", JavaMailSender.class);
MailSendingMessageHandler mailSendingHandler = new MailSendingMessageHandler(mailSender);
MailSendingMessageHandler
has various mapping strategies that use Spring’s MailMessage
abstraction.
If the received message’s payload is already a MailMessage
instance, it is sent directly.
Therefore, we generally recommend that you precede this consumer with a transformer for non-trivial MailMessage
construction requirements.
However, Spring Integration supports a few simple message mapping strategies.
For example, if the message payload is a byte array, that is mapped to an attachment.
For simple text-based emails, you can provide a string-based message payload.
In that case, a MailMessage
is created with that String
as the text content.
If you work with a message payload type whose toString()
method returns appropriate mail text content, consider adding Spring Integration’s ObjectToStringTransformer
prior to the outbound mail adapter (see the example in Configuring a Transformer with XML for more detail).
You can also configure the outbound MailMessage
with certain values from MessageHeaders
.
If available, values are mapped to the outbound mail’s properties, such as the recipients (To, Cc, and BCc), the from, the reply-to, and the subject.
The header names are defined by the following constants:
MailHeaders.SUBJECT
MailHeaders.TO
MailHeaders.CC
MailHeaders.BCC
MailHeaders.FROM
MailHeaders.REPLY_TO
MailHeaders also lets you override corresponding MailMessage values.
For example, if MailMessage.to is set to<EMAIL_ADDRESS>and the MailHeaders.TO message header is provided, it takes precedence and overrides the corresponding value in MailMessage .
|
24.2. Mail-receiving Channel Adapter
Spring Integration also provides support for inbound email with the MailReceivingMessageSource
.
It delegates to a configured instance of Spring Integration’s own MailReceiver
interface.
There are two implementations: Pop3MailReceiver
and ImapMailReceiver
.
The easiest way to instantiate either of these is by passing the 'uri' for a mail store to the receiver’s constructor, as the following example shows:
MailReceiver receiver = new Pop3MailReceiver("pop3://usr:pwd@localhost/INBOX");
Another option for receiving mail is the IMAP idle
command (if supported by your mail server).
Spring Integration provides the ImapIdleChannelAdapter
, which is itself a message-producing endpoint.
It delegates to an instance of the ImapMailReceiver
but enables asynchronous reception of mail messages.
The next section has examples of configuring both types of inbound channel adapter with Spring Integration’s namespace support in the 'mail' schema.
|
Normally, when the |
To<EMAIL_ADDRESS>From<EMAIL_ADDRESS>Subject: Test Email
something
With a simple MimeMessage
, getContent()
returns the mail body (something
in the preceding example).
Starting with version 2.2, the framework eagerly fetches IMAP messages and exposes them as an internal subclass of MimeMessage
.
This had the undesired side effect of changing the getContent()
behavior.
This inconsistency was further exacerbated by the Mail Mapping enhancement introduced in version 4.3, because, when a header mapper was provided, the payload was rendered by the IMAPMessage.getContent()
method.
This meant that the IMAP content differed, depending on whether or not a header mapper was provided.
Starting with version 5.0, messages originating from an IMAP source render the content in accordance with IMAPMessage.getContent()
behavior, regardless of whether a header mapper is provided.
If you do not use a header mapper and you wish to revert to the previous behavior of rendering only the body, set the simpleContent
boolean property on the mail receiver to true
.
This property now controls the rendering regardless of whether a header mapper is used.
It now allows body-only rendering when a header mapper is provided.
24.3. Inbound Mail Message Mapping
By default, the payload of messages produced by the inbound adapters is the raw MimeMessage
.
You can use that object to interrogate the headers and content.
Starting with version 4.3, you can provide a HeaderMapper<MimeMessage>
to map the headers to MessageHeaders
.
For convenience, Spring Integration provides a DefaultMailHeaderMapper
for this purpose.
It maps the following headers:
-
mail_from
: AString
representation of thefrom
address. -
mail_bcc
: AString
array containing thebcc
addresses. -
mail_cc
: AString
array containing thecc
addresses. -
mail_to
: AString
array containing theto
addresses. -
mail_replyTo
: AString
representation of thereplyTo
address. -
mail_subject
: The mail subject. -
mail_lineCount
: A line count (if available). -
mail_receivedDate
: The received date (if available). -
mail_size
: The mail size (if available). -
mail_expunged
: A boolean indicating if the message is expunged. -
mail_raw
: AMultiValueMap
containing all the mail headers and their values. -
mail_contentType
: The content type of the original mail message. -
contentType
: The payload content type (see below).
When message mapping is enabled, the payload depends on the mail message and its implementation.
Email contents are usually rendered by a DataHandler
within the MimeMessage
.
For a text/*
email, the payload is a String
and the contentType
header is the same as mail_contentType
.
For a messages with embedded javax.mail.Part
instances, the DataHandler
usually renders a Part
object.
These objects are not Serializable
and are not suitable for serialization with alternative technologies such as Kryo
.
For this reason, by default, when mapping is enabled, such payloads are rendered as a raw byte[]
containing the Part
data.
Examples of Part
are Message
and Multipart
.
The contentType
header is application/octet-stream
in this case.
To change this behavior and receive a Multipart
object payload, set embeddedPartsAsBytes
to false
on MailReceiver
.
For content types that are unknown to the DataHandler
, the contents are rendered as a byte[]
with a contentType
header of application/octet-stream
.
When you do not provide a header mapper, the message payload is the MimeMessage
presented by javax.mail
.
The framework provides a MailToStringTransformer
that you can use to convert the message by using a strategy to convert the mail contents to a String
.
This is also available by using the XML namespace, as the following example shows:
<int-mail:mail-to-string-transformer ... >
The following example does the same thing with Java configuration:
@Bean
@Transformer(inputChannel="...", outputChannel="...")
public Transformer transformer() {
return new MailToStringTransformer();
}
The following example does the same thing with the Java DSL:
...
.transform(Mail.toStringTransformer())
...
Starting with version 4.3, the transformer handles embedded Part
instances (as well as Multipart
instances, which were handled previously).
The transformer is a subclass of AbstractMailTransformer
that maps the address and subject headers from the preceding list.
If you wish to perform some other transformation on the message, consider subclassing AbstractMailTransformer
.
24.4. Mail Namespace Support
Spring Integration provides a namespace for mail-related configuration. To use it, configure the following schema locations:
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-mail="http://www.springframework.org/schema/integration/mail"
xsi:schemaLocation="http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/mail
https://www.springframework.org/schema/integration/mail/spring-integration-mail.xsd">
To configure an outbound channel adapter, provide the channel from which to receive and the MailSender, as the following example shows:
<int-mail:outbound-channel-adapter channel="outboundMail"
mail-sender="mailSender"/>
Alternatively, you can provide the host, username, and password, as the following example shows:
<int-mail:outbound-channel-adapter channel="outboundMail"
host="somehost" username="someuser" password="somepassword"/>
Starting with version 5.1.3, the host
, username
ane mail-sender
can be omitted, if java-mail-properties
is provided.
However the host
and username
has to be configured with appropriate Java mail properties, e.g. for SMTP:
<EMAIL_ADDRESS>mail.smtp.host=smtp.gmail.com
mail.smtp.port=587
As with any outbound Channel Adapter, if the referenced channel is a PollableChannel , you should provide a <poller> element (see Endpoint Namespace Support).
|
When you use the namespace support, you can also use a header-enricher
message transformer.
Doing so simplifies the application of the headers mentioned earlier to any message prior to sending to the mail outbound channel adapter.
The following example assumes the payload is a Java bean with appropriate getters for the specified properties, but you can use any SpEL expression:
<int-mail:header-enricher input-channel="expressionsInput" default-overwrite="false">
<int-mail:to expression="payload.to"/>
<int-mail:cc expression="payload.cc"/>
<int-mail:bcc expression="payload.bcc"/>
<int-mail:from expression="payload.from"/>
<int-mail:reply-to expression="payload.replyTo"/>
<int-mail:subject expression="payload.subject" overwrite="true"/>
</int-mail:header-enricher>
Alternatively, you can use the value
attribute to specify a literal.
You also can specify default-overwrite
and individual overwrite
attributes to control the behavior with existing headers.
To configure an inbound channel adapter, you have the choice between polling or event-driven (assuming your mail server supports IMAP idle
— if not, then polling is the only option).
A polling channel adapter requires the store URI and the channel to which to send inbound messages.
The URI may begin with pop3
or imap
.
The following example uses an imap
URI:
<int-mail:inbound-channel-adapter id="imapAdapter"
<EMAIL_ADDRESS>java-mail-properties="javaMailProperties"
channel="receiveChannel"
should-delete-messages="true"
should-mark-messages-as-read="true"
auto-startup="true">
<int:poller max-messages-per-poll="1" fixed-rate="5000"/>
</int-mail:inbound-channel-adapter>
If you do have IMAP idle
support, you may want to configure the imap-idle-channel-adapter
element instead.
Since the idle
command enables event-driven notifications, no poller is necessary for this adapter.
It sends a message to the specified channel as soon as it receives the notification that new mail is available.
The following example configures an IMAP idle
mail channel:
<int-mail:imap-idle-channel-adapter id="customAdapter"
<EMAIL_ADDRESS>channel="receiveChannel"
auto-startup="true"
should-delete-messages="false"
should-mark-messages-as-read="true"
java-mail-properties="javaMailProperties"/>
You can provide javaMailProperties
by creating and populating a regular java.utils.Properties
object — for example, by using the util
namespace provided by Spring.
| If your username contains the '@' character, use '%40' instead of '@' to avoid parsing errors from the underlying JavaMail API. |
The following example shows how to configure a java.util.Properties
object:
<util:properties id="javaMailProperties">
<prop key="mail.imap.socketFactory.class">javax.net.ssl.SSLSocketFactory</prop>
<prop key="mail.imap.socketFactory.fallback">false</prop>
<prop key="mail.store.protocol">imaps</prop>
<prop key="mail.debug">false</prop>
</util:properties>
By default, the ImapMailReceiver
searches for messages based on the default SearchTerm
, which is all mail messages that:
-
Are RECENT (if supported)
-
Are NOT ANSWERED
-
Are NOT DELETED
-
Are NOT SEEN
-
hHave not been processed by this mail receiver (enabled by the use of the custom USER flag or simply NOT FLAGGED if not supported)
The custom user flag is spring-integration-mail-adapter
, but you can configure it.
Since version 2.2, the SearchTerm
used by the ImapMailReceiver
is fully configurable with SearchTermStrategy
, which you can inject by using the search-term-strategy
attribute.
SearchTermStrategy
is a strategy interface with a single method that lets you create an instance of the SearchTerm
used by the ImapMailReceiver
.
The following listing shows the SearchTermStrategy
interface:
public interface SearchTermStrategy {
SearchTerm generateSearchTerm(Flags supportedFlags, Folder folder);
}
The following example relies TestSearchTermStrategy
rather than the default SearchTermStrategy
:
<mail:imap-idle-channel-adapter id="customAdapter"
store-uri="imap:something"
…
search-term-strategy="searchTermStrategy"/>
<bean id="searchTermStrategy"
class="o.s.i.mail.config.ImapIdleChannelAdapterParserTests.TestSearchTermStrategy"/>
See Marking IMAP Messages When \Recent
Is Not Supported for information about message flagging.
|
Important: IMAP PEEK
Starting with version 4.1.1, the IMAP mail receiver uses the |
24.4.1. IMAP idle
and Lost Connections
When using an IMAP idle
channel adapter, connections to the server may be lost (for example, through network failure) and, since the JavaMail documentation explicitly states that the actual IMAP API is experimental, it is important to understand the differences in the API and how to deal with them when configuring IMAP idle
adapters.
Currently, Spring Integration mail adapters were tested with JavaMail 1.4.1 and JavaMail 1.4.3.
Depending on which one is used, you must pay special attention to some of the JavaMail properties that need to be set with regard to auto-reconnect.
| The following behavior was observed with Gmail but should provide you with some tips on how to solve re-connect issue with other providers. However feedback is always welcome. Again, the following notes are based on Gmail. |
With JavaMail 1.4.1, if you set the mail.imaps.timeout
property to a relatively short period of time (approximately 5 min in our testing), IMAPFolder.idle()
throws FolderClosedException
after this timeout.
However, if this property is not set (it should be indefinite) the IMAPFolder.idle()
method never returns and never throws an exception.
It does, however, reconnect automatically if the connection was lost for a short period of time (under 10 min in our testing).
However, if the connection was lost for a long period of time (over 10 min), IMAPFolder.idle()
, does not throw FolderClosedException
and does not re-establish the connection, and remains in the blocked state indefinitely, thus leaving you no possibility to reconnect without restarting the adapter.
Consequently, the only way to make re-connecting work with JavaMail 1.4.1 is to set the mail.imaps.timeout
property explicitly to some value, but it also means that such value should be relatively short (under 10 min) and the connection should be re-established relatively quickly.
Again, it may be different with providers other than Gmail.
With JavaMail 1.4.3 introduced significant improvements to the API, ensuring that there is always a condition that forces the IMAPFolder.idle()
method to return StoreClosedException
or FolderClosedException
or to simply return, thus letting you proceed with auto-reconnecting.
Currently auto-reconnecting runs infinitely making attempts to reconnect every ten seconds.
In both configurations, channel and should-delete-messages are required attributes.
You should understand why should-delete-messages is required.
The issue is with the POP3 protocol, which does not have any knowledge of messages that were read.
It can only know what has been read within a single session.
This means that, when your POP3 mail adapter runs, emails are successfully consumed as as they become available during each poll
and no single email message is delivered more then once.
However, as soon as you restart your adapter and begin a new session, all the email messages that might have been retrieved in the previous session are retrieved again.
That is the nature of POP3.
Some might argue that should-delete-messages should be true by default.
In other words, there are two valid and mutually exclusive use that make it very hard to pick a single best default.
You may want to configure your adapter as the only email receiver, in which case you want to be able to restart your adapter without fear that previously delivered messages are not delivered again.
In this case, setting should-delete-messages to true would make the most sense.
However, you may have another use case where you may want to have multiple adapters monitor email servers and their content.
In other words, you want to 'peek but not touch'.
Then setting should-delete-messages to false is much more appropriate.
So since it is hard to choose what should be the right default value for the should-delete-messages attribute, we made it a required attribute to be set by you.
Leaving it up to you also means that you are less likely to end up with unintended behavior.
|
When configuring a polling email adapter’s should-mark-messages-as-read attribute, you should be aware of the protocol you are configuring to retrieve messages.
For example, POP3 does not support this flag, which means setting it to either value has no effect, as messages are not marked as read.
|
|
You should understand that that these actions (marking messages read and deleting messages) are performed after the messages are received but before they are processed. This can cause messages to be lost. You may wish to consider using transaction synchronization instead. See Transaction Synchronization. |
The <imap-idle-channel-adapter/>
also accepts the 'error-channel' attribute.
If a downstream exception is thrown and an 'error-channel' is specified, a MessagingException
message containing the failed message and the original exception is sent to this channel.
Otherwise, if the downstream channels are synchronous, any such exception is logged as a warning by the channel adapter.
Beginning with the 3.0 release, the IMAP idle adapter emits application events (specifically ImapIdleExceptionEvent instances) when exceptions occur.
This allows applications to detect and act on those exceptions.
You can obtain the events by using an <int-event:inbound-channel-adapter> or any ApplicationListener configured to receive an ImapIdleExceptionEvent or one of its super classes.
|
24.5. Marking IMAP Messages When \Recent
Is Not Supported
If shouldMarkMessagesAsRead
is true, the IMAP adapters set the \Seen
flag.
In addition, when an email server does not support the \Recent
flag, the IMAP adapters mark messages with a user flag (by default, spring-integration-mail-adapter
), as long as the server supports user flags.
If not, Flag.FLAGGED
is set to true
.
These flags are applied regardless of the shouldMarkMessagesRead
setting.
As discussed in [search-term], the default SearchTermStrategy
ignore messages that are so flagged.
Starting with version 4.2.2, you can set the name of the user flag by using setUserFlag
on the MailReceiver
.
Doing so lets multiple receivers use a different flag (as long as the mail server supports user flags).
The user-flag
attribute is available when configuring the adapter with the namespace.
24.6. Email Message Filtering
Very often, you may encounter a requirement to filter incoming messages (for example, you want to read only emails that have 'Spring Integration' in the Subject
line).
You can accomplish this by connecting an inbound mail adapter with an expression-based Filter
.
Although it would work, there is a downside to this approach.
Since messages would be filtered after going through the inbound mail adapter, all such messages would be marked as read (SEEN
) or unread (depending on the value of should-mark-messages-as-read
attribute).
However, in reality, it be more useful to mark messages as SEEN
only if they pass the filtering criteria.
This is similar to looking at your email client while scrolling through all the messages in the preview pane, but only flagging messages that were actually opened and read as SEEN
.
Spring Integration 2.0.4 introduced the mail-filter-expression
attribute on inbound-channel-adapter
and imap-idle-channel-adapter
.
This attribute lets you provide an expression that is a combination of SpEL and a regular expression.
For example if you would like to read only emails that contain 'Spring Integration' in the subject line, you would configure the mail-filter-expression
attribute like as follows: mail-filter-expression="subject matches '(?i).Spring Integration."
.
Since javax.mail.internet.MimeMessage
is the root context of the SpEL evaluation context, you can filter on any value available through MimeMessage
, including the actual body of the message.
This one is particularly important, since reading the body of the message typically results in such messages being marked as SEEN
by default.
However, since we now set the PEEK
flag of every incoming message to 'true', only messages that were explicitly marked as SEEN
are marked as read.
So, in the following example, only messages that match the filter expression are output by this adapter and only those messages are marked as read:
<int-mail:imap-idle-channel-adapter id="customAdapter"
<EMAIL_ADDRESS>channel="receiveChannel"
should-mark-messages-as-read="true"
java-mail-properties="javaMailProperties"
mail-filter-expression="subject matches '(?i).*Spring Integration.*'"/>
In the preceding example, thanks to the mail-filter-expression
attribute, only messages that contain 'Spring Integration' in the subject line are produced by this adapter.
Another reasonable question is what happens on the next poll or idle event or what happens when such an adapter is restarted.
Can there be duplication of massages to be filtered? In other words, if, on the last retrieval where you had five new messages and only one passed the filter, what would happen with the other four?
Would they go through the filtering logic again on the next poll or idle? After all, they were not marked as SEEN
.
The answer is no.
They would not be subject to duplicate processing due to another flag (RECENT
) that is set by the email server and is used by the Spring Integration mail search filter.
Folder implementations set this flag to indicate that this message is new to this folder.
That is, it has arrived since the last time this folder was opened.
In other words, while our adapter may peek at the email, it also lets the email server know that such email was touched and should therefore be marked as RECENT
by the email server.
24.7. Transaction Synchronization
Transaction synchronization for inbound adapters lets you take different actions after a transaction commits or rolls back.
You can enable transaction synchronization by adding a <transactional/>
element to the poller for the polled <inbound-adapter/>
or to the <imap-idle-inbound-adapter/>
.
Even if there is no 'real' transaction involved, you can still enable this feature by using a PseudoTransactionManager
with the <transactional/>
element.
For more information, see Transaction Synchronization.
Because of the many different mail servers and specifically the limitations that some have, at this time we provide only a strategy for these transaction synchronizations. You can send the messages to some other Spring Integration components or invoke a custom bean to perform some action. For example, to move an IMAP message to a different folder after the transaction commits, you might use something similar to the following:
<int-mail:imap-idle-channel-adapter id="customAdapter"
<EMAIL_ADDRESS>channel="receiveChannel"
auto-startup="true"
should-delete-messages="false"
java-mail-properties="javaMailProperties">
<int:transactional synchronization-factory="syncFactory"/>
</int-mail:imap-idle-channel-adapter>
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit expression="@syncProcessor.process(payload)"/>
</int:transaction-synchronization-factory>
<bean id="syncProcessor" class="thing1.thing2.Mover"/>
The following example shows what the Mover
class might look like:
public class Mover {
public void process(MimeMessage message) throws Exception{
Folder folder = message.getFolder();
folder.open(Folder.READ_WRITE);
String messageId = message.getMessageID();
Message[] messages = folder.getMessages();
FetchProfile contentsProfile = new FetchProfile();
contentsProfile.add(FetchProfile.Item.ENVELOPE);
contentsProfile.add(FetchProfile.Item.CONTENT_INFO);
contentsProfile.add(FetchProfile.Item.FLAGS);
folder.fetch(messages, contentsProfile);
// find this message and mark for deletion
for (int i = 0; i < messages.length; i++) {
if (((MimeMessage) messages[i]).getMessageID().equals(messageId)) {
messages[i].setFlag(Flags.Flag.DELETED, true);
break;
}
}
Folder somethingFolder = store.getFolder("SOMETHING"));
somethingFolder.appendMessages(new MimeMessage[]{message});
folder.expunge();
folder.close(true);
somethingFolder.close(false);
}
}
| For the message to be still available for manipulation after the transaction, should-delete-messages must be set to 'false'. |
24.8. Configuring channel adapters with the Java DSL
To configure mail mail component in Java DSL, the framework provides a o.s.i.mail.dsl.Mail
factory, which can be used like this:
@SpringBootApplication
public class MailApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MailApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow imapMailFlow() {
return IntegrationFlows
.from(Mail.imapInboundAdapter("imap://user:pw@host:port/INBOX")
.searchTermStrategy(this::fromAndNotSeenTerm)
.userFlag("testSIUserFlag")
.simpleContent(true)
.javaMailProperties(p -> p.put("mail.debug", "false")),
e -> e.autoStartup(true)
.poller(p -> p.fixedDelay(1000)))
.channel(MessageChannels.queue("imapChannel"))
.get();
}
@Bean
public IntegrationFlow sendMailFlow() {
return IntegrationFlows.from("sendMailChannel")
.enrichHeaders(Mail.headers()
.subjectFunction(m -> "foo")
.from("foo@bar")
.toFunction(m -> new String[] { "bar@baz" }))
.handle(Mail.outboundAdapter("gmail")
.port(smtpServer.getPort())
.credentials("user", "pw")
.protocol("smtp")),
e -> e.id("sendMailEndpoint"))
.get();
}
}
25. MongoDb Support
Version 2.1 introduced support for MongoDB: a “high-performance, open source, document-oriented database”.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mongodb</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-mongodb:5.1.5.RELEASE"
To download, install, and run MongoDB, see the MongoDB documentation.
25.1. Connecting to MongoDb
To begin interacting with MongoDB, you first need to connect to it.
Spring Integration builds on the support provided by another Spring project, Spring Data MongoDB.
It provides a factory class called MongoDbFactory
, which simplifies integration with the MongoDB Client API.
25.1.1. Using MongoDbFactory
To connect to MongoDB you can use an implementation of the MongoDbFactory
interface.
The following listing shows the MongoDbFactory
interface:
public interface MongoDbFactory {
/**
* Creates a default {@link DB} instance.
*
* @return the DB instance
* @throws DataAccessException
*/
DB getDb() throws DataAccessException;
/**
* Creates a {@link DB} instance to access the database with the given name.
*
* @param dbName must not be {@literal null} or empty.
*
* @return the DB instance
* @throws DataAccessException
*/
DB getDb(String dbName) throws DataAccessException;
}
The following example shows how to use SimpleMongoDbFactory
, the out-of-the-box implementation, in Java:
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
The following example shows how to use SimpleMongoDbFactory
in XML configuration:
<bean id="mongoDbFactory" class="o.s.data.mongodb.core.SimpleMongoDbFactory">
<constructor-arg>
<bean class="com.mongodb.Mongo"/>
</constructor-arg>
<constructor-arg value="test"/>
</bean>
SimpleMongoDbFactory
takes two arguments: a Mongo
instance and a String
that specifies the name of the database.
If you need to configure properties such as host
, port
, and others, you can pass those by using one of the constructors provided by the underlying Mongo
class.
For more information on how to configure MongoDB, see the Spring-Data-MongoDB reference.
25.2.
Doing so can be useful when dealing with components that have the ability to buffer messages (QueueChannel
, aggregator
, resequencer
, and others.) if reliability is a concern.
In Spring Integration, the MessageStore
strategy also provides the foundation for the claim check pattern, which is described in EIP as well.
Spring Integration’s MongoDB module provides the MongoDbMessageStore
, which is an implementation of both the MessageStore
strategy (mainly used by the claim check pattern) and the MessageGroupStore
strategy (mainly used by the aggregator and resequencer patterns).
The following example configures a MongoDbMessageStore
to use a QueueChannel
and an aggregator
:
<bean id="mongoDbMessageStore" class="o.s.i.mongodb.store.MongoDbMessageStore">
<constructor-arg ref="mongoDbFactory"/>
</bean>
<int:channel id="somePersistentQueueChannel">
<int:queue message-store="mongoDbMessageStore"/>
<int:channel>
<int:aggregator input-channel="inputChannel" output-channel="outputChannel"
message-store="mongoDbMessageStore"/>
The preceding example is a simple bean configuration, and it expects a MongoDbFactory
as a constructor argument.
The MongoDbMessageStore
expands the Message
as a Mongo document with all nested properties by using the Spring Data Mongo mapping mechanism.
It is useful when you need to have access to the payload
or headers
for auditing or analytics — for example, against stored messages.
The MongoDbMessageStore uses a custom MappingMongoConverter implementation to store Message instances as MongoDB documents, and there are some limitations for the properties (payload and header values) of the Message .
For example, there is no ability to configure custom converters for complex domain payload instances or header values.
There is also no way to provide a custom MongoTemplate (or MappingMongoConverter ).
To achieve these capabilities, an alternative MongoDB MessageStore implementation has been introduced (we describe it next).
|
Spring Integration 3.0 introduced the ConfigurableMongoDbMessageStore
.
It implements both the MessageStore
and MessageGroupStore
interfaces.
This class can receive, as a constructor argument, a MongoTemplate
, with which you can, for example, configure a custom WriteConcern
.
Another constructor requires a MappingMongoConverter
and a MongoDbFactory
, which lets you provide some custom conversions for Message
instances and their properties.
Note that, by default, the ConfigurableMongoDbMessageStore
uses standard Java serialization to write and read Message
instances to and from MongoDB (see MongoDbMessageBytesConverter
) and relies on default values for other properties from MongoTemplate
.
It builds a MongoTemplate
from the provided MongoDbFactory
and MappingMongoConverter
.
The default name for the collection stored by the ConfigurableMongoDbMessageStore
is configurableStoreMessages
.
We recommend using this implementation to create robust and flexible solutions when messages contain complex data types.
25.2.1. MongoDB Channel Message Store
Version 4.0 introduced the new MongoDbChannelMessageStore
.
It is an optimized MessageGroupStore
for use in QueueChannel
instances.
With priorityEnabled = true
, you can use it in <int:priority-queue>
instances to achieve priority-order polling for persisted messages.
The priority MongoDB document field is populated from the IntegrationMessageHeaderAccessor.PRIORITY
(priority
) message header.
In addition, all MongoDB MessageStore
instances now have a sequence
field for MessageGroup
documents.
The sequence
value is the result of an $inc
operation for a simple sequence
document from the same collection, which is created on demand.
The sequence
field is used in poll
operations to provide first-in-first-out (FIFO) message order (within priority, if configured) when messages are stored within the same millisecond.
We do not recommend using the same MongoDbChannelMessageStore bean for priority and non-priority, because the priorityEnabled option applies to the entire store.
However, the same collection can be used for both MongoDbChannelMessageStore types, because message polling from the store is sorted and uses indexes.
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.mongodb.store.MongoDbChannelMessageStore">
<constructor-arg name="mongoDbFactory" ref="mongoDbFactory"/>
</bean>
<int:channel id="queueChannel">
<int:queue message-store="store"/>
</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>
25.2.2. MongoDB Metadata Store
Spring Integration 4.2 introduced a new MongoDB-based MetadataStore
(see Metadata Store) implementation.
You can use the MongoDbMetadataStore
to maintain metadata state across application restarts.
You can use this new MetadataStore
implementation with adapters such as:
To instruct these adapters to use the new MongoDbMetadataStore
, declare a Spring bean with a bean name of metadataStore
.
The feed inbound channel adapter automatically picks up and use the declared MongoDbMetadataStore
.
The following example shows how to declare a bean with a name of metadataStore
:
@Bean
public MetadataStore metadataStore(MongoDbFactory factory) {
return new MongoDbMetadataStore(factory, "integrationMetadataStore");
}
The MongoDbMetadataStore
also implements ConcurrentMetadataStore
, letting it be reliably shared across multiple application instances, where only one instance is allowed to store or modify a key’s value.
All these operations are atomic, thanks to MongoDB guarantees.
25.3. MongoDB Inbound Channel Adapter
The MongoDB inbound channel adapter is a polling consumer that reads data from MongoDB and sends it as a Message
payload.
The following example shows how to configure a MongoDB inbound channel adapter:
<int-mongodb:inbound-channel-adapter id="mongoInboundAdapter"
channel="replyChannel"
query="{'name' : 'Bob'}"
entity-class="java.lang.Object"
auto-startup="false">
<int:poller fixed-rate="100"/>
</int-mongodb:inbound-channel-adapter>
As the preceding configuration shows, you configure a MongoDb inbound channel adapter by using the inbound-channel-adapter
element and providing values for various attributes, such as:
-
query
: A JSON query (see MongoDB Querying) -
query-expression
: A SpEL expression that is evaluated to a JSON query string (as thequery
attribute above) or to an instance ofo.s.data.mongodb.core.query.Query
. Mutually exclusive with thequery
attribute. -
entity-class
: The type of the payload object. If not supplied, acom.mongodb.DBObject
is returned. -
collection-name
orcollection-name-expression
: Identifies the name of the MongoDB collection to use. -
mongodb-factory
: Reference to an instance ofo.s.data.mongodb.MongoDbFactory
-
mongo-template
: Reference to an instance ofo.s.data.mongodb.core.MongoTemplate
-
Other attributes that are common across all other inbound adapters (such as 'channel').
You cannot set both mongo-template and mongodb-factory .
|
The preceding example is relatively simple and static, since it has a literal value for the query
and uses the default name for a collection
.
Sometimes, you may need to change those values at runtime, based on some condition.
To do so, use their -expression
equivalents (query-expression
and collection-name-expression
), where the provided expression can be any valid SpEL expression.
Also, you may wish to do some post-processing to the successfully processed data that was read from the MongoDB. For example; you may want to move or remove a document after it has been processed. You can do so by using that transaction synchronization feature Spring Integration 2.2 added, as the following example shows:
<int-mongodb:inbound-channel-adapter id="mongoInboundAdapter"
channel="replyChannel"
query-expression="new BasicQuery('{''name'' : ''Bob''}').limit(100)"
entity-class="java.lang.Object"
auto-startup="false">
<int:poller fixed-rate="200" max-messages-per-poll="1">
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-mongodb:inbound-channel-adapter>
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit
expression="@documentCleaner.remove(#mongoTemplate, payload, headers.mongo_collectionName)"
channe="someChannel"/>
</int:transaction-synchronization-factory>
<bean id="documentCleaner" class="thing1.thing2.DocumentCleaner"/>
<bean id="transactionManager" class="o.s.i.transaction.PseudoTransactionManager"/>
The following example shows the DocumentCleaner
referenced in the preceding example:
public class DocumentCleaner {
public void remove(MongoOperations mongoOperations, Object target, String collectionName) {
if (target instanceof List<?>){
List<?> documents = (List<?>) target;
for (Object document : documents) {
mongoOperations.remove(new BasicQuery(JSON.serialize(document)), collectionName);
}
}
}
}
You can declare your poller to be transactional by using the transactional
element.
This element can reference a real transaction manager (for example, if some other part of your flow invokes JDBC).
If you do not have a “real” transaction, you can use an instance of o.s.i.transaction.PseudoTransactionManager
, which is an implementation of Spring’s PlatformTransactionManager
and enables the use of the transaction synchronization features of the Mongo adapter when there is no actual transaction.
| Doing so does not make MongoDB itself transactional. It lets the synchronization of actions be taken before or after success (commit) or after failure (rollback). |
Once your poller is transactional, you can set an instance of the o.s.i.transaction.TransactionSynchronizationFactory
on the transactional
element.
A TransactionSynchronizationFactory
creates an instance of the TransactionSynchronization
.

Expressions for before-commit, after-commit, and after-rollback events are supported, together with a channel for each event where the evaluation result (if any) is sent.
For each child element, you can specify expression
and channel
attributes.
If only the channel
attribute is present, the received message is sent there as part of the particular synchronization scenario.
If only the expression
attribute is present and the result of an expression is a non-null value, a message with the result as the payload is generated and sent to a default channel (NullChannel
) and appears in the logs (on the DEBUG
level).
If you want the evaluation result to go to a specific channel, add a channel
attribute.
If the result of an expression is null or void, no message is generated.
For more information about transaction synchronization, see Transaction Synchronization.
25.4. MongoDB Outbound Channel Adapter
The MongoDB outbound channel adapter lets you write the message payload to a MongoDB document store, as the following example shows:
<int-mongodb:outbound-channel-adapter id="fullConfigWithCollectionExpression"
collection-name="myCollection"
mongo-converter="mongoConverter"
mongodb-factory="mongoDbFactory" />
As the preceding configuration shows, you can configure a MongoDB outbound channel adapter by using the outbound-channel-adapter
element, providing values for various attributes, such as:
-
collection-name
orcollection-name-expression
: Identifies the name of the MongoDb collection to use. -
mongo-converter
: Reference to an instance ofo.s.data.mongodb.core.convert.MongoConverter
that assists with converting a raw Java object to a JSON document representation. -
mongodb-factory
: Reference to an instance ofo.s.data.mongodb.MongoDbFactory
. -
mongo-template
: Reference to an instance ofo.s.data.mongodb.core.MongoTemplate
. NOTE: you can not have both mongo-template and mongodb-factory set. -
Other attributes that are common across all other inbound adapters (such as 'channel').
The preceding example is relatively simple and static, since it has a literal value for the collection-name
.
Sometimes, you may need to change this value at runtime, based on some condition.
To do that, use collection-name-expression
, where the provided expression is any valid SpEL expression.
25.5. MongoDB Outbound Gateway
Version 5.0 introduced the MongoDB outbound gateway. It allows you query a database by sending a message to its request channel. The gateway then send the response to the reply channel. You can use the message payload and headers to specify the query and the collection name, as the following example shows:
<int-mongodb:outbound-gateway id="gatewayQuery"
mongodb-factory="mongoDbFactory"
mongo-converter="mongoConverter"
query="{firstName: 'Bob'}"
collection-name="myCollection"
request-channel="in"
reply-channel="out"
entity-class="org.springframework.integration.mongodb.test.entity$Person"/>
You can use the following attributes with a MongoDB outbound Gateway:
-
collection-name
orcollection-name-expression
: Identifies the name of the MongoDB collection to use. -
mongo-converter
: Reference to an instance ofo.s.data.mongodb.core.convert.MongoConverter
that assists with converting a raw Java object to a JSON document representation. -
mongodb-factory
: Reference to an instance ofo.s.data.mongodb.MongoDbFactory
. -
mongo-template
: Reference to an instance ofo.s.data.mongodb.core.MongoTemplate
. NOTE: you can not set bothmongo-template
andmongodb-factory
. -
entity-class
: The fully qualified name of the entity class to be passed to thefind(..)
andfindOne(..)
methods in MongoTemplate. If this attribute is not provided, the default value isorg.bson.Document
. -
query
orquery-expression
: Specifies the MongoDB query. See the MongoDB documentation for more query samples. -
collection-callback
: Reference to an instance oforg.springframework.data.mongodb.core.CollectionCallback
. Preferable an instance ofo.s.i.mongodb.outbound.MessageCollectionCallback
since 5.0.11 with the request message context. See its Javadocs for more information. NOTE: You can not have bothcollection-callback
and any of the query attributes.
25.5.1. 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 MongoDbJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MongoDbJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private MongoDbFactory mongoDbFactory;
@Bean
@ServiceActivator(inputChannel = "requestChannel")
public MessageHandler mongoDbOutboundGateway() {
MongoDbOutboundGateway gateway = new MongoDbOutboundGateway(this.mongoDbFactory);
gateway.setCollectionNameExpressionString("'myCollection'");
gateway.setQueryExpressionString("'{''name'' : ''Bob''}'");
gateway.setExpectSingleResult(true);
gateway.setEntityClass(Person.class);
gateway.setOutputChannelName("replyChannel");
return gateway;
}
@Bean
@ServiceActivator(inputChannel = "replyChannel")
public MessageHandler handler() {
return message -> System.out.println(message.getPayload());
}
}
25.5.2. Configuring with the Java DSL
The following Spring Boot application show an example of how to configure the outbound gateway with the Java DSL:
@SpringBootApplication
public class MongoDbJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MongoDbJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private MongoDbFactory;
@Autowired
private MongoConverter;
@Bean
public IntegrationFlow gatewaySingleQueryFlow() {
return f -> f
.handle(queryOutboundGateway())
.channel(c -> c.queue("retrieveResults"));
}
private MongoDbOutboundGatewaySpec queryOutboundGateway() {
return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
.query("{name : 'Bob'}")
.collectionNameFunction(m -> m.getHeaders().get("collection"))
.expectSingleResult(true)
.entityClass(Person.class);
}
}
As an alternate to the query
and query-expression
properties, you can specify other database operations by using the collectionCallback
property as a reference to the MessageCollectionCallback
functional interface implementation.
The following example specifies a count operation:
private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway() {
return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
.collectionCallback((collection, requestMessage) -> collection.count())
.collectionName("myCollection");
}
26. MQTT Support
Spring Integration provides inbound and outbound channel adapters to support the Message Queueing Telemetry Transport (MQTT) protocol.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-mqtt:5.1.5.RELEASE"
The current implementation uses the Eclipse Paho MQTT Client library.
Configuration of both adapters is achieved using the DefaultMqttPahoClientFactory
.
Refer to the Paho documentation for more information about configuration options.
We recommend configuring an MqttConnectOptions object and injecting it into the factory, instead of setting the (deprecated) options on the factory itself.
|
26.1. Inbound (Message-driven) Channel Adapter
The inbound channel adapter is implemented by the MqttPahoMessageDrivenChannelAdapter
.
For convenience, you can configure it by using the namespace.
A minimal configuration might be as follows: