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

Removing already deprecated instances of *FileList*Filters in favor of their entries.* equivalents which are generics-friendly and can be reused across adapters. Also, added some documentaton to the AbstractInboundRemoteFileSystemSynchroniz*.java classes so that the lifecycle hooks are explained for subsequent file system adapter implementations

Browse Source
This commit is contained in:
Josh Long
2010-08-20 18:08:32 +00:00
parent 1562d53118
commit b5b0e06e81
9 changed files with 53 additions and 547 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -11,7 +11,10 @@ import java.util.concurrent.ScheduledFuture;
/**
* This handles a lot of the common ground in our approach for synchronizing a remote file system locally
* Strategy class charged with knowing how to connect to a remote file system, scan it for new files and then downloading the file.
* <p/>
* The implementation should run through any configured {@link org.springframework.integration.file.entries.EntryListFilter}s
* to ensure the entry is worth downloading.
*
* @author Josh Long
*/
@@ -20,7 +23,8 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<T> extends Abst
* Should we <emphasis>delete</emphasis> the <b>source</b> file?
* For an FTP server, for example, this would delete the original FTPFile instance
* <p/>
* At the moment I can simply see this triggering the setting
* At the moment I can simply see this triggering an implementation specific {@link org.springframework.integration.file.AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy}
* implementation that knows how to delete an entry on the remote file system.
*/
protected boolean shouldDeleteSourceFile;
@@ -48,9 +52,9 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<T> extends Abst
* Obviously thread safe - simply provides a NOOP impl so we don't have to keep dancing around NPE's
*/
private EntryAcknowledgmentStrategy<T> noOpEntryAcknowledgmentStrategy = new EntryAcknowledgmentStrategy<T>() {
public void acknowledge(Object o, T msg) {
}
};
public void acknowledge(Object o, T msg) {
}
};
public void setEntryAcknowledgmentStrategy(EntryAcknowledgmentStrategy<T> entryAcknowledgmentStrategy) {
this.entryAcknowledgmentStrategy = entryAcknowledgmentStrategy;
@@ -69,22 +73,27 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<T> extends Abst
}
/**
* @param usefulContextOrClientData
* @param t leverages strategy implementations to enable different behavior. It's a hook to the entry ({@link T}) after it's been successfully downloaded.
* Conceptually, you might delete the remote one or rename it or something
* @throws Throwable
* @param usefulContextOrClientData this is context information to be passed to the individual {@link org.springframework.integration.file.AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy} implementation.
* {@link org.springframework.integration.file.AbstractInboundRemoteFileSystemSychronizer.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 t leverages strategy implementations to enable different behavior. It's a hook to the entry ({@link T}) after it's been successfully downloaded.
* Conceptually, you might delete the remote one or rename it or something
* @throws Throwable escape hatch exception, let the adapter deal with it.
*/
protected void acknowledge(Object usefulContextOrClientData, T t)
throws Throwable {
throws Throwable {
Assert.notNull(this.entryAcknowledgmentStrategy != null, "entryAcknowledgmentStrategy can't be null!");
this.entryAcknowledgmentStrategy.acknowledge(usefulContextOrClientData, t);
}
/**
* This is the callback where we need the implementation to do some specific work
*
* @throws Exception thrown if anything goes wrong
*/
protected abstract void syncRemoteToLocalFileSystem()
throws Exception ;
throws Exception;
/**
* {@inheritDoc}
@@ -114,9 +123,13 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<T> extends Abst
}
/**
* Strategy interface to expose a hook for dispatching, moving, or deleting the file once it's been delivered
* Strategy interface to expose a hook for dispatching, moving, or deleting the file once it's 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 org.springframework.integration.file.AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy}
* into the pipeline and also some more advanced scenarios (i.e., 'move file to another folder on delete ', or 'rename on delete')
*
* @author Josh Long
* @param <T> the entry type (file, sftp, ftp, ...)
*/
public static interface EntryAcknowledgmentStrategy<T> {
@@ -141,7 +154,7 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<T> extends Abst
try {
syncRemoteToLocalFileSystem();
} catch (Exception e) {
throw new RuntimeException(e) ;
throw new RuntimeException(e);
}
}
}

