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

INT-3827: ResettableFileListFilter

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3827

Provide a hook to enable removing a file from an `AcceptOnceFileListFilter`,
for example after a message processing failure.

Make the `CompositeFileListFilter` a `ReversibleFileListFilter` so it can
delegate to any of its composed filters that are reversible.

INT-3827: Polishing - PR Comments
This commit is contained in:
Gary Russell
2015-09-29 12:08:42 -04:00
committed by Artem Bilan
parent 4c171a54ed
commit 0d721739e9
12 changed files with 284 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java
View File

@@ -35,7 +35,7 @@ import org.springframework.util.Assert;
*
*/
public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractFileListFilter<F>
implements ReversibleFileListFilter<F>, Closeable {
implements ReversibleFileListFilter<F>, ResettableFileListFilter<F>, Closeable {
protected final ConcurrentMetadataStore store;
@@ -102,12 +102,18 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
rollingBack = true;
}
if (rollingBack) {
this.store.remove(buildKey(fileToRollback));
flushIfNeeded();
remove(fileToRollback);
}
}
}
@Override
public boolean remove(F fileToRemove) {
String removed = this.store.remove(buildKey(fileToRemove));
flushIfNeeded();
return removed != null;
}
@Override
public void close() throws IOException {
if (this.store instanceof Closeable) {

19
spring-integration-file/src/main/java/org/springframework/integration/file/filters/AcceptOnceFileListFilter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -34,7 +34,8 @@ import java.util.concurrent.LinkedBlockingQueue;
* @author Gary Russell
* @since 1.0.0
*/
public class AcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> implements ReversibleFileListFilter<F> {
public class AcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> implements ReversibleFileListFilter<F>,
ResettableFileListFilter<F> {
private final Queue<F> seen;
@@ -93,13 +94,19 @@ public class AcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> imple
rollingBack = true;
}
if (rollingBack) {
this.seenSet.remove(fileToRollback);
if (this.seen != null) {
this.seen.remove(fileToRollback);
}
remove(fileToRollback);
}
}
}
}
@Override
public boolean remove(F fileToRemove) {
boolean removed = this.seenSet.remove(fileToRemove);
if (this.seen != null) {
this.seen.remove(fileToRemove);
}
return removed;
}
}

11
spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java
View File

@@ -39,7 +39,7 @@ import org.springframework.util.Assert;
*
* @param <F> The type that will be filtered.
*/
public class CompositeFileListFilter<F> implements FileListFilter<F>, Closeable {
public class CompositeFileListFilter<F> implements ReversibleFileListFilter<F>, Closeable {
private final Set<FileListFilter<F>> fileFilters;
@@ -111,4 +111,13 @@ public class CompositeFileListFilter<F> implements FileListFilter<F>, Closeable
return results;
}
@Override
public void rollback(F file, List<F> files) {
for (FileListFilter<F> fileFilter : this.fileFilters) {
if (fileFilter instanceof ReversibleFileListFilter) {
((ReversibleFileListFilter<F>) fileFilter).rollback(file, files);
}
}
}
}

34
spring-integration-file/src/main/java/org/springframework/integration/file/filters/ResettableFileListFilter.java Normal file
View File

@@ -0,0 +1,34 @@
/*
* Copyright 2015 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.filters;
/**
* A {@link FileListFilter} that can be reset by removing a specific file from its
* state.
* @author Gary Russell
* @since 4.1.7
*
*/
public interface ResettableFileListFilter<F> extends FileListFilter<F> {
/**
* Remove the specified file from the filter so it will pass on the next attempt.
* @param f the element to remove.
* @return true if the file was removed as a result of this call.
*/
boolean remove(F f);
}

9
spring-integration-file/src/test/java/org/springframework/integration/file/filters/AcceptOnceFileListFilterTests.java
View File

