INT-4545: Fix RPMHWrapper registration for `@SA

JIRA: https://jira.spring.io/browse/INT-4545

Improve the logic in the `AbstractMethodAnnotationPostProcessor` to
check for existing `MessageHandler` and be sure that we need to register
a `ReplyProducingMessageHandlerWrapper` bean

**Cherry-pick to 5.0.x**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java

@@ -135,9 +135,11 @@ public Object postProcess(Object bean, String beanName, Method method, List<Anno
 			}
 		}
 
-		boolean handlerExists = false;
+		Object sourceHandler = null;
 		if (beanAnnotationAware() && AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
-			handlerExists = MessageHandler.class.isAssignableFrom(method.getReturnType());
+			if (MessageHandler.class.isAssignableFrom(method.getReturnType())) {
+				sourceHandler = resolveTargetBeanFromMethodWithBeanAnnotation(method);
+			}
 		}
 
 		MessageHandler handler = createHandler(bean, method, annotations);
@@ -167,7 +169,7 @@ public Object postProcess(Object bean, String beanName, Method method, List<Anno
 			}
 		}
 
-		if (!handlerExists) {
+		if (handler != sourceHandler) {
 			String handlerBeanName = generateHandlerBeanName(beanName, method);
 			if (handler instanceof ReplyProducingMessageHandlerWrapper
 					&& StringUtils.hasText(MessagingAnnotationUtils.endpointIdValue(method))) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -19,12 +19,13 @@
 import org.springframework.context.Lifecycle;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.util.Assert;
 
 /**
  * The {@link AbstractReplyProducingMessageHandler} wrapper around raw {@link MessageHandler}
  * for request-reply scenarios, e.g. {@code @ServiceActivator} annotation configuration.
  * <p>
- * This class is used internally by Framework in cased when request-reply is important
+ * This class is used internally by Framework in cases when request-reply is important
  * and there is no other way to apply advice chain.
  * <p>
  * The lifecycle control is delegated to the {@code target} {@link MessageHandler}.
@@ -39,6 +40,7 @@ public class ReplyProducingMessageHandlerWrapper extends AbstractReplyProducingM
 	private final MessageHandler target;
 
 	public ReplyProducingMessageHandlerWrapper(MessageHandler target) {
+		Assert.notNull(target, "'target' must not be null");
 		this.target = target;
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java

@@ -40,13 +40,15 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 import java.util.Arrays;
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.apache.commons.logging.Log;
 import org.hamcrest.BaseMatcher;
@@ -109,6 +111,7 @@
 import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.expression.SpelPropertyAccessorRegistrar;
 import org.springframework.integration.gateway.GatewayProxyFactoryBean;
+import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.history.MessageHistoryConfigurer;
 import org.springframework.integration.json.JsonPropertyAccessor;
@@ -718,6 +721,24 @@ public void testIntegrationEvaluationContextCustomization() {
 		assertEquals(ClassUtils.getStaticMethod(TestSpelFunction.class, "bar", Object.class), testSpelFunction);
 	}
 
+	@Autowired
+	private MessageChannel myHandlerChannel;
+
+	@Autowired
+	private PollableChannel myHandlerSuccessChannel;
+
+	@Test
+	public void testAdvicedServiceActivator() {
+		Date testDate = new Date();
+
+		this.myHandlerChannel.send(new GenericMessage<>(testDate));
+
+		Message<?> receive = this.myHandlerSuccessChannel.receive(10_000);
+
+		assertNotNull(receive);
+		assertEquals(testDate, receive.getPayload());
+	}
+
 	@Configuration
 	@ComponentScan
 	@IntegrationComponentScan
@@ -1124,6 +1145,25 @@ public SpelPropertyAccessorRegistrar spelPropertyAccessorRegistrar() {
 			return new SpelPropertyAccessorRegistrar(new JsonPropertyAccessor(), new EnvironmentAccessor());
 		}
 
+		@Bean
+		@ServiceActivator(inputChannel = "myHandlerChannel", adviceChain = "myHandlerAdvice")
+		public MessageHandler myHandler() {
+			return message -> { };
+		}
+
+		@Bean
+		public Advice myHandlerAdvice() {
+			ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
+			advice.setOnSuccessExpressionString("payload");
+			advice.setSuccessChannel(myHandlerSuccessChannel());
+			return advice;
+		}
+
+		@Bean
+		public QueueChannel myHandlerSuccessChannel() {
+			return new QueueChannel();
+		}
+
 	}
 
 	@Configuration