29
spring-integration-file/src/main/java/org/springframework/integration/file/AbstractInboundRemoteFileSystemSynchronizingMessageSource.java
View File

@@ -1,22 +1,28 @@
package org.springframework.integration.file;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.file.entries.*;
import org.springframework.util.Assert;
import java.io.File;
import java.util.regex.Pattern;
/**
* Ultimately, this factors out a lot of the common logic between the FTP and SFTP adapters
* Ultimately, this factors out a lot of the common logic between the FTP and SFTP adapters. Designed to be extendable to handle
* adapters whose task it is to synchronize a remote file system with a local file system (NB: this does *NOT* handle pushing files TO the remote
* file system that exist uniquely in the local file system. It only handles bringing down the remote file system - as you'd expect
* an 'inbound' adapter would).
* <p/>
* The base class supports configuration of whether the remote file system and local file system's directories should
* be created on start (what 'creating a directory' means to the specific adapter is of course implementaton specific).
* <p/>
* This class is to be used as a pair with an implementation of
* {@link org.springframework.integration.file.AbstractInboundRemoteFileSystemSychronizer<T>}. This synchronizer
* must handle the work of actually connecting to the remote file system and delivering new {@link java.io.File}s.
* The synchronizer is designed to be
*
* @author Josh Long
*/
@@ -32,7 +38,7 @@ public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<
protected volatile boolean autoCreateDirectories = true;
/**
* An implementation that will handle the chores of actually syncing up the remote FS
* An implementation that will handle the chores of actually connecting to and syncing up the remote FS with the local one, in an inbound direction
*/
protected volatile T synchronizer;
@@ -80,13 +86,10 @@ public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<
}
if (this.autoCreateDirectories) {
if((this.localDirectory != null) && !this.localDirectory.exists() && this.localDirectory.getFile().mkdirs())
logger.debug( "the localDirectory " + this.localDirectory + " doesn't exist");
if ((this.localDirectory != null) && !this.localDirectory.exists() && this.localDirectory.getFile().mkdirs())
logger.debug("the localDirectory " + this.localDirectory + " doesn't exist");
}
/**
* Handles making sure the remote files get here in one piece
*/

115
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FtpFileSource.java
View File

@@ -1,115 +0,0 @@
/*
* Copyright 2002-2008 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.ftp;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.io.Resource;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.entries.*;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import java.io.File;
import java.io.IOException;
import java.util.regex.Pattern;
/**
* A source adapter for receiving files via FTP.
*
* @author Iwein Fuld
*/
@Deprecated
public class FtpFileSource implements MessageSource<File>, InitializingBean, Lifecycle {
private FileReadingMessageSource fileSource;
private FtpInboundSynchronizer synchronizer;
private EntryNamer fileEntryName = new FileEntryNamer();
public FtpFileSource() {
this(new FileReadingMessageSource(), new FtpInboundSynchronizer());
}
public FtpFileSource(FileReadingMessageSource fileSource, FtpInboundSynchronizer synchronizer) {
this.fileSource = fileSource;
this.synchronizer = synchronizer;
Pattern completePattern = Pattern.compile("^.*(?<!" + FtpInboundSynchronizer.INCOMPLETE_EXTENSION + ")$");
EntryListFilter<File> f = new CompositeEntryListFilter<File>(new AcceptOnceEntryFileListFilter<File>(),
new PatternMatchingEntryListFilter(fileEntryName, completePattern));
fileSource.setFilter(f);
}
public void setFileSource(FileReadingMessageSource fileSource) {
this.fileSource = fileSource;
}
public void setSynchronizer(FtpInboundSynchronizer synchronizer) {
this.synchronizer = synchronizer;
}
public void setLocalWorkingDirectory(Resource localWorkingDirectory) {
this.synchronizer.setLocalDirectory(localWorkingDirectory);
try {
this.fileSource.setDirectory(localWorkingDirectory.getFile());
} catch (IOException e) {
// oops
}
}
public void setTrigger(Trigger trigger) {
synchronizer.setTrigger(trigger);
}
public void setTaskScheduler(TaskScheduler scheduler) {
synchronizer.setTaskScheduler(scheduler);
}
public void setClientPool(FtpClientPool pool) {
synchronizer.setClientPool(pool);
}
public void afterPropertiesSet() throws Exception {
synchronizer.afterPropertiesSet();
}
public Message<File> receive() {
return fileSource.receive();
}
public void onFailure(Message<File> failedMessage, Throwable t) {
fileSource.onFailure(failedMessage);
}
public void onSend(Message<File> sentMessage) {
fileSource.onSend(sentMessage);
}
public boolean isRunning() {
return synchronizer.isRunning();
}
public void start() {
synchronizer.start();
}
public void stop() {
synchronizer.stop();
}
}

