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

INT-1614 replaced AcknowledgmentStrategy with simple delete flag option only

Browse Source
This commit is contained in:
Mark Fisher
2010-11-21 13:53:56 -05:00
parent 9c0e122958
commit 9e3a039d2b
4 changed files with 21 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java
View File

@@ -24,7 +24,6 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.remote.session.Session;
/**
* Base class charged with knowing how to connect to a remote file system,
@@ -35,89 +34,35 @@ import org.springframework.integration.file.remote.session.Session;
* ensure the file entry is acceptable.
*
* @author Josh Long
* @author Mark Fisher
* @since 2.0
*/
public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileSynchronizer, InitializingBean {
protected final Log logger = LogFactory.getLog(this.getClass());
/**
* Should we <emphasis>delete</emphasis> the <b>source</b> file? For an FTP
* server, for example, this would delete the original FTPFile instance.
*/
protected boolean shouldDeleteSourceFile;
/**
* An {@link FileListFilter} that runs against the <emphasis>remote</emphasis> file system view.
*/
private volatile FileListFilter<F> filter;
/**
* The {@link AcknowledgmentStrategy} implementation.
* Should we <emphasis>delete</emphasis> the <b>source</b> file? For an FTP
* server, for example, this would delete the original FTPFile instance.
*/
private AcknowledgmentStrategy<F> acknowledgmentStrategy;
protected boolean shouldDeleteSourceFile;
public void setFilter(FileListFilter<F> filter) {
this.filter = filter;
}
public void setAcknowledgmentStrategy(AcknowledgmentStrategy<F> acknowledgmentStrategy) {
this.acknowledgmentStrategy = acknowledgmentStrategy;
}
public void setShouldDeleteSourceFile(boolean shouldDeleteSourceFile) {
this.shouldDeleteSourceFile = shouldDeleteSourceFile;
}
/**
* @param session
* session that was used to retrieve the file. Will be passed to the {@link EntryAcknowledgmentStrategy}.
* The {@link EntryAcknowledgmentStrategy#acknowledge(Object, Object)} will be called in line with the
* {@link org.springframework.integration.core.MessageSource#receive()} call so this could conceivably
* be a 'live' stateful client (a connection?) that is inappropriate to cache as it has per-request state.
* @param file
* leverages strategy implementations to enable different
* behavior. It's a hook to the file entry after it's been
* successfully downloaded. Conceptually, you might delete the
* remote one or rename it, etc.
* @throws Exception
* escape hatch exception, let the adapter deal with it.
*/
protected final void acknowledge(Session session, F file) throws Exception {
if (this.acknowledgmentStrategy != null) {
this.acknowledgmentStrategy.acknowledge(session, file);
}
}
protected final List<F> filterFiles(F[] files) {
return (this.filter != null) ? this.filter.filterFiles(files) : Arrays.asList(files);
}
/**
* Strategy interface to expose a hook for dispatching, moving, or deleting
* the file once it has been delivered. Adapters should (for consistency)
* expose an attribute dictating whether the adapter will delete the
* <emphasis>source</emphasis> entry on the remote file system. This is the
* file-system version of an <code>ack-mode</code>.
*
* @param <F> the file entry type (file, sftp, ftp, ...)
*/
public static interface AcknowledgmentStrategy<F> {
/**
* Semantics are simple. You get a pointer to the file just processed
* and the FTP Session that processed it.
*
* @param session
* the FTP session
* @param file
* the file that has been processed
* @throws Exception in case of an error while acknowledging
*/
void acknowledge(Session session, F file) throws Exception;
}
}

27
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java
View File

@@ -23,8 +23,6 @@ import java.io.IOException;
import java.io.InputStream;
import java.util.Collection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.integration.MessagingException;
@@ -64,9 +62,6 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer<
public void afterPropertiesSet() {
Assert.notNull(this.sessionFactory, "sessionFactory must not be null");
if (this.shouldDeleteSourceFile) {
this.setAcknowledgmentStrategy(new DeletionAcknowledgmentStrategy());
}
}
public void synchronizeToLocalDirectory(File localDirectory) {
@@ -114,7 +109,9 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer<
return false;
}
FileCopyUtils.copy(inputStream, fileOutputStream);
acknowledge(session, ftpFile);
if (this.shouldDeleteSourceFile) {
this.deleteRemoteFile(session, ftpFile);
}
}
catch (Exception e) {
if (e instanceof RuntimeException){
@@ -133,19 +130,11 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer<
return false;
}
/**
* An acknowledgment strategy that deletes the file.
*/
private static class DeletionAcknowledgmentStrategy implements AcknowledgmentStrategy<FTPFile> {
private final Log logger = LogFactory.getLog(this.getClass());
public void acknowledge(Session session, FTPFile ftpFile) throws Exception {
if ((ftpFile != null) && session.rm(ftpFile.getName())) {
if (logger.isDebugEnabled()) {
logger.debug("deleted " + ftpFile.getName());
}
// TODO: make this an abstract method in the base class once the code that calls this is refactored upward
private void deleteRemoteFile(Session session, FTPFile ftpFile) {
if ((ftpFile != null) && session.rm(ftpFile.getName())) {
if (logger.isDebugEnabled()) {
logger.debug("deleted " + ftpFile.getName());
}
}
}

21
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java
View File

@@ -63,9 +63,6 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
public void afterPropertiesSet() throws Exception {
Assert.notNull(this.remotePath, "'remotePath' must not be null");
if (this.shouldDeleteSourceFile) {
this.setAcknowledgmentStrategy(new DeletionAcknowledgmentStrategy());
}
}
public void synchronizeToLocalDirectory(File localDirectory) {
@@ -122,7 +119,9 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
fileOutputStream.close();
}
if (tmpLocalTarget.renameTo(localFile)) {
this.acknowledge(session, entry);
if (this.shouldDeleteSourceFile) {
this.deleteRemoteFile(session, entry);
}
}
return true;
}
@@ -140,15 +139,11 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
}
}
private class DeletionAcknowledgmentStrategy implements AcknowledgmentStrategy<ChannelSftp.LsEntry> {
public void acknowledge(Session session, ChannelSftp.LsEntry msg) throws Exception {
String remoteFqPath = remotePath + "/" + msg.getFilename();
session.rm(remoteFqPath);
if (logger.isDebugEnabled()) {
logger.debug("deleted " + msg.getFilename());
}
private void deleteRemoteFile(Session session, ChannelSftp.LsEntry msg) {
String remoteFqPath = remotePath + "/" + msg.getFilename();
session.rm(remoteFqPath);
if (logger.isDebugEnabled()) {
logger.debug("deleted " + msg.getFilename());
}
}

2
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java
View File

@@ -29,7 +29,6 @@ import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.AcknowledgmentStrategy;
import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
import org.springframework.util.ReflectionUtils;
@@ -88,7 +87,6 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
@Test
public void testCopyAndRenameWhenLocalFileDoesntExist() throws Exception {
SftpInboundFileSynchronizer synchronizer = new SftpInboundFileSynchronizer(mock(SessionFactory.class));
synchronizer.setAcknowledgmentStrategy(mock(AcknowledgmentStrategy.class));
Method method =
ReflectionUtils.findMethod(synchronizer.getClass(), "copyFromRemoteToLocalDirectory", Session.class, LsEntry.class, Resource.class);
method.setAccessible(true);