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

INT-2892 Add 'local-filter' to (S)FTP Adapters

Browse Source 
Allow the replacement of the default AcceptOnceFileListFilter.
This commit is contained in:
Gary Russell
2013-04-17 11:51:25 -04:00
committed by Gunnar Hillert
parent e6753efd50
commit 14a7953ff6
11 changed files with 122 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2013 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.
@@ -31,6 +31,7 @@ import org.w3c.dom.Element;
*
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Gary Russell
* @since 2.0
*/
public abstract class AbstractRemoteFileInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@@ -62,6 +63,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
if (StringUtils.hasText(comparator)) {
messageSourceBuilder.addConstructorArgReference(comparator);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(messageSourceBuilder, element, "local-filter");
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "local-directory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "auto-create-local-directory");
String localFileGeneratorExpression = element.getAttribute("local-filename-generator-expression");

18
spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java
View File

@@ -79,6 +79,8 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends M
*/
private final FileReadingMessageSource fileSource;
private volatile FileListFilter<File> localFileListFilter = new AcceptOnceFileListFilter<File>();
public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer<F> synchronizer) {
this(synchronizer, null);
@@ -104,6 +106,20 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends M
this.localDirectory = localDirectory;
}
/**
* A {@link FileListFilter} used to determine which files will generate messages
* after they have been synchronized. It will be combined with a filter that
* will prevent accessing files that are in the process of being synchronized
* (files having the {@link AbstractInboundFileSynchronizer#getTemporaryFileSuffix()}).
* <p>
* The default is an {@link AcceptOnceFileListFilter} which filters duplicate file
* names (processed during the current execution).
* @param localFileListFilter
*/
public void setLocalFilter(FileListFilter<File> localFileListFilter) {
this.localFileListFilter = localFileListFilter;
}
@Override
protected void onInit() {
Assert.notNull(this.localDirectory, "localDirectory must not be null");
@@ -153,7 +169,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends M
private FileListFilter<File> buildFilter() {
Pattern completePattern = Pattern.compile("^.*(?<!" + this.synchronizer.getTemporaryFileSuffix() + ")$");
return new CompositeFileListFilter<File>(Arrays.asList(
new AcceptOnceFileListFilter<File>(),
this.localFileListFilter,
new RegexPatternFileListFilter(completePattern)));
}

23
spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-3.0.xsd
View File

@@ -190,7 +190,28 @@
<xsd:documentation>
Allows you to specify a reference to a
[org.springframework.integration.file.filters.FileListFilter]
bean.
bean. This filter is applied to files on the remote server and
only files that pass the filter are retrieved.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="local-filter" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.file.filters.FileListFilter" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Allows you to specify a reference to a
[org.springframework.integration.file.filters.FileListFilter]
bean. This filter is applied to files after they have been
retrieved. The default is an AcceptOnceFileListFilter which means that,
even if a new instance of a file is retrieved from the remote server,
a message won't be generated. The filter provided here is combined
with a filter that prevents the message source from processing
files that are currently being downloaded.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

3
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml
View File

@@ -24,11 +24,14 @@
local-filename-generator-expression="#this.toUpperCase() + '.a'"
comparator="comparator"
temporary-file-suffix=".foo"
local-filter="acceptAllFilter"
remote-directory="foo/bar">
<int:poller fixed-rate="1000">
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-ftp:inbound-channel-adapter>
<bean id="acceptAllFilter" class="org.springframework.integration.file.filters.AcceptAllFileListFilter" />
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit expression="'foo'" channel="successChannel"/>

4
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
View File

@@ -24,6 +24,7 @@ import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.Collection;
import java.util.Comparator;
import java.util.Map;
import java.util.concurrent.PriorityBlockingQueue;
@@ -34,6 +35,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.remote.session.CachingSessionFactory;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
@@ -77,6 +79,8 @@ public class FtpInboundChannelAdapterParserTests {
assertNotNull(filter);
Object sessionFactory = TestUtils.getPropertyValue(fisync, "sessionFactory");
assertTrue(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass()));
FileListFilter<?> acceptAllFilter = ac.getBean("acceptAllFilter", FileListFilter.class);
assertTrue(TestUtils.getPropertyValue(inbound, "fileSource.scanner.filter.fileFilters", Collection.class).contains(acceptAllFilter));
}
@Test

23
spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-3.0.xsd
View File

@@ -151,7 +151,8 @@
<xsd:documentation>
Allows you to specify a reference to a
[org.springframework.integration.file.filters.FileListFilter]
bean.
bean. This filter is applied to files on the remote server and
only files that pass the filter are retrieved.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -195,6 +196,26 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="local-filter" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.file.filters.FileListFilter" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Allows you to specify a reference to a
[org.springframework.integration.file.filters.FileListFilter]
bean. This filter is applied to files after they have been
retrieved. The default is an AcceptOnceFileListFilter which means that,
even if a new instance of a file is retrieved from the remote server,
a message won't be generated. The filter provided here is combined
with a filter that prevents the message source from processing
files that are currently being downloaded.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="local-directory" type="xsd:string"
use="required">
<xsd:annotation>

3
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml
View File

