From e6e43100b522a6ea18d4965088f5fb78ba29e50f Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 6 Jul 2012 13:02:55 -0400 Subject: [PATCH] INT-2633 Add File Disposition to (S)FTP Inbound FileReadingMessageSource supports disposition of the payload after the message is sent (or via transaction synchronization). The (S)FTP adapters delegate to an FRMS; add support for the file disposition expression, result channel, and send timeout to the (S)FTP adapters. Also tested with ftp sample - INTSAMPLES-83 - patch will be committed once this is in a milestone. --- .../integration/file/FileMessageHolder.java | 43 ++++++++++++++++ .../file/FileReadingMessageSource.java | 13 ----- ...RemoteFileInboundChannelAdapterParser.java | 10 ++-- .../FileInboundChannelAdapterParser.java | 11 +---- .../file/config/FileNamespaceUtils.java | 43 ++++++++++++++++ ...InboundFileSynchronizingMessageSource.java | 49 ++++++++++++++++--- .../file/FileReadingMessageSourceTests.java | 1 - .../FtpInboundChannelAdapterParser.java | 20 ++++---- .../ftp/config/spring-integration-ftp-2.2.xsd | 36 ++++++++++++++ ...boundChannelAdapterParserTests-context.xml | 7 ++- .../FtpInboundChannelAdapterParserTests.java | 22 ++++++--- .../SftpInboundChannelAdapterParser.java | 20 ++++---- .../config/spring-integration-sftp-2.2.xsd | 36 ++++++++++++++ ...boundChannelAdapterParserTests-context.xml | 7 ++- .../InboundChannelAdapterParserTests.java | 25 ++++++---- 15 files changed, 271 insertions(+), 72 deletions(-) create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/FileMessageHolder.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceUtils.java diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileMessageHolder.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileMessageHolder.java new file mode 100644 index 0000000000..ff844aa424 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileMessageHolder.java @@ -0,0 +1,43 @@ +/* + * Copyright 2002-2012 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 + * + * http://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.file; + +import java.io.File; + +import org.springframework.integration.Message; + +/** + * A simple wrapper for a Message; used for + * file disposition after the send completes, or + * after the transaction commits with a transactional + * poller. + * @author Gary Russell + * @since 2.2 + * + */ +public class FileMessageHolder { + + private Message message; + + Message getMessage() { + return message; + } + + void setMessage(Message message) { + this.message = message; + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 767f40af02..f24517bb76 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -38,7 +38,6 @@ import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageSource; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.core.PseudoTransactionalMessageSource; -import org.springframework.integration.file.FileReadingMessageSource.FileMessageHolder; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.support.MessageBuilder; @@ -389,16 +388,4 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement this.afterCommit(resource); } - class FileMessageHolder { - - private Message message; - - Message getMessage() { - return message; - } - - void setMessage(Message message) { - this.message = message; - } - } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java index 686ff21aa6..cebcac8d0f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java @@ -16,8 +16,6 @@ package org.springframework.integration.file.config; -import org.w3c.dom.Element; - import org.springframework.beans.BeanMetadataElement; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; @@ -26,10 +24,11 @@ import org.springframework.integration.config.xml.AbstractPollingInboundChannelA import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.file.remote.session.SessionFactoryFactoryBean; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; /** * Abstract base class for parsing remote file inbound channel adapters. - * + * * @author Oleg Zhurakousky * @author Mark Fisher * @since 2.0 @@ -44,9 +43,9 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(SessionFactoryFactoryBean.class); sessionFactoryBuilder.addConstructorArgReference(element.getAttribute("session-factory")); sessionFactoryBuilder.addConstructorArgValue(element.getAttribute("cache-sessions")); - + synchronizerBuilder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition()); - + // configure the InboundFileSynchronizer properties IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "remote-directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "delete-remote-files"); @@ -71,6 +70,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst localFileGeneratorExpressionBuilder.addConstructorArgValue(localFileGeneratorExpression); synchronizerBuilder.addPropertyValue("localFilenameGeneratorExpression", localFileGeneratorExpressionBuilder.getBeanDefinition()); } + FileNamespaceUtils.setDispositionAttributes(element, messageSourceBuilder); return messageSourceBuilder.getBeanDefinition(); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java index 7838c5d496..a3fcebcce1 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java @@ -21,9 +21,7 @@ import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; -import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.integration.config.ExpressionFactoryBean; import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.file.locking.NioFileLocker; @@ -49,14 +47,7 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-create-directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-size"); - String dispositionExpression = element.getAttribute("disposition-expression"); - if (StringUtils.hasText(dispositionExpression)) { - RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class); - expressionDef.getConstructorArgumentValues().addGenericArgumentValue(dispositionExpression); - builder.addPropertyValue("dispositionExpression", expressionDef); - } - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "disposition-result-channel"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "disposition-send-timeout"); + FileNamespaceUtils.setDispositionAttributes(element, builder); String filterBeanName = this.registerFilter(element, parserContext); String lockerBeanName = registerLocker(element, parserContext); if (lockerBeanName != null) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceUtils.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceUtils.java new file mode 100644 index 0000000000..186de34831 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceUtils.java @@ -0,0 +1,43 @@ +/* + * Copyright 2002-2012 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 + * + * http://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.file.config; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.integration.config.ExpressionFactoryBean; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; +import org.w3c.dom.Element; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +public class FileNamespaceUtils { + + public static void setDispositionAttributes(Element element, BeanDefinitionBuilder builder) { + String dispositionExpression = element.getAttribute("disposition-expression"); + if (StringUtils.hasText(dispositionExpression)) { + RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class); + expressionDef.getConstructorArgumentValues().addGenericArgumentValue(dispositionExpression); + builder.addPropertyValue("dispositionExpression", expressionDef); + } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "disposition-result-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "disposition-send-timeout"); + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java index 08fd94df3c..b3e3e1c980 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -22,10 +22,13 @@ import java.util.Arrays; import java.util.Comparator; import java.util.regex.Pattern; +import org.springframework.expression.Expression; import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; import org.springframework.integration.MessagingException; -import org.springframework.integration.core.MessageSource; +import org.springframework.integration.core.PseudoTransactionalMessageSource; import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.file.FileMessageHolder; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.CompositeFileListFilter; @@ -50,11 +53,13 @@ import org.springframework.util.Assert; * {@link AbstractInboundFileSynchronizer}. The synchronizer must * handle the work of actually connecting to the remote file system and * delivering new {@link File}s. - * + * * @author Josh Long * @author Oleg Zhurakousky + * @author Gary Russell */ -public abstract class AbstractInboundFileSynchronizingMessageSource extends MessageProducerSupport implements MessageSource { +public abstract class AbstractInboundFileSynchronizingMessageSource extends MessageProducerSupport + implements PseudoTransactionalMessageSource { /** * Should the endpoint attempt to create the local directory? True by default. @@ -81,7 +86,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource extends M public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer synchronizer) { this(synchronizer, null); } - + public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer synchronizer, Comparator comparator) { Assert.notNull(synchronizer, "synchronizer must not be null"); this.synchronizer = synchronizer; @@ -90,7 +95,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource extends M } else { this.fileSource = new FileReadingMessageSource(comparator); - } + } } @@ -102,6 +107,18 @@ public abstract class AbstractInboundFileSynchronizingMessageSource extends M this.localDirectory = localDirectory; } + public void setDispositionExpression(Expression dispositionExpression) { + this.fileSource.setDispositionExpression(dispositionExpression); + } + + public void setDispositionResultChannel(MessageChannel dispositionResultChannel) { + this.fileSource.setDispositionResultChannel(dispositionResultChannel); + } + + public void setDispositionSendTimeout(long dispositionSendTimeout) { + this.fileSource.setDispositionSendTimeout(dispositionSendTimeout); + } + @Override protected void onInit() { Assert.notNull(this.localDirectory, "localDirectory must not be null"); @@ -155,4 +172,24 @@ public abstract class AbstractInboundFileSynchronizingMessageSource extends M new RegexPatternFileListFilter(completePattern))); } + public FileMessageHolder getResource() { + return this.fileSource.getResource(); + } + + public void afterCommit(FileMessageHolder resource) { + this.fileSource.afterCommit(resource); + } + + public void afterRollback(FileMessageHolder resource) { + this.fileSource.afterRollback(resource); + } + + public void afterReceiveNoTx(FileMessageHolder resource) { + this.fileSource.afterReceiveNoTx(resource); + } + + public void afterSendNoTx(FileMessageHolder resource) { + this.fileSource.afterSendNoTx(resource); + } + } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java index 9e6cfc0061..64fcfb4907 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java @@ -40,7 +40,6 @@ import org.mockito.runners.MockitoJUnitRunner; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.file.FileReadingMessageSource.FileMessageHolder; import org.springframework.integration.message.GenericMessage; /** diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java index ac300ce95f..92bec17832 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -17,36 +17,38 @@ package org.springframework.integration.ftp.config; import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser; +import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; +import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; +import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer; +import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource; /** * Parser for the FTP 'inbound-channel-adapter' element. - * + * * @author Mark Fisher + * @author Gary Russell * @since 2.0 */ public class FtpInboundChannelAdapterParser extends AbstractRemoteFileInboundChannelAdapterParser { - private static final String BASE_PACKAGE = "org.springframework.integration.ftp"; - - @Override protected String getMessageSourceClassname() { - return BASE_PACKAGE + ".inbound.FtpInboundFileSynchronizingMessageSource"; + return FtpInboundFileSynchronizingMessageSource.class.getName(); } @Override protected String getInboundFileSynchronizerClassname() { - return BASE_PACKAGE + ".inbound.FtpInboundFileSynchronizer"; + return FtpInboundFileSynchronizer.class.getName(); } @Override protected String getSimplePatternFileListFilterClassname() { - return BASE_PACKAGE + ".filters.FtpSimplePatternFileListFilter"; + return FtpSimplePatternFileListFilter.class.getName(); } @Override protected String getRegexPatternFileListFilterClassname() { - return BASE_PACKAGE + ".filters.FtpRegexPatternFileListFilter"; + return FtpRegexPatternFileListFilter.class.getName(); } } diff --git a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.2.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.2.xsd index 233c830b83..2463f4ba5a 100644 --- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.2.xsd +++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.2.xsd @@ -218,6 +218,42 @@ + + + + SpEL expression to be executed after the message has been sent. If running in a transactional + poller, it will be executed after the transaction commits. If running in a non-transactional + poller it will execute after the message is sent. Note that the actual point of execution + depends on any asynchronous handoffs on the downstream flow. It will be executed when the + current thread returns from the channel send. The root object of the expression is the + original message (with a File payload). Examples: "payload.delete()", + "payload.renameTo('/foo/bar/' + payload.name)", "@someBean.doSomething(payload)". + + + + + + + If a 'disposition-expression' is provided, and that expression returns a result, the result + is sent to this channel, with the original message payload and a 'file_dispositionResult' + header containing the result of the expression execution. + + + + + + + + + + + + If a 'disposition-expression' is provided, and that expression returns a result, the result + is sent to this disposition-result-channel. This timeout specifies how long to wait if + that channel might block (such as a bounded queue channel that is full). Default infinity. + + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml index c5b39ac817..c817e8d4cd 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml @@ -24,9 +24,14 @@ local-filename-generator-expression="#this.toUpperCase() + '.a'" comparator="comparator" temporary-file-suffix=".foo" - remote-directory="foo/bar"> + remote-directory="foo/bar" + disposition-expression="'foo'" + disposition-result-channel="dispoChannel" + disposition-send-timeout="123"> + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java index 7049eee7d3..76b885578d 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java @@ -18,7 +18,6 @@ package org.springframework.integration.ftp.config; import static junit.framework.Assert.assertEquals; import static junit.framework.Assert.assertNotNull; - import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; @@ -30,12 +29,13 @@ import java.util.Comparator; import java.util.Map; import org.junit.Test; - import org.springframework.beans.factory.FactoryBean; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.expression.Expression; import org.springframework.integration.MessageChannel; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; @@ -54,7 +54,7 @@ public class FtpInboundChannelAdapterParserTests { @SuppressWarnings("unchecked") @Test public void testFtpInboundChannelAdapterComplete() throws Exception{ - ApplicationContext ac = + ApplicationContext ac = new ClassPathXmlApplicationContext("FtpInboundChannelAdapterParserTests-context.xml", this.getClass()); SourcePollingChannelAdapter adapter = ac.getBean("ftpInbound", SourcePollingChannelAdapter.class); assertFalse(TestUtils.getPropertyValue(adapter, "autoStartup", Boolean.class)); @@ -64,10 +64,10 @@ public class FtpInboundChannelAdapterParserTests { assertEquals("ftp:inbound-channel-adapter", adapter.getComponentType()); assertNotNull(TestUtils.getPropertyValue(adapter, "poller")); assertEquals(ac.getBean("ftpChannel"), TestUtils.getPropertyValue(adapter, "outputChannel")); - FtpInboundFileSynchronizingMessageSource inbound = + FtpInboundFileSynchronizingMessageSource inbound = (FtpInboundFileSynchronizingMessageSource) TestUtils.getPropertyValue(adapter, "source"); - - FtpInboundFileSynchronizer fisync = + + FtpInboundFileSynchronizer fisync = (FtpInboundFileSynchronizer) TestUtils.getPropertyValue(inbound, "synchronizer"); assertNotNull(TestUtils.getPropertyValue(fisync, "localFilenameGeneratorExpression")); assertEquals(".foo", TestUtils.getPropertyValue(fisync, "temporaryFileSuffix", String.class)); @@ -78,6 +78,12 @@ public class FtpInboundChannelAdapterParserTests { assertNotNull(filter); Object sessionFactory = TestUtils.getPropertyValue(fisync, "sessionFactory"); assertTrue(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass())); + FileReadingMessageSource source = TestUtils.getPropertyValue(inbound, "fileSource", FileReadingMessageSource.class); + assertEquals("foo", TestUtils.getPropertyValue(source, "dispositionExpression", Expression.class).getValue()); + assertSame(ac.getBean("dispoChannel"), TestUtils.getPropertyValue( + TestUtils.getPropertyValue(source, "dispositionMessagingTemplate"), "defaultChannel")); + assertEquals(123L, TestUtils.getPropertyValue( + TestUtils.getPropertyValue(source, "dispositionMessagingTemplate"), "sendTimeout")); } @Test @@ -87,7 +93,7 @@ public class FtpInboundChannelAdapterParserTests { SourcePollingChannelAdapter adapter = ac.getBean("simpleAdapter", SourcePollingChannelAdapter.class); Object sessionFactory = TestUtils.getPropertyValue(adapter, "source.synchronizer.sessionFactory"); assertEquals(CachingSessionFactory.class, sessionFactory.getClass()); - FtpInboundFileSynchronizer fisync = + FtpInboundFileSynchronizer fisync = TestUtils.getPropertyValue(adapter, "source.synchronizer", FtpInboundFileSynchronizer.class); String remoteFileSeparator = (String) TestUtils.getPropertyValue(fisync, "remoteFileSeparator"); assertNotNull(remoteFileSeparator); @@ -96,7 +102,7 @@ public class FtpInboundChannelAdapterParserTests { @Test public void testFtpInboundChannelAdapterCompleteNoId() throws Exception{ - ApplicationContext ac = + ApplicationContext ac = new ClassPathXmlApplicationContext("FtpInboundChannelAdapterParserTests-context.xml", this.getClass()); Map spcas = ac.getBeansOfType(SourcePollingChannelAdapter.class); SourcePollingChannelAdapter adapter = null; diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java index 51c95948d8..39874bcefe 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -17,36 +17,38 @@ package org.springframework.integration.sftp.config; import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser; +import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter; +import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter; +import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer; +import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource; /** * Parser for 'sftp:inbound-channel-adapter' - * + * * @author Mark Fisher + * @author Gary Russell * @since 2.0 */ public class SftpInboundChannelAdapterParser extends AbstractRemoteFileInboundChannelAdapterParser { - private static final String BASE_PACKAGE = "org.springframework.integration.sftp"; - - @Override protected String getMessageSourceClassname() { - return BASE_PACKAGE + ".inbound.SftpInboundFileSynchronizingMessageSource"; + return SftpInboundFileSynchronizingMessageSource.class.getName(); } @Override protected String getInboundFileSynchronizerClassname() { - return BASE_PACKAGE + ".inbound.SftpInboundFileSynchronizer"; + return SftpInboundFileSynchronizer.class.getName(); } @Override protected String getSimplePatternFileListFilterClassname() { - return BASE_PACKAGE + ".filters.SftpSimplePatternFileListFilter"; + return SftpSimplePatternFileListFilter.class.getName(); } @Override protected String getRegexPatternFileListFilterClassname() { - return BASE_PACKAGE + ".filters.SftpRegexPatternFileListFilter"; + return SftpRegexPatternFileListFilter.class.getName(); } } diff --git a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.2.xsd b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.2.xsd index 4f19ca2365..3e351087f6 100644 --- a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.2.xsd +++ b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.2.xsd @@ -221,6 +221,42 @@ + + + + SpEL expression to be executed after the message has been sent. If running in a transactional + poller, it will be executed after the transaction commits. If running in a non-transactional + poller it will execute after the message is sent. Note that the actual point of execution + depends on any asynchronous handoffs on the downstream flow. It will be executed when the + current thread returns from the channel send. The root object of the expression is the + original message (with a File payload). Examples: "payload.delete()", + "payload.renameTo('/foo/bar/' + payload.name)", "@someBean.doSomething(payload)". + + + + + + + If a 'disposition-expression' is provided, and that expression returns a result, the result + is sent to this channel, with the original message payload and a 'file_dispositionResult' + header containing the result of the expression execution. + + + + + + + + + + + + If a 'disposition-expression' is provided, and that expression returns a result, the result + is sent to this disposition-result-channel. This timeout specifies how long to wait if + that channel might block (such as a bounded queue channel that is full). Default infinity. + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml index e59cdf71a3..7332dcd96b 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml @@ -48,9 +48,14 @@ local-filename-generator-expression="#this.toUpperCase() + '.a'" temporary-file-suffix=".bar" comparator="comparator" - delete-remote-files="${delete.remote.files}"> + delete-remote-files="${delete.remote.files}" + disposition-expression="'foo'" + disposition-result-channel="dispoChannel" + disposition-send-timeout="123"> + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java index 5b03ba4c4f..055d1cf794 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java @@ -29,14 +29,15 @@ import org.junit.After; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; - import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.expression.Expression; import org.springframework.integration.MessageChannel; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer; import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource; import org.springframework.integration.test.util.TestUtils; @@ -46,17 +47,17 @@ import org.springframework.integration.test.util.TestUtils; * @author Gary Russell */ public class InboundChannelAdapterParserTests { - + @Before public void prepare(){ new File("foo").delete(); } - + @Test public void testAutoStartup() throws Exception{ ApplicationContext context = new ClassPathXmlApplicationContext("SftpInboundAutostartup-context.xml", this.getClass()); - + SourcePollingChannelAdapter adapter = context.getBean("sftpAutoStartup", SourcePollingChannelAdapter.class); assertFalse(adapter.isRunning()); } @@ -67,10 +68,10 @@ public class InboundChannelAdapterParserTests { ApplicationContext context = new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass()); assertTrue(new File("src/main/resources").exists()); - + Object adapter = context.getBean("sftpAdapterAutoCreate"); assertTrue(adapter instanceof SourcePollingChannelAdapter); - SftpInboundFileSynchronizingMessageSource source = + SftpInboundFileSynchronizingMessageSource source = (SftpInboundFileSynchronizingMessageSource) TestUtils.getPropertyValue(adapter, "source"); assertNotNull(source); Comparator comparator = TestUtils.getPropertyValue(adapter, "source.fileSource.toBeReceived.q.comparator", Comparator.class); @@ -83,6 +84,12 @@ public class InboundChannelAdapterParserTests { assertEquals(".", remoteFileSeparator); PollableChannel requestChannel = context.getBean("requestChannel", PollableChannel.class); assertNotNull(requestChannel.receive(2000)); + FileReadingMessageSource fileSource = TestUtils.getPropertyValue(source, "fileSource", FileReadingMessageSource.class); + assertEquals("foo", TestUtils.getPropertyValue(fileSource, "dispositionExpression", Expression.class).getValue()); + assertSame(context.getBean("dispoChannel"), TestUtils.getPropertyValue( + TestUtils.getPropertyValue(fileSource, "dispositionMessagingTemplate"), "defaultChannel")); + assertEquals(123L, TestUtils.getPropertyValue( + TestUtils.getPropertyValue(fileSource, "dispositionMessagingTemplate"), "sendTimeout")); } @Test @@ -108,19 +115,19 @@ public class InboundChannelAdapterParserTests { new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass()); assertTrue(new File("target").exists()); } - + @Test public void testLocalDirAutoCreated() throws Exception{ assertFalse(new File("foo").exists()); new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass()); assertTrue(new File("foo").exists()); } - + @Test(expected=BeanCreationException.class) public void testLocalDirAutoCreateFailed() throws Exception{ new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context-fail-autocreate.xml", this.getClass()); } - + @After public void cleanUp() throws Exception{ new File("foo").delete();