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

INT-1614 polishing

Browse Source
This commit is contained in:
Mark Fisher
2010-11-19 08:24:00 -05:00
parent 1d3445c538
commit 4d05b9c559
3 changed files with 89 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java
View File

@@ -61,14 +61,6 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<F> implements I
protected EntryAcknowledgmentStrategy<F> entryAcknowledgmentStrategy;
public void setEntryAcknowledgmentStrategy(EntryAcknowledgmentStrategy<F> entryAcknowledgmentStrategy) {
this.entryAcknowledgmentStrategy = entryAcknowledgmentStrategy;
}
public void setShouldDeleteSourceFile(boolean shouldDeleteSourceFile) {
this.shouldDeleteSourceFile = shouldDeleteSourceFile;
}
public void setLocalDirectory(Resource localDirectory) {
this.localDirectory = localDirectory;
}
@@ -77,6 +69,14 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<F> implements I
this.filter = filter;
}
public void setEntryAcknowledgmentStrategy(EntryAcknowledgmentStrategy<F> entryAcknowledgmentStrategy) {
this.entryAcknowledgmentStrategy = entryAcknowledgmentStrategy;
}
public void setShouldDeleteSourceFile(boolean shouldDeleteSourceFile) {
this.shouldDeleteSourceFile = shouldDeleteSourceFile;
}
/**
* @param usefulContextOrClientData
* this is context information to be passed to the individual {@link EntryAcknowledgmentStrategy}.
@@ -88,10 +88,10 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<F> implements I
* 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 Throwable
* @throws Exception
* escape hatch exception, let the adapter deal with it.
*/
protected void acknowledge(Object usefulContextOrClientData, F file) throws Throwable {
protected void acknowledge(Object usefulContextOrClientData, F file) throws Exception {
if (this.entryAcknowledgmentStrategy != null) {
this.entryAcknowledgmentStrategy.acknowledge(usefulContextOrClientData, file);
}

73
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java
View File

@@ -22,6 +22,8 @@ import java.io.FileOutputStream;
import java.io.IOException;
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;
@@ -40,7 +42,7 @@ import org.springframework.util.Assert;
*/
public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemoteFileSystemSychronizer<FTPFile> {
protected volatile FtpClientPool clientPool;
private volatile FtpClientPool clientPool;
/**
@@ -59,39 +61,6 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot
}
}
private boolean copyFileToLocalDirectory(FTPClient client, FTPFile ftpFile, Resource localDirectory)
throws IOException, FileNotFoundException {
String remoteFileName = ftpFile.getName();
String localFileName = localDirectory.getFile().getPath() + "/" + remoteFileName;
File localFile = new File(localFileName);
if (!localFile.exists()) {
String tempFileName = localFileName +
AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION;
File file = new File(tempFileName);
FileOutputStream fos = new FileOutputStream(file);
try {
client.retrieveFile(remoteFileName, fos);
// Perhaps we have some dispatch of the source file to do?
acknowledge(client, ftpFile);
}
catch (Throwable th) {
if (th instanceof RuntimeException){
throw (RuntimeException)th;
}
else {
throw new MessagingException("Failed to copy file", th);
}
}
finally {
fos.close();
}
file.renameTo(localFile);
return true;
}
return false;
}
@Override
protected void syncRemoteToLocalFileSystem() {
try {
@@ -117,11 +86,45 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot
}
}
private boolean copyFileToLocalDirectory(FTPClient client, FTPFile ftpFile, Resource localDirectory)
throws IOException, FileNotFoundException {
String remoteFileName = ftpFile.getName();
String localFileName = localDirectory.getFile().getPath() + "/" + remoteFileName;
File localFile = new File(localFileName);
if (!localFile.exists()) {
String tempFileName = localFileName + AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION;
File file = new File(tempFileName);
FileOutputStream fos = new FileOutputStream(file);
try {
client.retrieveFile(remoteFileName, fos);
// Perhaps we have some dispatch of the source file to do?
acknowledge(client, ftpFile);
}
catch (Exception e) {
if (e instanceof RuntimeException){
throw (RuntimeException) e;
}
else {
throw new MessagingException("Failed to copy file", e);
}
}
finally {
fos.close();
}
file.renameTo(localFile);
return true;
}
return false;
}
/**
* An acknowledgment strategy that deletes the file.
*/
private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy<FTPFile> {
private static class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy<FTPFile> {
private final Log logger = LogFactory.getLog(this.getClass());
public void acknowledge(Object useful, FTPFile fptFile) throws Exception {
FTPClient ftpClient = (FTPClient) useful;

76
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java
View File

@@ -47,7 +47,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
* the path on the remote mount
*/
private volatile String remotePath;
private volatile boolean autoCreateDirectories;
/**
@@ -116,39 +116,6 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
return false;
}
private boolean copyFromRemoteToLocalDirectory(SftpSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir) throws Exception {
File fileForLocalDir = localDir.getFile();
File localFile = new File(fileForLocalDir, entry.getFilename());
if (!localFile.exists()) {
InputStream in = null;
FileOutputStream fileOutputStream = null;
try {
File tmpLocalTarget = new File(localFile.getAbsolutePath() +
AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION);
fileOutputStream = new FileOutputStream(tmpLocalTarget);
String remoteFqPath = this.remotePath + "/" + entry.getFilename();
in = sftpSession.getChannel().get(remoteFqPath);
try {
IOUtils.copy(in, fileOutputStream);
}
finally {
IOUtils.closeQuietly(in);
IOUtils.closeQuietly(fileOutputStream);
}
if (tmpLocalTarget.renameTo(localFile) && this.entryAcknowledgmentStrategy != null) {
this.acknowledge(sftpSession, entry);
}
return true;
}
catch (Throwable th) {
throw new MessagingException("Failure occurred while copying from remote to local directory", th);
}
}
else {
return true;
}
}
@Override
@SuppressWarnings("unchecked")
protected void syncRemoteToLocalFileSystem() {
@@ -178,8 +145,47 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
}
}
private boolean copyFromRemoteToLocalDirectory(SftpSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir) throws Exception {
File fileForLocalDir = localDir.getFile();
File localFile = new File(fileForLocalDir, entry.getFilename());
if (!localFile.exists()) {
InputStream in = null;
FileOutputStream fileOutputStream = null;
try {
File tmpLocalTarget = new File(localFile.getAbsolutePath() +
AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION);
fileOutputStream = new FileOutputStream(tmpLocalTarget);
String remoteFqPath = this.remotePath + "/" + entry.getFilename();
in = sftpSession.getChannel().get(remoteFqPath);
try {
IOUtils.copy(in, fileOutputStream);
}
finally {
IOUtils.closeQuietly(in);
IOUtils.closeQuietly(fileOutputStream);
}
if (tmpLocalTarget.renameTo(localFile) && this.entryAcknowledgmentStrategy != null) {
this.acknowledge(sftpSession, entry);
}
return true;
}
catch (Exception e) {
if (e instanceof RuntimeException){
throw (RuntimeException) e;
}
else {
throw new MessagingException("Failure occurred while copying from remote to local directory", e);
}
}
}
else {
return true;
}
}
private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy<ChannelSftp.LsEntry> {
public void acknowledge(Object useful, ChannelSftp.LsEntry msg) throws Exception {
SftpSession sftpSession = (SftpSession) useful;
String remoteFqPath = remotePath + "/" + msg.getFilename();