From 45a91b25de34cbaad79856888908cb7251544e38 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 3 Mar 2020 11:54:29 -0500 Subject: [PATCH] Add Reactive TX Manager support (#3201) * 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 `` 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 * * Revert `MongoDbTests` change - to support transactions we need the latest MongoDb server with replica enabled * Document reactive transactions * * Address PR reviews --- .../config/ConsumerEndpointFactoryBean.java | 21 ++++++----- .../integration/dsl/ConsumerEndpointSpec.java | 12 +++---- .../integration/dsl/PollerSpec.java | 8 ++--- .../TransactionHandleMessageAdvice.java | 16 +++++---- .../TransactionInterceptorBuilder.java | 10 +++--- .../integration/config/spring-integration.xsd | 36 +++++++++---------- src/reference/asciidoc/transactions.adoc | 9 +++++ src/reference/asciidoc/whats-new.adoc | 4 +++ 8 files changed, 69 insertions(+), 47 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 746ecbdbec..acc1a70a81 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -200,10 +200,12 @@ public class ConsumerEndpointFactoryBean } if (!(this.handler instanceof ReactiveMessageHandlerAdapter)) { - adviceChain(); + this.handler = adviceChain(this.handler); } else if (!CollectionUtils.isEmpty(this.adviceChain)) { - LOGGER.warn("the advice chain cannot be applied to a 'ReactiveMessageHandler'"); + this.handler = + new ReactiveMessageHandlerAdapter( + adviceChain(((ReactiveMessageHandlerAdapter) this.handler).getDelegate())); } if (this.channelResolver == null) { this.channelResolver = ChannelResolverUtils.getChannelResolver(this.beanFactory); @@ -234,7 +236,9 @@ public class ConsumerEndpointFactoryBean } } - private void adviceChain() { + @SuppressWarnings("unchecked") + private H adviceChain(H handler) { + H theHandler = handler; if (!CollectionUtils.isEmpty(this.adviceChain)) { /* * ARPMHs advise the handleRequestMessage method internally and already have the advice chain injected. @@ -243,24 +247,25 @@ public class ConsumerEndpointFactoryBean * If the handler is already advised, * add the configured advices to its chain, otherwise create a proxy. */ - Class targetClass = AopUtils.getTargetClass(this.handler); + Class targetClass = AopUtils.getTargetClass(theHandler); boolean replyMessageHandler = AbstractReplyProducingMessageHandler.class.isAssignableFrom(targetClass); for (Advice advice : this.adviceChain) { if (!replyMessageHandler || advice instanceof HandleMessageAdvice) { NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(advice); handlerAdvice.addMethodName("handleMessage"); - if (this.handler instanceof Advised) { - ((Advised) this.handler).addAdvisor(handlerAdvice); + if (theHandler instanceof Advised) { + ((Advised) theHandler).addAdvisor(handlerAdvice); } else { - ProxyFactory proxyFactory = new ProxyFactory(this.handler); + ProxyFactory proxyFactory = new ProxyFactory(theHandler); proxyFactory.addAdvisor(handlerAdvice); - this.handler = (MessageHandler) proxyFactory.getProxy(this.beanClassLoader); + theHandler = (H) proxyFactory.getProxy(this.beanClassLoader); } } } } + return theHandler; } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java index a3144aadd0..16f4c5da8d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -31,7 +31,7 @@ import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.transaction.TransactionInterceptorBuilder; import org.springframework.messaging.MessageHandler; import org.springframework.scheduling.TaskScheduler; -import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionManager; import org.springframework.transaction.interceptor.TransactionInterceptor; import org.springframework.util.Assert; @@ -112,10 +112,10 @@ public abstract class ConsumerEndpointSpec, * {@code PlatformTransactionManager} and default * {@link org.springframework.transaction.interceptor.DefaultTransactionAttribute} * for the {@link MessageHandler}. - * @param transactionManager the {@link PlatformTransactionManager} to use. + * @param transactionManager the {@link TransactionManager} to use. * @return the spec. */ - public S transactional(PlatformTransactionManager transactionManager) { + public S transactional(TransactionManager transactionManager) { return transactional(transactionManager, false); } @@ -124,14 +124,14 @@ public abstract class ConsumerEndpointSpec, * {@code PlatformTransactionManager} and default * {@link org.springframework.transaction.interceptor.DefaultTransactionAttribute} * for the {@link MessageHandler}. - * @param transactionManager the {@link PlatformTransactionManager} to use. + * @param transactionManager the {@link TransactionManager} to use. * @param handleMessageAdvice the flag to indicate the target {@link Advice} type: * {@code false} - regular {@link TransactionInterceptor}; {@code true} - * {@link org.springframework.integration.transaction.TransactionHandleMessageAdvice} * extension. * @return the spec. */ - public S transactional(PlatformTransactionManager transactionManager, boolean handleMessageAdvice) { + public S transactional(TransactionManager transactionManager, boolean handleMessageAdvice) { return transactional(new TransactionInterceptorBuilder(handleMessageAdvice) .transactionManager(transactionManager) .build()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java index c56283ab64..efbcf020dd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -31,7 +31,7 @@ import org.springframework.integration.transaction.TransactionInterceptorBuilder import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.messaging.MessageChannel; import org.springframework.scheduling.Trigger; -import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionManager; import org.springframework.transaction.interceptor.TransactionInterceptor; import org.springframework.util.ErrorHandler; @@ -147,10 +147,10 @@ public final class PollerSpec extends IntegrationComponentSpec * When the {@code handleMessageAdvice} option is in use, this builder produces @@ -91,7 +91,7 @@ public class TransactionInterceptorBuilder { return this; } - public TransactionInterceptorBuilder transactionManager(PlatformTransactionManager transactionManager) { + public TransactionInterceptorBuilder transactionManager(TransactionManager transactionManager) { this.transactionInterceptor.setTransactionManager(transactionManager); return this; } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd index 3a188499c2..a6cafa938a 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd @@ -4155,21 +4155,21 @@ - + + The bean name of the PlatformTransactionManager to use. + - + - + + The transaction propagation behavior. + @@ -4177,9 +4177,9 @@ - + + The transaction isolation level. + @@ -4187,24 +4187,24 @@ - + + The transaction timeout value (in seconds). + - + + Is this transaction read-only? + - Reference to an instance of org.springframework.integration.transaction.TransactionSynchronizationFactory which will return an instance of org.springframework.transaction.support.TransactionSynchronization via its create(..) method. - ]]> + ---- ==== + +[[reactive-transactions]] +=== Reactive Transactions + +Starting with version 5.3, a `ReactiveTransactionManager` can also be used together with a `TransactionInterceptor` advice for endpoints returning a reactive type. +This includes `MessageSource` and `ReactiveMessageHandler` implementations (e.g. `ReactiveMongoDbMessageSource`) which produce a message with a `Flux` or `Mono` payload. +All other reply producing message handler implementations can rely on a `ReactiveTransactionManager` when their reply payload is also some reactive type. + + diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 05617fb369..190e802122 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -57,6 +57,10 @@ A new `publishSubscribeChannel()` operator, based on the `BroadcastCapableChanne This fluent API has its advantage when we configure sub-flows as pub-sub subscribers for broker-backed channels like `SubscribableJmsChannel`, `SubscribableRedisChannel` etc. See <<./dsl.adoc#java-dsl-subflows,Sub-flows support>> for more information. +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. +See `TransactionInterceptorBuilder` for more information. +See also <<./transactions.adoc#reactive-transactions,Reactive Transactions>>. + [[x5.3-amqp]] === AMQP Changes