201
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FtpFileSourceFactoryBean.java
View File

@@ -1,201 +0,0 @@
package org.springframework.integration.ftp;
import org.apache.commons.lang.SystemUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceEditor;
import org.springframework.core.io.ResourceLoader;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.entries.CompositeEntryListFilter;
import org.springframework.integration.file.entries.EntryListFilter;
import org.springframework.integration.file.entries.PatternMatchingEntryListFilter;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import org.springframework.util.StringUtils;
import java.io.File;
import java.util.Map;
/**
* Makes it easier to assemble the moving pieces involved in standing up an {@link org.springframework.integration.ftp.FtpFileSource}
*
* @author Josh Long
*/
@Deprecated
public class FtpFileSourceFactoryBean extends AbstractFactoryBean<FtpFileSource> implements ResourceLoaderAware, ApplicationContextAware {
private int port;
private boolean autoCreateDirectories;
private String filenamePattern;
private String username;
private String password;
private String host;
private String remoteDirectory;
private String localWorkingDirectory;
private ApplicationContext applicationContext;
private Resource localDirectoryResource;
private FtpInboundSynchronizer ftpInboundSynchronizer;
private TaskScheduler taskScheduler;
private ResourceLoader resourceLoader;
private FileReadingMessageSource fileReadingMessageSource;
private int clientMode = FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE;
/**
* Used to teach the FTP adapter what files you are interested in receiving
*/
private EntryListFilter<FTPFile> filter;
public void setFilenamePattern(String filenamePattern) {
this.filenamePattern = filenamePattern;
}
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = applicationContext;
}
public void setPort(int port) {
this.port = port;
}
public void setPassword(String password) {
this.password = password;
}
public void setUsername(String username) {
this.username = username;
}
public void setAutoCreateDirectories(boolean autoCreateDirectories) {
this.autoCreateDirectories = autoCreateDirectories;
}
public void setRemoteDirectory(String remoteDirectory) {
this.remoteDirectory = remoteDirectory;
}
public void setHost(String host) {
this.host = host;
}
public void setLocalWorkingDirectory(String localWorkingDirectory) {
this.localWorkingDirectory = localWorkingDirectory;
}
public void setResourceLoader(ResourceLoader resourceLoader) {
this.resourceLoader = resourceLoader;
}
public Class<?extends FtpFileSource> getObjectType() {
return FtpFileSource.class;
}
public void setClientMode(int clientMode) {
this.clientMode = clientMode;
}
public void setFilter(EntryListFilter<FTPFile> filter) {
this.filter = filter;
}
private FtpFileEntryNamer ftpFileEntryNamer =new FtpFileEntryNamer();
@Override
protected FtpFileSource createInstance() throws Exception {
// setup local dir
if (!StringUtils.hasText(this.localWorkingDirectory)) {
File tmp = SystemUtils.getJavaIoTmpDir();
File ftpTmp = new File(tmp, "ftpInbound");
this.localWorkingDirectory = "file://" + ftpTmp.getAbsolutePath();
}
Assert.hasText(this.localWorkingDirectory, "the local working directory can't be null!");
ResourceEditor resourceEditor = new ResourceEditor(this.resourceLoader);
resourceEditor.setAsText(this.localWorkingDirectory);
this.localDirectoryResource = (Resource) resourceEditor.getValue();
fileReadingMessageSource = new FileReadingMessageSource();
this.ftpInboundSynchronizer = new FtpInboundSynchronizer();
CompositeEntryListFilter<FTPFile> compositeFtpFileListFilter = new CompositeEntryListFilter<FTPFile>() ;
if( StringUtils.hasText( this.filenamePattern)){
PatternMatchingEntryListFilter<FTPFile> ftpFilePatternMatchingEntryListFilter = new PatternMatchingEntryListFilter<FTPFile>(ftpFileEntryNamer,filenamePattern) ;
compositeFtpFileListFilter.addFilter( ftpFilePatternMatchingEntryListFilter);
}
if(this.filter != null)
compositeFtpFileListFilter.addFilter( this.filter);
this.ftpInboundSynchronizer.setFilter(compositeFtpFileListFilter);
if (this.taskScheduler == null) {
Map<String, TaskScheduler> tss = null;
if ((tss = applicationContext.getBeansOfType(TaskScheduler.class)).keySet().size() != 0) {
taskScheduler = tss.get(tss.keySet().iterator().next());
}
}
if (null == taskScheduler) {
ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
threadPoolTaskScheduler.setPoolSize(10);
threadPoolTaskScheduler.setErrorHandler(new ErrorHandler() {
public void handleError(Throwable t) {
logger.debug("Error! ", t);
}
});
threadPoolTaskScheduler.setWaitForTasksToCompleteOnShutdown(true);
threadPoolTaskScheduler.initialize();
this.taskScheduler = threadPoolTaskScheduler;
}
DefaultFtpClientFactory defaultFtpClientFactory = new DefaultFtpClientFactory();
defaultFtpClientFactory.setHost(this.host);
defaultFtpClientFactory.setPassword(this.password);
defaultFtpClientFactory.setPort(this.port);
defaultFtpClientFactory.setRemoteWorkingDirectory(this.remoteDirectory);
defaultFtpClientFactory.setUsername(this.username);
defaultFtpClientFactory.setClientMode(this.clientMode);
QueuedFtpClientPool queuedFtpClientPool = new QueuedFtpClientPool(15, defaultFtpClientFactory);
this.ftpInboundSynchronizer.setClientPool(queuedFtpClientPool);
this.ftpInboundSynchronizer.setLocalDirectory(this.localDirectoryResource);
this.ftpInboundSynchronizer.setTaskScheduler(this.taskScheduler);
assert this.localDirectoryResource != null : "the 'localDirectoryResource' can't be null at this point";
if (this.autoCreateDirectories) {
if (!this.localDirectoryResource.exists()) {
try {
if (!localDirectoryResource.getFile().mkdirs()) {
logger.debug("attempted to ensure the existence of the local directory '" + this.localDirectoryResource.getFile().getAbsolutePath() + "' but didn't succeed");
}
} catch (Throwable th) {
logger.debug("attempted to ensure the existence of the local directory '" + this.localDirectoryResource.getFile().getAbsolutePath() + "' but didn't succeed");
}
}
}
this.ftpInboundSynchronizer.afterPropertiesSet();
FtpFileSource ftpFileSource = new FtpFileSource(this.fileReadingMessageSource, this.ftpInboundSynchronizer);
ftpFileSource.setClientPool(queuedFtpClientPool);
ftpFileSource.setLocalWorkingDirectory(this.localDirectoryResource);
ftpFileSource.setSynchronizer(this.ftpInboundSynchronizer);
ftpFileSource.setTaskScheduler(this.taskScheduler);
ftpFileSource.setFileSource(this.fileReadingMessageSource);
ftpFileSource.afterPropertiesSet();
ftpFileSource.start();
return ftpFileSource;
}
}