@@ -24,6 +24,7 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.Set;
@@ -76,6 +77,14 @@ public class AcceptOnceFileListFilterTests {
doTestRollback(filter);
}
@Test
public void testRollbackComposite() {
AcceptOnceFileListFilter<String> filter = new AcceptOnceFileListFilter<String>();
CompositeFileListFilter<String> composite = new CompositeFileListFilter<String>(
Collections.singletonList(filter));
doTestRollback(composite);
}
protected void doTestRollback(ReversibleFileListFilter<String> filter) {
String[] files = new String[] {"foo", "bar", "baz"};
List<String> passed = filter.filterFiles(files);

1
spring-integration-sftp/.gitignore vendored
View File

@@ -1,3 +1,4 @@
local-test-dir/*.test
local-test-dir/rollback/*.txt
remote-target-dir/*foo*
remote-target-dir/*test*

2
spring-integration-sftp/local-test-dir/readme.txt
View File

@@ -1 +1 @@
Don't delete. This directory is used by test cases
don't delete this dir used for testing

1
spring-integration-sftp/local-test-dir/rollback/sftpSource1.txt
View File

@@ -1 +0,0 @@
source1

39
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml Normal file
View File

@@ -0,0 +1,39 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd">
<int-sftp:inbound-channel-adapter id="sftpAdapterAutoCreate"
session-factory="sftpSessionFactory"
channel="requestChannel"
remote-directory-expression="'/sftpSource'"
local-directory="file:local-test-dir/rollback"
auto-create-local-directory="true"
filename-pattern="sftpSource1.txt"
local-filter="acceptOnceFilter">
<int:poller fixed-rate="1000" max-messages-per-poll="2" error-channel="nullChannel">
<int:transactional synchronization-factory="syncFactory" />
</int:poller>
</int-sftp:inbound-channel-adapter>
<int:channel id="requestChannel" />
<int:service-activator input-channel="requestChannel" ref="crash" method="handle" />
<bean id="crash" class="org.springframework.integration.sftp.inbound.RollbackLocalFilterTests$Crash" />
<bean id="acceptOnceFilter" class="org.springframework.integration.file.filters.AcceptOnceFileListFilter" />
<int:transaction-synchronization-factory id="syncFactory">
<int:after-rollback expression="@acceptOnceFilter.remove(payload)" />
</int:transaction-synchronization-factory>
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager" />
<bean id="sftpServerConfig" class="org.springframework.integration.sftp.TestSftpServerConfig" />
</beans>

76
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java Normal file
View File

@@ -0,0 +1,76 @@
/*
* Copyright 2015 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.sftp.inbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 4.1.7
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class RollbackLocalFilterTests {
@Autowired
private Crash crash;
@Test
public void testRollback() throws Exception {
assertTrue(this.crash.getLatch().await(10, TimeUnit.SECONDS));
assertEquals("sftpSource1.txt", this.crash.getFile().getName());
}
public static class Crash {
private final CountDownLatch latch = new CountDownLatch(2);
private final AtomicBoolean shouldCrash = new AtomicBoolean();
private volatile File file;
public CountDownLatch getLatch() {
return latch;
}
public File getFile() {
return file;
}
public void handle(File in) {
latch.countDown();
if (this.shouldCrash.compareAndSet(false, true)) {
throw new RuntimeException("foo");
}
this.file = in;
}
}
}

46
src/reference/asciidoc/ftp.adoc
View File

@@ -274,6 +274,52 @@ Only then is the poll operation considered complete, and the poller will wait fo
You can alternatively set the 'max-messages-per-poll' value to a positive value indicating the upward limit of Messages to be created from files with each poll.
For example, a value of 10 means that on each poll it will attempt to process no more than 10 files.
==== Recovering from Failures
It is important to understand the architecture of the adapter.
There is a file synchronizer which fetches the files, and a `FileReadingMessageSource` to emit a message for each
synchronized file.
As discussed above, there are two filters involved.
The `filter` attribute (and patterns) refers to the remote (FTP) file list - to avoid fetching files that have already
been fetched.
The `local-filter` is used by the `FileReadingMessageSource` to determine which files are to be sent as messages.
The synchronizer lists the remote files and consults its filter; the files are then transferred.
If an IO error occurs during file transfer, any files that have already been added to the filter are removed so they
are eligible to be re-fetched on the next poll.
This only applies if the filter implements `ReversibleFileListFilter` (such as the `AcceptOnceFileListFilter`).
If, after synchronizing the files, an error occurs on the downstream flow processing a file, there is __no__ automatic
rollback of the filter so the failed file will __not__ be reprocessed by default.
If you wish to reprocess such files after a failure, you can use configuration similar to the following to facilitate
the removal of the failed file from the filter.
This will work for any `ResettableFileListFilter`.
[source, xml]
----
<int-ftp:inbound-channel-adapter id="ftpAdapter"
session-factory="ftpSessionFactory"
channel="requestChannel"
remote-directory-expression="'/sftpSource'"
local-directory="file:myLocalDir"
auto-create-local-directory="true"
filename-pattern="*.txt"
local-filter="acceptOnceFilter">
<int:poller fixed-rate="1000">
<int:transactional synchronization-factory="syncFactory" />
</int:poller>
</int-ftp:inbound-channel-adapter>
<bean id="acceptOnceFilter" class="org.springframework.integration.file.filters.AcceptOnceFileListFilter" />
<int:transaction-synchronization-factory id="syncFactory">
<int:after-rollback expression="@acceptOnceFilter.remove(payload)" />
</int:transaction-synchronization-factory>
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager" />
----
[[ftp-outbound]]
=== FTP Outbound Channel Adapter

46
src/reference/asciidoc/sftp.adoc
View File

@@ -359,6 +359,52 @@ If you need a custom filter implementation simply include a reference in your ad
----
==== Recovering from Failures
It is important to understand the architecture of the adapter.
There is a file synchronizer which fetches the files, and a `FileReadingMessageSource` to emit a message for each
synchronized file.
As discussed above, there are two filters involved.
The `filter` attribute (and patterns) refers to the remote (SFTP) file list - to avoid fetching files that have already
been fetched.
The `local-filter` is used by the `FileReadingMessageSource` to determine which files are to be sent as messages.
The synchronizer lists the remote files and consults its filter; the files are then transferred.
If an IO error occurs during file transfer, any files that have already been added to the filter are removed so they
are eligible to be re-fetched on the next poll.
This only applies if the filter implements `ReversibleFileListFilter` (such as the `AcceptOnceFileListFilter`).
If, after synchronizing the files, an error occurs on the downstream flow processing a file, there is _no_ automatic
rollback of the filter so the failed file will _not_ be reprocessed by default.
If you wish to reprocess such files after a failure, you can use configuration similar to the following to facilitate
the removal of the failed file from the filter.
This will work for any `ResettableFileListFilter`.
[source, xml]
----
<int-sftp:inbound-channel-adapter id="sftpAdapter"
session-factory="sftpSessionFactory"
channel="requestChannel"
remote-directory-expression="'/sftpSource'"
local-directory="file:myLocalDir"
auto-create-local-directory="true"
filename-pattern="*.txt"
local-filter="acceptOnceFilter">
<int:poller fixed-rate="1000">
<int:transactional synchronization-factory="syncFactory" />
</int:poller>
</int-sftp:inbound-channel-adapter>
<bean id="acceptOnceFilter" class="org.springframework.integration.file.filters.AcceptOnceFileListFilter" />
<int:transaction-synchronization-factory id="syncFactory">
<int:after-rollback expression="@acceptOnceFilter.remove(payload)" />
</int:transaction-synchronization-factory>
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager" />
----
[[sftp-outbound]]
=== SFTP Outbound Channel Adapter