Spring/spring-integration-extensions
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INTEXT-114 Fix outputChannel population AOP issue

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-114
This commit is contained in:
Artem Bilan
2014-09-04 17:35:49 +03:00
committed by Gary Russell
parent eb5456f9d7
commit ff4a229bc5
2 changed files with 59 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -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);
}
}
}

34
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java
View File

@@ -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")