178
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FtpInboundSynchronizer.java
View File

@@ -1,178 +0,0 @@
/*
* Copyright 2002-2008 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.ftp;
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.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.io.Resource;
import org.springframework.integration.MessagingException;
import org.springframework.integration.file.entries.AcceptAllEntryListFilter;
import org.springframework.integration.file.entries.EntryListFilter;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ScheduledFuture;
/**
* <code>FtpInboundSynchronizer</code> will keep a local directory in sync with a remote Ftp directory.
* It will NOT move new files put into the local directory to the remote server.
*
* @author Iwein Fuld
*/
@Deprecated
public class FtpInboundSynchronizer implements InitializingBean, Lifecycle {
private static final Log logger = LogFactory.getLog(FtpInboundSynchronizer.class);
static final String INCOMPLETE_EXTENSION = ".INCOMPLETE";
private static final long DEFAULT_REFRESH_RATE = 10000;
private volatile TaskScheduler taskScheduler;
private volatile Trigger trigger = new PeriodicTrigger(DEFAULT_REFRESH_RATE);
private volatile FtpClientPool clientPool;
private volatile Resource localDirectory;
private boolean running = false;
private ScheduledFuture<?> scheduledFuture;
private EntryListFilter<FTPFile> filter;
private EntryListFilter<FTPFile> acceptAllFtpFileListFilter = new AcceptAllEntryListFilter<FTPFile>();
public void setFilter(EntryListFilter<FTPFile> filter) {
this.filter = filter;
}
public void setTaskScheduler(TaskScheduler scheduler) {
this.taskScheduler = scheduler;
}
public void setTrigger(Trigger trigger) {
this.trigger = trigger;
}
public void setLocalDirectory(Resource localDirectory) {
this.localDirectory = localDirectory;
}
public void setClientPool(FtpClientPool pool) {
clientPool = pool;
}
public void afterPropertiesSet() throws Exception {
Assert.notNull(localDirectory, "'localDirectory' is required.");
if (this.filter == null) {
this.filter = acceptAllFtpFileListFilter;
}
}
private void synchronize() {
try {
FTPClient client = this.clientPool.getClient();
Assert.state(client != null, FtpClientPool.class.getSimpleName() + " returned 'null' client this most likely a bug in the pool implementation.");
Collection<FTPFile> fileList = this.filter.filterEntries(client.listFiles());
try {
for (FTPFile ftpFile : fileList) {
/*
* according to the FTPFile javadoc the list can contain
* nulls if files couldn't be parsed
*/
if ((ftpFile != null) && ftpFile.isFile()) {
copyFileToLocalDirectory(client, ftpFile, this.localDirectory);
}
}
} finally {
this.clientPool.releaseClient(client);
}
} catch (IOException e) {
throw new MessagingException("Problem occurred while synchronizing remote to local directory", e);
}
}
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 + INCOMPLETE_EXTENSION;
File file = new File(tempFileName);
FileOutputStream fos = new FileOutputStream(file);
try {
client.retrieveFile(remoteFileName, fos);
} finally {
fos.close();
}
file.renameTo(localFile);
return true;
} else {
return false;
}
}
public boolean isRunning() {
return running;
}
public void start() {
if (running) {
return;
}
Assert.state(taskScheduler != null, "'taskScheduler' is required");
scheduledFuture = taskScheduler.schedule(new SynchronizeTask(), trigger);
// future.get();
this.running = true;
if (logger.isInfoEnabled()) {
logger.info("Started " + this);
}
}
public void stop() {
if (!running) {
return;
}
this.scheduledFuture.cancel(true);
this.running = false;
if (logger.isInfoEnabled()) {
logger.info("Stopped " + this);
}
}
private class SynchronizeTask implements Runnable {
public void run() {
synchronize();
}
}
}