@@ -48,11 +48,14 @@
local-filename-generator-expression="#this.toUpperCase() + '.a'"
temporary-file-suffix=".bar"
comparator="comparator"
local-filter="acceptAllFilter"
delete-remote-files="${delete.remote.files}">
<poller fixed-rate="1000">
<transactional synchronization-factory="syncFactory"/>
</poller>
</sftp:inbound-channel-adapter>
<beans:bean id="acceptAllFilter" class="org.springframework.integration.file.filters.AcceptAllFileListFilter" />
<transaction-synchronization-factory id="syncFactory">
<after-commit expression="'foo'" channel="successChannel"/>

7
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java
View File

@@ -19,10 +19,11 @@ package org.springframework.integration.sftp.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.util.Collection;
import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;
@@ -30,7 +31,6 @@ 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;
@@ -38,6 +38,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource;
import org.springframework.integration.test.util.TestUtils;
@@ -87,6 +88,8 @@ public class InboundChannelAdapterParserTests {
assertEquals(".", remoteFileSeparator);
PollableChannel requestChannel = context.getBean("requestChannel", PollableChannel.class);
assertNotNull(requestChannel.receive(2000));
FileListFilter<?> acceptAllFilter = context.getBean("acceptAllFilter", FileListFilter.class);
assertTrue(TestUtils.getPropertyValue(source, "fileSource.scanner.filter.fileFilters", Collection.class).contains(acceptAllFilter));
}
@Test

18
src/reference/docbook/ftp.xml
View File

@@ -151,6 +151,7 @@ protected void postProcessClientBeforeConnect(T client) throws IOException {
remote-directory="some/remote/path"
remote-file-separator="/"
local-filename-generator-expression="#this.toUpperCase() + '.a'"
local-filter="myFilter"
local-directory=".">
<int:poller fixed-rate="1000"/>
</int-ftp:inbound-channel-adapter>]]></programlisting>
@@ -174,13 +175,20 @@ protected void postProcessClientBeforeConnect(T client) throws IOException {
(e.g. <code>filename-regex=".*\.test$"</code>). And of course if you need complete control you can use <code>filter</code>
attribute and provide a reference to any custom implementation of the
<classname>org.springframework.integration.file.filters.FileListFilter</classname>, a strategy interface for filtering a
list of files.
list of files. This filter determines which remote files are retrieved.
</para>
<note>
As of Spring Integration 2.0.2, we have added a 'remote-file-separator' attribute. That allows you to configure a
file separator character to use if the default '/' is not applicable for your particular environment.
<note>
Beginning with 3.0, you can also specify a filter used to filter the files locally, once they have
been retrieved. The default filter is an <classname>AcceptOnceFileListFilter</classname> which prevents processing
files with the same name multiple times in the same JVM execution; this can now be overridden
(for example with an <classname>AcceptAllFileListFilter</classname>), using the <code>local-filter</code> attribute.
Previously, the default <classname>AcceptOnceFileListFilter</classname> could not be overridden.
</note>
<para>
<para>
The 'remote-file-separator' attribute allows you to configure a
file separator character to use if the default '/' is not applicable for your particular environment.
</para>
<para>
Please refer to the schema for more details on these attributes.
</para>
<para>

10
src/reference/docbook/sftp.xml
View File

@@ -227,6 +227,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
local-directory="file:target/foo"
auto-create-local-directory="true"
local-filename-generator-expression="#this.toUpperCase() + '.a'"
local-filter="myFilter"
delete-remote-files="false">
<int:poller fixed-rate="1000"/>
</int-sftp:inbound-channel-adapter>]]></programlisting>
@@ -251,8 +252,15 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
(e.g. <code>filename-regex=".*\.test$"</code>). And of course if you need complete control you can use the <code>filter</code>
attribute to provide a reference to a custom implementation of the
<classname>org.springframework.integration.file.filters.FileListFilter</classname> - a strategy interface for filtering a
list of files.
list of files. This filter determines which remote files are retrieved.
</para>
<note>
Beginning with 3.0, you can also specify a filter used to filter the files locally, once they have
been retrieved. The default filter is an <classname>AcceptOnceFileListFilter</classname> which prevents processing
files with the same name multiple times in the same JVM execution; this can now be overridden
(for example with an <classname>AcceptAllFileListFilter</classname>), using the <code>local-filter</code> attribute.
Previously, the default <classname>AcceptOnceFileListFilter</classname> could not be overridden.
</note>
<para>
Please refer to the schema for more detail on these attributes.
</para>

21
src/reference/docbook/whats-new.xml
View File

@@ -88,6 +88,27 @@
URI-schemes supported by Spring Web Services. For more information see <xref linkend="outbound-uri"/>.
</para>
</section>
<section id="3.0-xFTP-ib">
<title>(S)FTP(S) Inbound Adapters</title>
<para>
Previously, there was no way to override the default filter used to process files retrieved
from a remote server. The <code>filter</code> attribute determines which files are retrieved
but the <classname>FileReadingMessageSource</classname> uses an
<classname>AcceptOnceFileListFilter</classname>. This means that if a new copy of a file
is retrieved, with the same name as a previously copied file, no message was sent from the
adapter.
</para>
<para>
With this release, a new attribute <code>local-filter</code> allows you to override the
default filter, for example with an <classname>AcceptAllFileListFilter</classname>, or some
other custom filter.
</para>
<para>
For users that wish the behavior of the <classname>AcceptOnceFileListFilter</classname>
to be maintained across JVM executions, a custom filter that retains state, perhaps on
the file system, can now be configured.
</para>
</section>
<section id="3.0-xFTP-gw">
<title>(S)FTP(S) Gateways</title>
<para>