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

INT-1614 fixed issue with FTP retrieve stream

Browse Source
This commit is contained in:
Mark Fisher
2010-11-19 20:02:55 -05:00
parent fc9fff58d4
commit 0a6842dd7d
5 changed files with 26 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<F> 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<F> 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<F> 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 <emphasis>source</emphasis>
* entry on the remote file system. This is the file-system version of an
* <code>ack-mode</code>. 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
* <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 EntryAcknowledgmentStrategy<F> {
/**
* 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;
}

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

@@ -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());
}
}
}

2
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/outbound/FtpSendingMessageHandler.java
View File

@@ -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";

4
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java
View File

@@ -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()) {

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

@@ -139,10 +139,9 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundFileSynchronizer.EntryAcknowledgmentStrategy<ChannelSftp.LsEntry> {
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());
}