From ff4a229bc5c2d320c1da1dc5fe6429bd27100547 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Thu, 4 Sep 2014 17:35:49 +0300
Subject: [PATCH] INTEXT-114 Fix outputChannel population AOP issue

JIRA: https://jira.spring.io/browse/INTEXT-114
---
 .../dsl/IntegrationFlowDefinition.java        | 39 +++++++++++++++----
 .../dsl/test/flows/IntegrationFlowTests.java  | 34 ++++++++++++----
 2 files changed, 59 insertions(+), 14 deletions(-)

diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
index 2dc6f5f..bce2c23 100644
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
@@ -21,6 +21,8 @@ import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Set;
 
+import org.springframework.aop.framework.Advised;
+import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler;
@@ -614,9 +616,16 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 				if (outputChannel instanceof MessageChannelReference) {
 					channelName = ((MessageChannelReference) outputChannel).getName();
 				}
-				if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) {
+
+				Object currentComponent = this.currentComponent;
+
+				if (AopUtils.isAopProxy(currentComponent)) {
+					currentComponent = extractProxyTarget(currentComponent);
+				}
+
+				if (currentComponent instanceof AbstractReplyProducingMessageHandler) {
 					AbstractReplyProducingMessageHandler messageProducer =
-							(AbstractReplyProducingMessageHandler) this.currentComponent;
+							(AbstractReplyProducingMessageHandler) currentComponent;
 					if (channelName != null) {
 						messageProducer.setOutputChannelName(channelName);
 					}
@@ -624,9 +633,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 						messageProducer.setOutputChannel(outputChannel);
 					}
 				}
-				else if (this.currentComponent instanceof SourcePollingChannelAdapterSpec) {
+				else if (currentComponent instanceof SourcePollingChannelAdapterSpec) {
 					SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean =
-							((SourcePollingChannelAdapterSpec) this.currentComponent).get().getT1();
+							((SourcePollingChannelAdapterSpec) currentComponent).get().getT1();
 					if (channelName != null) {
 						pollingChannelAdapterFactoryBean.setOutputChannelName(channelName);
 					}
@@ -634,9 +643,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 						pollingChannelAdapterFactoryBean.setOutputChannel(outputChannel);
 					}
 				}
-				else if (this.currentComponent instanceof AbstractCorrelatingMessageHandler) {
+				else if (currentComponent instanceof AbstractCorrelatingMessageHandler) {
 					AbstractCorrelatingMessageHandler messageProducer =
-							(AbstractCorrelatingMessageHandler) this.currentComponent;
+							(AbstractCorrelatingMessageHandler) currentComponent;
 					if (channelName != null) {
 						messageProducer.setOutputChannelName(channelName);
 					}
@@ -654,7 +663,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 					}
 				}
 				else {
-					throw new BeanCreationException("The 'currentComponent' (" + this.currentComponent +
+					throw new BeanCreationException("The 'currentComponent' (" + currentComponent +
 							") is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. " +
 							"This is the end of the integration flow.");
 				}
@@ -674,4 +683,20 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return aClass.isSynthetic() && !aClass.isAnonymousClass() && !aClass.isLocalClass();
 	}
 
+	private static Object extractProxyTarget(Object target) {
+		if (!(target instanceof Advised)) {
+			return target;
+		}
+		Advised advised = (Advised) target;
+		if (advised.getTargetSource() == null) {
+			return null;
+		}
+		try {
+			return extractProxyTarget(advised.getTargetSource().getTarget());
+		}
+		catch (Exception e) {
+			throw new BeanCreationException("Could not extract target", e);
+		}
+	}
+
 }
diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java
index 1a7cb9e..671f65d 100644
--- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java
+++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java
@@ -47,6 +47,10 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Matcher;
 
+import javax.management.MBeanServer;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
+
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
@@ -1271,6 +1275,17 @@ public class IntegrationFlowTests {
 				Matchers.containsString(dir + "subSftpSource"));
 	}
 
+	@Autowired
+	private MBeanServer mBeanServer;
+
+	@Test
+	public void testMBeansForDSL() throws MalformedObjectNameException {
+		assertFalse(this.mBeanServer.queryMBeans(ObjectName.getInstance("org.springframework.integration:" +
+				"bean=anonymous,name=sftpMgetInputChannel,type=MessageHandler"), null).isEmpty());
+		assertFalse(this.mBeanServer.queryMBeans(ObjectName.getInstance("org.springframework.integration:" +
+				"type=MessageHandler,name=ftpMgetInputChannel,bean=anonymous"), null).isEmpty());
+	}
+
 	@MessagingGateway(defaultRequestChannel = "controlBus")
 	private static interface ControlBusGateway {
 
@@ -1444,19 +1459,24 @@ public class IntegrationFlowTests {
 			return new QueueChannel();
 		}
 
+		@Bean
+		public MessageHandler ftpOutboundGateway() {
+			return Ftp.outboundGateway(this.ftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET,
+					"payload")
+					.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
+					.regexFileNameFilter("(subFtpSource|.*1.txt)")
+					.localDirectoryExpression("@ftpServer.targetLocalDirectoryName + #remoteDirectory")
+					.localFilenameGeneratorExpression("#remoteFileName.replaceFirst('ftpSource', " +
+							"'localTarget')").get();
+		}
+
 		@Bean
 		public IntegrationFlow ftpMGetFlow() {
 			return IntegrationFlows.from("ftpMgetInputChannel")
-					.handle(Ftp.outboundGateway(this.ftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET,
-							"payload")
-							.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
-							.regexFileNameFilter("(subFtpSource|.*1.txt)")
-							.localDirectoryExpression("@ftpServer.targetLocalDirectoryName + #remoteDirectory")
-							.localFilenameGeneratorExpression("#remoteFileName.replaceFirst('ftpSource', 'localTarget')"))
+					.handle(ftpOutboundGateway())
 					.channel(remoteFileOutputChannel())
 					.get();
 		}
-
 		@Bean
 		public IntegrationFlow sftpMGetFlow() {
 			return IntegrationFlows.from("sftpMgetInputChannel")