Skip to content

Commit 4d7008e

Browse files
artembilanartembilan
committed
Add Reactive TX Manager support
* Change `TransactionInterceptorBuilder` and `TransactionHandleMessageAdvice` to reply on a generic `TransactionManager`, so we can configure reactive one as well * Change a `<transactional>` XML element to support a generic `TransactionManager` reference, so we can configure reactive one as well * Support `adviceChain` configuration for the `ReactiveMessageHandler` in the `ConsumerEndpointFactoryBean` * Attemp to apply reactive transaction for the Reactive MongoDB channel adapter
1 parent 1277531 commit 4d7008e

File tree

8 files changed

+61
-47
lines changed

8 files changed

+61
-47
lines changed

‎spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -200,10 +200,13 @@ public void afterPropertiesSet() {
200200
}
201201

202202
if (!(this.handler instanceof ReactiveMessageHandlerAdapter)) {
203-
adviceChain();
203+
this.handler = adviceChain(this.handler);
204204
}
205205
else if (!CollectionUtils.isEmpty(this.adviceChain)) {
206-
LOGGER.warn("the advice chain cannot be applied to a 'ReactiveMessageHandler'");
206+
ReactiveMessageHandler reactiveMessageHandler =
207+
((ReactiveMessageHandlerAdapter) this.handler).getDelegate();
208+
reactiveMessageHandler = adviceChain(reactiveMessageHandler);
209+
this.handler = new ReactiveMessageHandlerAdapter(reactiveMessageHandler);
207210
}
208211
if (this.channelResolver == null) {
209212
this.channelResolver = ChannelResolverUtils.getChannelResolver(this.beanFactory);
@@ -234,7 +237,9 @@ private void populateComponentNameIfAny() {
234237
}
235238
}
236239

