From 0a6842dd7dabd78cb982e15e4c329976fedf3989 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 19 Nov 2010 20:02:55 -0500 Subject: [PATCH] INT-1614 fixed issue with FTP retrieve stream --- .../AbstractInboundFileSynchronizer.java | 39 ++++++++----------- .../inbound/FtpInboundFileSynchronizer.java | 9 ++--- .../outbound/FtpSendingMessageHandler.java | 2 +- .../integration/ftp/session/FtpSession.java | 4 +- .../inbound/SftpInboundFileSynchronizer.java | 5 +-- 5 files changed, 26 insertions(+), 33 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java index 5b849aff36..4fae5f6479 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java @@ -24,6 +24,7 @@ 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, @@ -70,9 +71,9 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS } /** - * @param usefulContextOrClientData - * this is context information to be passed to the individual {@link EntryAcknowledgmentStrategy}. - * {@link EntryAcknowledgmentStrategy#acknowledge(Object, Object)} will be called in line with the + * @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 @@ -83,9 +84,9 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS * @throws Exception * escape hatch exception, let the adapter deal with it. */ - protected final void acknowledge(Object usefulContextOrClientData, F file) throws Exception { + protected final void acknowledge(Session session, F file) throws Exception { if (this.entryAcknowledgmentStrategy != null) { - this.entryAcknowledgmentStrategy.acknowledge(usefulContextOrClientData, file); + this.entryAcknowledgmentStrategy.acknowledge(session, file); } } @@ -96,32 +97,26 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS /** * Strategy interface to expose a hook for dispatching, moving, or deleting - * the file once it has been delivered. This will typically be a NOOP for the - * implementation. Adapters should (for consistency) expose an attribute - * dictating whether the adapter will delete the source - * entry on the remote file system. This is the file-system version of an - * ack-mode. Future implementations should consider exposing a - * custom attribute that plugs a custom {@link EntryAcknowledgmentStrategy} - * into the pipeline and also some more advanced scenarios (i.e., 'move file - * to another folder on delete ', or 'rename on delete') + * the file once it has been delivered. Adapters should (for consistency) + * expose an attribute dictating whether the adapter will delete the + * source entry on the remote file system. This is the + * file-system version of an ack-mode. * * @param the file entry type (file, sftp, ftp, ...) */ public static interface EntryAcknowledgmentStrategy { /** - * Semantics are simple. You get a pointer to the entry just processed - * and any kind of helper data you could ask for. Since the strategy is - * a singleton and the clients you might ask for as context data are - * pooled, it's not recommended that you try to cache them. + * Semantics are simple. You get a pointer to the file just processed + * and the FTP Session that processed it. * - * @param useful - * any context data - * @param msg - * the data / file / entry you want to process -- specific to subclasses + * @param session + * the FTP session + * @param file + * the file that has been processed * @throws Exception in case of an error while acknowledging */ - void acknowledge(Object useful, F msg) throws Exception; + void acknowledge(Session session, F file) throws Exception; } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java index 88c8488d72..b18325fe50 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java @@ -25,7 +25,6 @@ import java.util.Collection; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; import org.springframework.core.io.Resource; @@ -105,7 +104,6 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer< File file = new File(tempFileName); FileOutputStream fileOutputStream = new FileOutputStream(file); try { - //InputStream inputStream = client.retrieveFileStream(remoteFileName); InputStream inputStream = session.get(remoteFileName); if (inputStream == null) { return false; @@ -138,11 +136,10 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer< private final Log logger = LogFactory.getLog(this.getClass()); - public void acknowledge(Object useful, FTPFile fptFile) throws Exception { - FTPClient ftpClient = (FTPClient) useful; - if ((fptFile != null) && ftpClient.deleteFile(fptFile.getName())) { + public void acknowledge(Session session, FTPFile ftpFile) throws Exception { + if ((ftpFile != null) && session.rm(ftpFile.getName())) { if (logger.isDebugEnabled()) { - logger.debug("deleted " + fptFile.getName()); + logger.debug("deleted " + ftpFile.getName()); } } } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/outbound/FtpSendingMessageHandler.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/outbound/FtpSendingMessageHandler.java index ebb8b611a3..f8163e7f8b 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/outbound/FtpSendingMessageHandler.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/outbound/FtpSendingMessageHandler.java @@ -48,7 +48,7 @@ import org.springframework.util.FileCopyUtils; * @author Josh Long * @author Oleg Zhurakousky */ -public class FtpSendingMessageHandler extends AbstractMessageHandler{ +public class FtpSendingMessageHandler extends AbstractMessageHandler { private static final String TEMPORARY_FILE_SUFFIX = ".writing"; diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java index ffe209c8cd..eabb91cbfe 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java @@ -82,7 +82,9 @@ public class FtpSession implements Session { public InputStream get(String source) { try { - return this.client.retrieveFileStream(source); + InputStream inputStream = this.client.retrieveFileStream(source); + this.client.completePendingCommand(); + return inputStream; } catch (IOException e) { if (logger.isWarnEnabled()) { diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java index 5332076765..a554bf3b9a 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java @@ -139,10 +139,9 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundFileSynchronizer.EntryAcknowledgmentStrategy { - public void acknowledge(Object useful, ChannelSftp.LsEntry msg) throws Exception { - Session sftpSession = (Session) useful; + public void acknowledge(Session session, ChannelSftp.LsEntry msg) throws Exception { String remoteFqPath = remotePath + "/" + msg.getFilename(); - sftpSession.rm(remoteFqPath); + session.rm(remoteFqPath); if (logger.isDebugEnabled()) { logger.debug("deleted " + msg.getFilename()); }