12
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/impl/FtpInboundRemoteFileSystemSynchronizer.java
View File

@@ -1,29 +1,21 @@
package org.springframework.integration.ftp.impl;
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;
import org.springframework.integration.MessagingException;
import org.springframework.integration.file.AbstractInboundRemoteFileSystemSychronizer;
import org.springframework.integration.file.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
import org.springframework.integration.ftp.FtpClientPool;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Collection;
import java.util.logging.Logger;
/**
@@ -53,7 +45,7 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot
}
protected boolean copyFileToLocalDirectory(FTPClient client, FTPFile ftpFile, Resource localDirectory)
throws IOException, FileNotFoundException {
throws IOException, FileNotFoundException {
String remoteFileName = ftpFile.getName();
String localFileName = localDirectory.getFile().getPath() + "/" + remoteFileName;
File localFile = new File(localFileName);
@@ -114,7 +106,7 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot
*/
class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy<FTPFile> {
public void acknowledge(Object useful, FTPFile msg)
throws Exception {
throws Exception {
FTPClient ftpClient = (FTPClient) useful;
if ((msg != null) && ftpClient.deleteFile(msg.getName())) {
if (logger.isDebugEnabled()) {

3
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/impl/FtpInboundRemoteFileSystemSynchronizingMessageSource.java
View File

@@ -1,7 +1,6 @@
package org.springframework.integration.ftp.impl;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.integration.file.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
import org.springframework.integration.ftp.FtpClientPool;
@@ -32,7 +31,5 @@ public class FtpInboundRemoteFileSystemSynchronizingMessageSource extends Abstra
protected void onInit() throws Exception {
super.onInit();
this.synchronizer.setClientPool(this.clientPool);
}
}

8
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/impl/FtpRemoteFileSystemSynchronizingMessageSourceFactoryBean.java
View File

@@ -3,23 +3,17 @@ package org.springframework.integration.ftp.impl;
import org.apache.commons.lang.SystemUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceEditor;
import org.springframework.core.io.ResourceLoader;
import org.springframework.integration.file.AbstractInboundRemoteFileSystemSychronizer;
import org.springframework.integration.file.entries.CompositeEntryListFilter;
import org.springframework.integration.file.entries.EntryListFilter;
import org.springframework.integration.file.entries.PatternMatchingEntryListFilter;
import org.springframework.integration.ftp.DefaultFtpClientFactory;
import org.springframework.integration.ftp.FtpFileEntryNamer;
import org.springframework.integration.ftp.QueuedFtpClientPool;
import org.springframework.util.StringUtils;
import java.io.File;
@@ -74,7 +68,7 @@ public class FtpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends Ab
@Override
protected FtpInboundRemoteFileSystemSynchronizingMessageSource createInstance()
throws Exception {
throws Exception {
boolean autoCreatDirs = Boolean.parseBoolean(this.autoCreateDirectories);
boolean ackRemoteDir = Boolean.parseBoolean(this.autoDeleteRemoteFilesOnSync);

13
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizer.java
View File

@@ -27,7 +27,7 @@ import java.util.Collection;
*/
public class SftpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemoteFileSystemSychronizer<ChannelSftp.LsEntry> {
/**
* the path on te remote mount
* the path on the remote mount
*/
private volatile String remotePath;
@@ -42,8 +42,8 @@ public class SftpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemo
@Override
protected void onInit() throws Exception {
Assert.notNull(this.clientPool, "clientPool can't be null");
Assert.notNull(this.clientPool, "'clientPool' can't be null");
Assert.notNull(this.remotePath, "'remotePath' can't be null");
if (this.shouldDeleteSourceFile) {
this.entryAcknowledgmentStrategy = new DeletionEntryAcknowledgmentStrategy();
}
@@ -56,7 +56,7 @@ public class SftpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemo
@SuppressWarnings("ignored")
private boolean copyFromRemoteToLocalDirectory(SftpSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir)
throws Exception {
throws Exception {
File fileForLocalDir = localDir.getFile();
File localFile = new File(fileForLocalDir, entry.getFilename());
@@ -66,7 +66,8 @@ public class SftpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemo
FileOutputStream fileOutputStream = null;
try {
File tmpLocalTarget = new File(localFile.getAbsolutePath() + AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION);
File tmpLocalTarget = new File(localFile.getAbsolutePath() +
AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION);
fileOutputStream = new FileOutputStream(tmpLocalTarget);
@@ -126,7 +127,7 @@ public class SftpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemo
class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy<ChannelSftp.LsEntry> {
public void acknowledge(Object useful, ChannelSftp.LsEntry msg)
throws Exception {
throws Exception {
SftpSession sftpSession = (SftpSession) useful;
String remoteFqPath = remotePath + "/" + msg.getFilename();