237-
private void adviceChain() {
240+
@SuppressWarnings("unchecked")
241+
private <H> H adviceChain(H handler) {
242+
H theHandler = handler;
238243
if (!CollectionUtils.isEmpty(this.adviceChain)) {
239244
/*
240245
* ARPMHs advise the handleRequestMessage method internally and already have the advice chain injected.
@@ -243,24 +248,25 @@ private void adviceChain() {
243248
* If the handler is already advised,
244249
* add the configured advices to its chain, otherwise create a proxy.
245250
*/
246-
Class<?> targetClass = AopUtils.getTargetClass(this.handler);
251+
Class<?> targetClass = AopUtils.getTargetClass(theHandler);
247252
boolean replyMessageHandler = AbstractReplyProducingMessageHandler.class.isAssignableFrom(targetClass);
248253

249254
for (Advice advice : this.adviceChain) {
250255
if (!replyMessageHandler || advice instanceof HandleMessageAdvice) {
251256
NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(advice);
252257
handlerAdvice.addMethodName("handleMessage");
253-
if (this.handler instanceof Advised) {
254-
((Advised) this.handler).addAdvisor(handlerAdvice);
258+
if (theHandler instanceof Advised) {
259+
((Advised) theHandler).addAdvisor(handlerAdvice);
255260
}
256261
else {
257-
ProxyFactory proxyFactory = new ProxyFactory(this.handler);
262+
ProxyFactory proxyFactory = new ProxyFactory(theHandler);
258263
proxyFactory.addAdvisor(handlerAdvice);
259-
this.handler = (MessageHandler) proxyFactory.getProxy(this.beanClassLoader);
264+
theHandler = (H) proxyFactory.getProxy(this.beanClassLoader);
260265
}
261266
}
262267
}
263268
}
269+
return theHandler;
264270
}
265271

266272
@Override

‎spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2019 the original author or authors.
2+
* Copyright 2016-2020 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -31,7 +31,7 @@
3131
import org.springframework.integration.transaction.TransactionInterceptorBuilder;
3232
import org.springframework.messaging.MessageHandler;
3333
import org.springframework.scheduling.TaskScheduler;
34-
import org.springframework.transaction.PlatformTransactionManager;
34+
import org.springframework.transaction.TransactionManager;
3535
import org.springframework.transaction.interceptor.TransactionInterceptor;
3636
import org.springframework.util.Assert;
3737

@@ -112,10 +112,10 @@ public S advice(Advice... advice) {
112112
* {@code PlatformTransactionManager} and default
113113
* {@link org.springframework.transaction.interceptor.DefaultTransactionAttribute}
114114
* for the {@link MessageHandler}.
115-
* @param transactionManager the {@link PlatformTransactionManager} to use.
115+
* @param transactionManager the {@link TransactionManager} to use.
116116
* @return the spec.
117117
*/
118-
public S transactional(PlatformTransactionManager transactionManager) {
118+
public S transactional(TransactionManager transactionManager) {
119119
return transactional(transactionManager, false);
120120
}
121121

@@ -124,14 +124,14 @@ public S transactional(PlatformTransactionManager transactionManager) {
124124
* {@code PlatformTransactionManager} and default
125125
* {@link org.springframework.transaction.interceptor.DefaultTransactionAttribute}
126126
* for the {@link MessageHandler}.
127-
* @param transactionManager the {@link PlatformTransactionManager} to use.
127+
* @param transactionManager the {@link TransactionManager} to use.
128128
* @param handleMessageAdvice the flag to indicate the target {@link Advice} type:
129129
* {@code false} - regular {@link TransactionInterceptor}; {@code true} -
130130
* {@link org.springframework.integration.transaction.TransactionHandleMessageAdvice}
131131
* extension.
132132
* @return the spec.
133133
*/
134-
public S transactional(PlatformTransactionManager transactionManager, boolean handleMessageAdvice) {
134+
public S transactional(TransactionManager transactionManager, boolean handleMessageAdvice) {
135135
return transactional(new TransactionInterceptorBuilder(handleMessageAdvice)
136136
.transactionManager(transactionManager)
137137
.build());

‎spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2019 the original author or authors.
2+
* Copyright 2016-2020 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -31,7 +31,7 @@
3131
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
3232
import org.springframework.messaging.MessageChannel;
3333
import org.springframework.scheduling.Trigger;
34-
import org.springframework.transaction.PlatformTransactionManager;
34+
import org.springframework.transaction.TransactionManager;
3535
import org.springframework.transaction.interceptor.TransactionInterceptor;
3636
import org.springframework.util.ErrorHandler;
3737

@@ -147,10 +147,10 @@ public PollerSpec advice(Advice... advice) {
147147
* provided {@code PlatformTransactionManager} and default
148148
* {@link org.springframework.transaction.interceptor.DefaultTransactionAttribute}
149149
* for the {@code pollingTask}.
150-
* @param transactionManager the {@link PlatformTransactionManager} to use.
150+
* @param transactionManager the {@link TransactionManager} to use.
151151
* @return the spec.
152152
*/
153-
public PollerSpec transactional(PlatformTransactionManager transactionManager) {
153+
public PollerSpec transactional(TransactionManager transactionManager) {
154154
return transactional(new TransactionInterceptorBuilder()
155155
.transactionManager(transactionManager)
156156
.build());

‎spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionHandleMessageAdvice.java

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2019 the original author or authors.
2+
* Copyright 2016-2020 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -19,7 +19,7 @@
1919
import java.util.Properties;
2020

2121
import org.springframework.integration.handler.advice.HandleMessageAdvice;
22-
import org.springframework.transaction.PlatformTransactionManager;
22+
import org.springframework.transaction.TransactionManager;
2323
import org.springframework.transaction.interceptor.TransactionAttributeSource;
2424
import org.springframework.transaction.interceptor.TransactionInterceptor;
2525

@@ -46,12 +46,16 @@ public class TransactionHandleMessageAdvice extends TransactionInterceptor imple
4646
public TransactionHandleMessageAdvice() {
4747
}
4848

49-
public TransactionHandleMessageAdvice(PlatformTransactionManager ptm, Properties attributes) {
50-
super(ptm, attributes);
49+
public TransactionHandleMessageAdvice(TransactionManager transactionManager, Properties transactionAttributes) {
50+
setTransactionManager(transactionManager);
51+
setTransactionAttributes(transactionAttributes);
5152
}
5253

53-
public TransactionHandleMessageAdvice(PlatformTransactionManager ptm, TransactionAttributeSource tas) {
54-
super(ptm, tas);
54+
public TransactionHandleMessageAdvice(TransactionManager transactionManager,
55+
TransactionAttributeSource transactionAttributeSource) {
56+
57+
setTransactionManager(transactionManager);
58+
setTransactionAttributeSource(transactionAttributeSource);
5559
}
5660

5761
}

‎spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionInterceptorBuilder.java

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2019 the original author or authors.
2+
* Copyright 2016-2020 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
1616

1717
package org.springframework.integration.transaction;
1818

19-
import org.springframework.transaction.PlatformTransactionManager;
19+
import org.springframework.transaction.TransactionManager;
2020
import org.springframework.transaction.annotation.Isolation;
2121
import org.springframework.transaction.annotation.Propagation;
2222
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
@@ -29,8 +29,8 @@
2929
* Provides a fluent API to build a transaction interceptor. See
3030
* {@link TransactionAttribute} for property meanings; if a {@link TransactionAttribute}
3131
* is provided, the individual properties are ignored. If a
32-
* {@link PlatformTransactionManager} is not provided, a single instance of
33-
* {@link PlatformTransactionManager} will be discovered at runtime; if you have more
32+
* {@link TransactionManager} is not provided, a single instance of
33+
* {@link TransactionManager} will be discovered at runtime; if you have more
3434
* than one transaction manager, you must inject the one you want to use here.
3535
* <p>
3636
* When the {@code handleMessageAdvice} option is in use, this builder produces
@@ -91,7 +91,7 @@ public final TransactionInterceptorBuilder transactionAttribute(TransactionAttri
9191
return this;
9292
}
9393

94-
public TransactionInterceptorBuilder transactionManager(PlatformTransactionManager transactionManager) {
94+
public TransactionInterceptorBuilder transactionManager(TransactionManager transactionManager) {
9595
this.transactionInterceptor.setTransactionManager(transactionManager);
9696
return this;
9797
}

‎spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -4155,56 +4155,56 @@
41554155
<xsd:complexType name="transactionalType">
41564156
<xsd:attribute name="transaction-manager" type="xsd:string" default="transactionManager">
41574157
<xsd:annotation>
4158-
<xsd:documentation><![CDATA[
4159-
The bean name of the PlatformTransactionManager to use.
4160-
]]></xsd:documentation>
4158+
<xsd:documentation>
4159+
The bean name of the PlatformTransactionManager to use.
4160+
</xsd:documentation>
41614161
<xsd:appinfo>
41624162
<tool:annotation kind="ref">
4163-
<tool:expected-type type="org.springframework.transaction.PlatformTransactionManager"/>
4163+
<tool:expected-type type="org.springframework.transaction.TransactionManager"/>
41644164
</tool:annotation>
41654165
</xsd:appinfo>
41664166
</xsd:annotation>
41674167
</xsd:attribute>
41684168
<xsd:attribute name="propagation" default="REQUIRED">
41694169
<xsd:annotation>
4170-
<xsd:documentation source="java:org.springframework.transaction.annotation.Propagation"><![CDATA[
4171-
The transaction propagation behavior.
4172-
]]></xsd:documentation>
4170+
<xsd:documentation source="java:org.springframework.transaction.annotation.Propagation">
4171+
The transaction propagation behavior.
4172+
</xsd:documentation>
41734173
</xsd:annotation>
41744174
<xsd:simpleType>
41754175
<xsd:union memberTypes="propagationEnumeration xsd:string"/>
41764176
</xsd:simpleType>
41774177
</xsd:attribute>
41784178
<xsd:attribute name="isolation" default="DEFAULT">
41794179
<xsd:annotation>
4180-
<xsd:documentation source="java:org.springframework.transaction.annotation.Isolation"><![CDATA[
4181-
The transaction isolation level.
4182-
]]></xsd:documentation>
4180+
<xsd:documentation source="java:org.springframework.transaction.annotation.Isolation">
4181+
The transaction isolation level.
4182+
</xsd:documentation>
41834183
</xsd:annotation>
41844184
<xsd:simpleType>
41854185
<xsd:union memberTypes="isolationEnumeration xsd:string"/>
41864186
</xsd:simpleType>
41874187
</xsd:attribute>
41884188
<xsd:attribute name="timeout" type="xsd:string" default="-1">
41894189
<xsd:annotation>
4190-
<xsd:documentation><![CDATA[
4191-
The transaction timeout value (in seconds).
4192-
]]></xsd:documentation>
4190+
<xsd:documentation>
4191+
The transaction timeout value (in seconds).
4192+
</xsd:documentation>
41934193
</xsd:annotation>
41944194
</xsd:attribute>
41954195
<xsd:attribute name="read-only" type="xsd:string" default="false">
41964196
<xsd:annotation>
4197-
<xsd:documentation><![CDATA[
4198-
Is this transaction read-only?
4199-
]]></xsd:documentation>
4197+
<xsd:documentation>
4198+
Is this transaction read-only?
4199+
</xsd:documentation>
42004200
</xsd:annotation>
42014201
</xsd:attribute>
42024202
<xsd:attribute name="synchronization-factory" type="xsd:string">
42034203
<xsd:annotation>
4204-
<xsd:documentation><![CDATA[
4204+
<xsd:documentation>
42054205
Reference to an instance of org.springframework.integration.transaction.TransactionSynchronizationFactory
42064206
which will return an instance of org.springframework.transaction.support.TransactionSynchronization via its create(..) method.
4207-
]]></xsd:documentation>
4207+
</xsd:documentation>
42084208
<xsd:appinfo>
42094209
<tool:annotation kind="ref">
42104210
<tool:expected-type

‎spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/dsl/MongoDbTests.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -423,6 +423,7 @@ public IntegrationFlow reactiveStore() {
423423
return f -> f
424424
.channel(MessageChannels.flux())
425425
.handle(MongoDb.reactiveOutboundChannelAdapter(REACTIVE_MONGO_DATABASE_FACTORY));
426+
// e -> e.transactional(new ReactiveMongoTransactionManager(REACTIVE_MONGO_DATABASE_FACTORY)));
426427
}
427428

428429

‎src/reference/asciidoc/whats-new.adoc

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,9 @@ A new `publishSubscribeChannel()` operator, based on the `BroadcastCapableChanne
5757
This fluent API has its advantage when we configure sub-flows as pub-sub subscribers for broker-backed channels like `SubscribableJmsChannel`, `SubscribableRedisChannel` etc.
5858
See <<./dsl.adoc#java-dsl-subflows,Sub-flows support>> for more information.
5959

60+
The transactional support in Spring Integration now also includes options to configure a `ReactiveTransactionManager` if a `MessageSource` or `MessageHandler` implementation produces a reactive type for payload to send.
61+
See `TransactionInterceptorBuilder` for more information.
62+
6063
[[x5.3-amqp]]
6164
=== AMQP Changes
6265

0 commit comments

Comments
 (0)