GH-3219: Add HandleMessageAdviceAdapter (#3234)

* GH-3219: Add `HandleMessageAdviceAdapter`

Fixes #3219

* * Fix language in doc

Co-Authored-By: Gary Russell <[email protected]>

Co-authored-by: Gary Russell <[email protected]>

Author: artembilan

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

@@ -22,12 +22,14 @@
 import java.util.function.BiFunction;
 
 import org.aopalliance.aop.Advice;
+import org.aopalliance.intercept.MethodInterceptor;
 import org.reactivestreams.Publisher;
 
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.advice.HandleMessageAdviceAdapter;
 import org.springframework.integration.handler.advice.ReactiveRequestHandlerAdvice;
 import org.springframework.integration.router.AbstractMessageRouter;
 import org.springframework.integration.scheduling.PollerMetadata;
@@ -101,6 +103,21 @@ public S taskScheduler(TaskScheduler taskScheduler) {
 		return _this();
 	}
 
+	/**
+	 * Configure a list of {@link MethodInterceptor} objects to be applied, in nested order, to the
+	 * endpoint's handler. The advice objects are applied to the {@code handleMessage()} method
+	 * and therefore to the whole sub-flow afterwards.
+	 * @param interceptors the advice chain.
+	 * @return the endpoint spec.
+	 * @since 5.3
+	 */
+	public S handleMessageAdvice(MethodInterceptor... interceptors) {
+		for (MethodInterceptor interceptor: interceptors) {
+			advice(new HandleMessageAdviceAdapter(interceptor));
+		}
+		return _this();
+	}
+
 	/**
 	 * Configure a list of {@link Advice} objects to be applied, in nested order, to the
 	 * endpoint's handler. The advice objects are applied only to the handler.

spring-integration-core/src/main/java/org/springframework/integration/handler/advice/HandleMessageAdviceAdapter.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.handler.advice;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+
+import org.springframework.util.Assert;
+
+/**
+ * A {@link HandleMessageAdvice} implementation with a plain delegation
+ * to the provided {@link MethodInterceptor}.
+ * <p> This advice should be used for consumer endpoints to proxy exactly
+ * a {@link org.springframework.messaging.MessageHandler#handleMessage} and the whole-subflow therefore;
+ * unlike regular proxying which is applied only for the
+ * {@link org.springframework.integration.handler.AbstractReplyProducingMessageHandler.RequestHandler#handleRequestMessage}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ */
+public class HandleMessageAdviceAdapter implements HandleMessageAdvice {
+
+	private final MethodInterceptor delegate;
+
+	public HandleMessageAdviceAdapter(MethodInterceptor delegate) {
+		Assert.notNull(delegate, "The 'delegate' must not be null");
+		this.delegate = delegate;
+	}
+
+
+	@Override
+	public Object invoke(MethodInvocation invocation) throws Throwable {
+		return this.delegate.invoke(invocation);
+	}
+
+}

spring-integration-jpa/src/test/java/org/springframework/integration/jpa/dsl/JpaTests.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.
@@ -41,6 +41,7 @@
 import org.springframework.integration.jpa.test.entity.StudentDomain;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.OnlyOnceTrigger;
+import org.springframework.integration.transaction.TransactionInterceptorBuilder;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
@@ -57,6 +58,7 @@
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
 
 /**
  * @author Artem Bilan
@@ -230,7 +232,12 @@ public IntegrationFlow outboundAdapterFlow(EntityManagerFactory entityManagerFac
 					.handle(Jpa.outboundAdapter(entityManagerFactory)
 									.entityClass(StudentDomain.class)
 									.persistMode(PersistMode.PERSIST),
-							e -> e.transactional(true));
+							e -> e.handleMessageAdvice(transactionInterceptor()));
+		}
+
+		@Bean
+		public TransactionInterceptor transactionInterceptor() {
+			return new TransactionInterceptorBuilder().build();
 		}
 
 		@Bean

src/reference/asciidoc/handler-advice.adoc

@@ -683,6 +683,11 @@ Note that, in that case, the entire downstream flow is within the transaction sc
 
 In the case of a `MessageHandler` that does not return a response, the advice chain order is retained.
 
+Starting with version 5.3, the `HandleMessageAdviceAdapter` is present to let apply any existing `MethodInterceptor` for the `MessageHandler.handleMessage()` and, therefore, whole sub-flow.
+For example a `RetryOperationsInterceptor` could be applied for the whole sub-flow starting from some endpoint, which is not possible by default because consumer endpoint applies advices only for the `AbstractReplyProducingMessageHandler.RequestHandler.handleRequestMessage()`.
+Starting with version 5.3, the `HandleMessageAdviceAdapter` is provided to apply any `MethodInterceptor` for the `MessageHandler.handleMessage()` method and, therefore, the whole sub-flow.
+For example, a `RetryOperationsInterceptor` could be applied to the whole sub-flow starting from some endpoint; this is not possible, by default, because the consumer endpoint applies advices only to the `AbstractReplyProducingMessageHandler.RequestHandler.handleRequestMessage()`.
+
 [[tx-handle-message-advice]]
 ==== Transaction Support
 

src/reference/asciidoc/whats-new.adoc

@@ -46,6 +46,12 @@ See <<./kotlin-dsl.adoc#kotlin-dsl,Kotlin DSL Chapter>> for more information.
 A `ReactiveRequestHandlerAdvice` is provided to customize `Mono` replies from message handlers.
 See <<./handler-advice.adoc#reactive-advice,Reactive Advice>> for more information.
 
+[[x5.3-reactive-request-handler-advice]]
+==== HandleMessageAdviceAdapter
+
+A `HandleMessageAdviceAdapter` is provided to wrap any `MethodInterceptor` for applying on the `MessageHandler.handleMessage()` instead of a default `AbstractReplyProducingMessageHandler.RequestHandler.handleRequestMessage()` behavior.
+See <<./handler-advice.adoc#handle-message-advice,Handling Message Advice>> for more information.
+
 [[x5.3-mongodb-reactive-channel-adapters]]
 ==== MongoDB Reactive Channel Adapters