Fixed teh SFTP inbound adapter - it's now based on the abstract base classes. finally removing the sftp-specific impl's of the EntryListFilter* hierarchy. no need since we have a generic EntryListFilter<T> hierarchy instead
This commit is contained in:
Josh Long
@@ -1,45 +0,0 @@
|
||||
/*
|
||||
* Copyright 2010 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.sftp;
|
||||
|
||||
import com.jcraft.jsch.ChannelSftp;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
|
||||
/**
|
||||
* Convenience implementation patterned off {@link org.springframework.integration.file.filters.FileListFilter}
|
||||
*
|
||||
* @author Josh Long
|
||||
*/ @Deprecated
|
||||
public abstract class AbstractSftpFileListFilter implements SftpFileListFilter {
|
||||
|
||||
abstract public boolean accept(ChannelSftp.LsEntry lsEntry);
|
||||
|
||||
public List<ChannelSftp.LsEntry> filterFiles(ChannelSftp.LsEntry[] files) {
|
||||
List<ChannelSftp.LsEntry> accepted = new ArrayList<ChannelSftp.LsEntry>();
|
||||
|
||||
if (files != null) {
|
||||
for (ChannelSftp.LsEntry lsEntry : files)
|
||||
if (this.accept(lsEntry)) {
|
||||
accepted.add(lsEntry);
|
||||
}
|
||||
}
|
||||
|
||||
return accepted;
|
||||
}
|
||||
}
|
||||
@@ -1,55 +0,0 @@
|
||||
/*
|
||||
* Copyright 2010 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.sftp;
|
||||
|
||||
import com.jcraft.jsch.ChannelSftp;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
|
||||
/**
|
||||
* Patterned very much on th
|
||||
*
|
||||
* @author Josh Long
|
||||
*/
|
||||
@Deprecated
|
||||
public class CompositeFtpFileListFilter implements SftpFileListFilter {
|
||||
private Set<SftpFileListFilter> filters;
|
||||
|
||||
public CompositeFtpFileListFilter(SftpFileListFilter... ftpFileListFilter) {
|
||||
this.filters = new LinkedHashSet<SftpFileListFilter>(Arrays.asList(ftpFileListFilter));
|
||||
}
|
||||
|
||||
public CompositeFtpFileListFilter(Collection<SftpFileListFilter> ftpFileListFilter) {
|
||||
this.filters = new LinkedHashSet<SftpFileListFilter>(ftpFileListFilter);
|
||||
}
|
||||
|
||||
public void addFilter(SftpFileListFilter ftpFileListFilter) {
|
||||
this.filters.add(ftpFileListFilter);
|
||||
}
|
||||
|
||||
public List<ChannelSftp.LsEntry> filterFiles(ChannelSftp.LsEntry[] files) {
|
||||
Assert.notNull(files, "files[] can't be null!");
|
||||
|
||||
List<ChannelSftp.LsEntry> leftOver = Arrays.asList(files);
|
||||
|
||||
for (SftpFileListFilter ff : this.filters)
|
||||
leftOver = ff.filterFiles(leftOver.toArray(new ChannelSftp.LsEntry[leftOver.size()]));
|
||||
|
||||
return leftOver;
|
||||
}
|
||||
}
|
||||
@@ -1,66 +0,0 @@
|
||||
/*
|
||||
* Copyright 2010 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.sftp;
|
||||
|
||||
import com.jcraft.jsch.ChannelSftp;
|
||||
import org.apache.commons.lang.builder.ToStringBuilder;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
|
||||
/**
|
||||
* Validates {@link com.jcraft.jsch.ChannelSftp.LsEntry}s against a {@link java.util.regex.Pattern}.
|
||||
*
|
||||
* @author Josh Long
|
||||
*/
|
||||
@Deprecated
|
||||
public class PatternMatchingSftpFileListFilter extends AbstractSftpFileListFilter implements InitializingBean {
|
||||
|
||||
private Log logger = LogFactory.getLog(getClass());
|
||||
|
||||
private Pattern pattern;
|
||||
private String patternExpression;
|
||||
|
||||
public void setPattern(Pattern pattern) {
|
||||
this.pattern = pattern;
|
||||
}
|
||||
|
||||
public void setPatternExpression(String patternExpression) {
|
||||
this.patternExpression = patternExpression;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean accept(ChannelSftp.LsEntry lsEntry) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("testing: " + ToStringBuilder.reflectionToString(lsEntry));
|
||||
}
|
||||
|
||||
return (lsEntry != null) && this.pattern.matcher(lsEntry.getFilename()).matches();
|
||||
}
|
||||
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
if (StringUtils.hasText(this.patternExpression) && (this.pattern == null)) {
|
||||
this.pattern = Pattern.compile(this.patternExpression);
|
||||
}
|
||||
|
||||
Assert.notNull(this.pattern, "the pattern must not be null");
|
||||
}
|
||||
}
|
||||
@@ -1,30 +0,0 @@
|
||||
/*
|
||||
* Copyright 2010 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.sftp;
|
||||
|
||||
import com.jcraft.jsch.ChannelSftp;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Filters out all the {@link com.jcraft.jsch.ChannelSftp.LsEntry} taken in a scan of the remote mount
|
||||
* and returns the balance. These are then sync'd to the local directory.
|
||||
*
|
||||
* @author Josh Long
|
||||
*/ @Deprecated
|
||||
public interface SftpFileListFilter {
|
||||
List<ChannelSftp.LsEntry> filterFiles (ChannelSftp.LsEntry [] files);
|
||||
}
|
||||
@@ -1,280 +0,0 @@
|
||||
/*
|
||||
* Copyright 2010 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.sftp;
|
||||
|
||||
import com.jcraft.jsch.ChannelSftp;
|
||||
import com.jcraft.jsch.SftpATTRS;
|
||||
import org.apache.commons.io.IOUtils;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.integration.MessagingException;
|
||||
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.FileOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
|
||||
|
||||
/**
|
||||
* This handles keeping the {@link #localDirectory} in sync with the contents of the remote mount. From there, files are deposited
|
||||
* into a folder where the {@link org.springframework.integration.file.FileReadingMessageSource} will eventually deliver them as events
|
||||
*
|
||||
* @author Josh Long
|
||||
* @author Mario Gray
|
||||
*/ @Deprecated
|
||||
public class SftpInboundSynchronizer implements InitializingBean {
|
||||
private static final long DEFAULT_REFRESH_RATE = 10 * 1000; // 10 seconds
|
||||
|
||||
// a lot of the approach for this (including the use of a FileReadingMessageSource and the regex / mask approach were lifted from FtpInboundSynchronizer
|
||||
static final String INCOMPLETE_EXTENSION = ".INCOMPLETE";
|
||||
private Log logger = LogFactory.getLog(getClass());
|
||||
private volatile Resource localDirectory;
|
||||
private volatile SftpSessionPool pool;
|
||||
private volatile ScheduledFuture<?> scheduledFuture;
|
||||
private volatile String remotePath;
|
||||
private volatile TaskScheduler taskScheduler;
|
||||
private volatile Trigger trigger = new PeriodicTrigger(DEFAULT_REFRESH_RATE);
|
||||
private volatile boolean autoCreatePath;
|
||||
private volatile boolean running;
|
||||
private SftpFileListFilter filter;
|
||||
|
||||
public void setFilter(SftpFileListFilter filter) {
|
||||
this.filter = filter;
|
||||
}
|
||||
|
||||
private volatile boolean shouldDeleteDownloadedRemoteFiles; //.. this is false
|
||||
|
||||
private SftpFileListFilter acceptAllFilteListFilter = new SftpFileListFilter(){
|
||||
public List<ChannelSftp.LsEntry> filterFiles(ChannelSftp.LsEntry[] files) {
|
||||
return Arrays.asList( files);
|
||||
}
|
||||
} ;
|
||||
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
Assert.state(taskScheduler != null, "taskScheduler can't be null!");
|
||||
Assert.state(localDirectory != null, "the localDirectory property must not be null!");
|
||||
|
||||
File localDir = localDirectory.getFile();
|
||||
|
||||
if (!localDir.exists()) {
|
||||
if (autoCreatePath) {
|
||||
if (!localDir.mkdirs()) {
|
||||
throw new RuntimeException(String.format("couldn't create localDirectory %s", this.localDirectory.getFile().getAbsolutePath()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if(this.filter == null)
|
||||
this.filter = acceptAllFilteListFilter;
|
||||
|
||||
}
|
||||
|
||||
public boolean isRunning() {
|
||||
return running;
|
||||
}
|
||||
|
||||
public boolean isShouldDeleteDownloadedRemoteFiles() {
|
||||
return shouldDeleteDownloadedRemoteFiles;
|
||||
}
|
||||
|
||||
public void setAutoCreatePath(boolean autoCreatePath) {
|
||||
this.autoCreatePath = autoCreatePath;
|
||||
}
|
||||
|
||||
public void setLocalDirectory(Resource localDirectory) {
|
||||
this.localDirectory = localDirectory;
|
||||
}
|
||||
|
||||
public void setPool(SftpSessionPool pool) {
|
||||
this.pool = pool;
|
||||
}
|
||||
|
||||
public void setRemotePath(String remotePath) {
|
||||
this.remotePath = remotePath;
|
||||
}
|
||||
|
||||
public void setScheduledFuture(ScheduledFuture<?> scheduledFuture) {
|
||||
this.scheduledFuture = scheduledFuture;
|
||||
}
|
||||
|
||||
public void setShouldDeleteDownloadedRemoteFiles(boolean shouldDeleteDownloadedRemoteFiles) {
|
||||
this.shouldDeleteDownloadedRemoteFiles = shouldDeleteDownloadedRemoteFiles;
|
||||
}
|
||||
|
||||
public void setTaskScheduler(TaskScheduler taskScheduler) {
|
||||
this.taskScheduler = taskScheduler;
|
||||
}
|
||||
|
||||
public void setTrigger(Trigger t) {
|
||||
this.trigger = t;
|
||||
}
|
||||
|
||||
public void start() {
|
||||
if (running) {
|
||||
return;
|
||||
}
|
||||
|
||||
Assert.state(checkThatRemotePathExists(remotePath), "the remotePath should exist before we can sync with it!");
|
||||
Assert.state(taskScheduler != null, "'taskScheduler' is required");
|
||||
|
||||
scheduledFuture = taskScheduler.schedule(new SynchronizeTask(), trigger);
|
||||
|
||||
this.running = true;
|
||||
}
|
||||
|
||||
public void stop() {
|
||||
if (!running) {
|
||||
return;
|
||||
}
|
||||
|
||||
Assert.state(scheduledFuture != null, "scheduledFuture is null!");
|
||||
this.scheduledFuture.cancel(true);
|
||||
this.running = false;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public void synchronize() throws Exception {
|
||||
SftpSession session = null;
|
||||
|
||||
try {
|
||||
session = pool.getSession();
|
||||
session.start();
|
||||
|
||||
ChannelSftp channelSftp = session.getChannel();
|
||||
Collection<ChannelSftp.LsEntry> beforeFilter = channelSftp.ls(remotePath);
|
||||
ChannelSftp.LsEntry [] entries = beforeFilter == null? new ChannelSftp.LsEntry[0] :
|
||||
beforeFilter.toArray(new ChannelSftp.LsEntry[ beforeFilter.size()]) ;
|
||||
Collection<ChannelSftp.LsEntry> files = this.filter.filterFiles( entries );
|
||||
|
||||
for (ChannelSftp.LsEntry lsEntry : files) {
|
||||
if ((lsEntry != null) && !lsEntry.getAttrs().isDir() && !lsEntry.getAttrs().isLink()) {
|
||||
copyFromRemoteToLocalDirectory(session, lsEntry, this.localDirectory);
|
||||
}
|
||||
}
|
||||
} catch (IOException e) {
|
||||
throw new MessagingException("couldn't synchronize remote to local directory", e);
|
||||
} finally {
|
||||
if ((session != null) && (pool != null)) {
|
||||
pool.release(session);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* there be dragons this way ... This method will check to ensure that the remote directory exists. If the directory
|
||||
* doesnt exist, and autoCreatePath is 'true,' then this method makes a few reasonably sane attempts
|
||||
* to create it. Otherwise, it fails fast.
|
||||
*
|
||||
* @param remotePath the path on the remote SSH / SFTP server to create.
|
||||
* @return whether or not the directory is there (regardless of whether we created it in this method or it already
|
||||
* existed.)
|
||||
*/
|
||||
private boolean checkThatRemotePathExists(String remotePath) {
|
||||
SftpSession session = null;
|
||||
ChannelSftp channelSftp = null;
|
||||
|
||||
try {
|
||||
session = pool.getSession();
|
||||
Assert.state(session != null, "session as returned from the pool should not be null. " + "If it is, it is most likely an error in the pool implementation. ");
|
||||
session.start();
|
||||
channelSftp = session.getChannel();
|
||||
|
||||
SftpATTRS attrs = channelSftp.stat(remotePath);
|
||||
assert (attrs != null) && attrs.isDir() : "attrs can't be null, and should indicate that it's a directory!";
|
||||
|
||||
return true;
|
||||
} catch (Throwable th) {
|
||||
if (this.autoCreatePath && (pool != null) && (session != null)) {
|
||||
try {
|
||||
if (channelSftp != null) {
|
||||
channelSftp.mkdir(remotePath);
|
||||
|
||||
if (channelSftp.stat(remotePath).isDir()) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if ((pool != null) && (session != null)) {
|
||||
pool.release(session);
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
@SuppressWarnings("ignored")
|
||||
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 fos = null;
|
||||
|
||||
try {
|
||||
File tmpLocalTarget = new File(localFile.getAbsolutePath() + INCOMPLETE_EXTENSION);
|
||||
|
||||
fos = new FileOutputStream(tmpLocalTarget);
|
||||
|
||||
String remoteFqPath = this.remotePath + "/" + entry.getFilename();
|
||||
in = sftpSession.getChannel().get(remoteFqPath);
|
||||
IOUtils.copy(in, fos);
|
||||
|
||||
if (tmpLocalTarget.renameTo(localFile)) {
|
||||
// last step
|
||||
if (isShouldDeleteDownloadedRemoteFiles()) {
|
||||
sftpSession.getChannel().rm(remoteFqPath);
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
} catch (Throwable th) {
|
||||
IOUtils.closeQuietly(in);
|
||||
IOUtils.closeQuietly(fos);
|
||||
}
|
||||
} else {
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
class SynchronizeTask implements Runnable {
|
||||
public void run() {
|
||||
try {
|
||||
synchronize();
|
||||
} catch (Throwable e) {
|
||||
// todo logger.debug("couldn't invoke synchronize()", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,290 +0,0 @@
|
||||
/*
|
||||
* Copyright 2010 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.sftp.config;
|
||||
|
||||
import org.apache.commons.lang.SystemUtils;
|
||||
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.sftp.*;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import org.springframework.scheduling.Trigger;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
import org.springframework.util.ErrorHandler;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.io.File;
|
||||
import java.util.Map;
|
||||
|
||||
|
||||
/**
|
||||
* Building a {@link org.springframework.integration.sftp.SftpMessageSource} is a complicated because we also
|
||||
* use a {@link org.springframework.integration.file.FileReadingMessageSource} to handle the "receipt" of files in
|
||||
* a {@link #localWorkingDirectory}.
|
||||
*
|
||||
* @author Josh Long
|
||||
*/
|
||||
public class SftpMessageSourceFactoryBean extends AbstractFactoryBean<SftpMessageSource> implements ApplicationContextAware, ResourceLoaderAware {
|
||||
private ApplicationContext applicationContext;
|
||||
private FileReadingMessageSource fileReadingMessageSource;
|
||||
private Resource localDirectoryResource;
|
||||
private ResourceLoader resourceLoader;
|
||||
private SftpInboundSynchronizer synchronizer;
|
||||
private String host;
|
||||
private String keyFile;
|
||||
private String keyFilePassword;
|
||||
private String localWorkingDirectory;
|
||||
private String password;
|
||||
private String remoteDirectory;
|
||||
private String username;
|
||||
private TaskScheduler taskScheduler;
|
||||
private Trigger trigger;
|
||||
private boolean autoCreateDirectories;
|
||||
private boolean autoDeleteRemoteFilesOnSync;
|
||||
private int port = 22;
|
||||
private SftpFileListFilter filter;
|
||||
private String filenamePattern;
|
||||
|
||||
public FileReadingMessageSource getFileReadingMessageSource() {
|
||||
return fileReadingMessageSource;
|
||||
}
|
||||
|
||||
public String getHost() {
|
||||
return host;
|
||||
}
|
||||
|
||||
public String getKeyFile() {
|
||||
return keyFile;
|
||||
}
|
||||
|
||||
public String getKeyFilePassword() {
|
||||
return keyFilePassword;
|
||||
}
|
||||
|
||||
public String getLocalWorkingDirectory() {
|
||||
return localWorkingDirectory;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Class<?extends SftpMessageSource> getObjectType() {
|
||||
return SftpMessageSource.class;
|
||||
}
|
||||
|
||||
public String getPassword() {
|
||||
return password;
|
||||
}
|
||||
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
|
||||
public String getRemoteDirectory() {
|
||||
return remoteDirectory;
|
||||
}
|
||||
|
||||
public SftpInboundSynchronizer getSynchronizer() {
|
||||
return synchronizer;
|
||||
}
|
||||
|
||||
public TaskScheduler getTaskScheduler() {
|
||||
return taskScheduler;
|
||||
}
|
||||
|
||||
public Trigger getTrigger() {
|
||||
return trigger;
|
||||
}
|
||||
|
||||
// this is the ultimate layer of control
|
||||
// users will configure theeir entire experience using this class and trust that a working
|
||||
// component comes out as a result of their input
|
||||
// we need to support user/pw/keys/host/port/auto-delete properties
|
||||
public String getUsername() {
|
||||
return username;
|
||||
}
|
||||
|
||||
public boolean isAutoCreateDirectories() {
|
||||
return autoCreateDirectories;
|
||||
}
|
||||
|
||||
public boolean isAutoDeleteRemoteFilesOnSync() {
|
||||
return autoDeleteRemoteFilesOnSync;
|
||||
}
|
||||
|
||||
public void setApplicationContext(final ApplicationContext applicationContext)
|
||||
throws BeansException {
|
||||
this.applicationContext = applicationContext;
|
||||
}
|
||||
|
||||
public void setAutoCreateDirectories(final boolean autoCreateDirectories) {
|
||||
this.autoCreateDirectories = autoCreateDirectories;
|
||||
}
|
||||
|
||||
public void setAutoDeleteRemoteFilesOnSync(final boolean autoDeleteRemoteFilesOnSync) {
|
||||
this.autoDeleteRemoteFilesOnSync = autoDeleteRemoteFilesOnSync;
|
||||
}
|
||||
|
||||
public void setFileReadingMessageSource(final FileReadingMessageSource fileReadingMessageSource) {
|
||||
this.fileReadingMessageSource = fileReadingMessageSource;
|
||||
}
|
||||
|
||||
public void setHost(final String host) {
|
||||
this.host = host;
|
||||
}
|
||||
|
||||
public void setKeyFile(final String keyFile) {
|
||||
this.keyFile = keyFile;
|
||||
}
|
||||
|
||||
public void setKeyFilePassword(final String keyFilePassword) {
|
||||
this.keyFilePassword = keyFilePassword;
|
||||
}
|
||||
|
||||
public void setLocalWorkingDirectory(final String lwd) {
|
||||
this.localWorkingDirectory = lwd;
|
||||
}
|
||||
|
||||
public void setPassword(final String password) {
|
||||
this.password = password;
|
||||
}
|
||||
|
||||
public void setPort(final int port) {
|
||||
this.port = port;
|
||||
}
|
||||
|
||||
public void setRemoteDirectory(final String remoteDirectory) {
|
||||
this.remoteDirectory = remoteDirectory;
|
||||
}
|
||||
|
||||
public void setResourceLoader(final ResourceLoader resourceLoader) {
|
||||
this.resourceLoader = resourceLoader;
|
||||
}
|
||||
|
||||
public void setSynchronizer(final SftpInboundSynchronizer synchronizer) {
|
||||
this.synchronizer = synchronizer;
|
||||
}
|
||||
|
||||
public void setTaskScheduler(final TaskScheduler taskScheduler) {
|
||||
this.taskScheduler = taskScheduler;
|
||||
}
|
||||
|
||||
public void setTrigger(final Trigger trigger) {
|
||||
this.trigger = trigger;
|
||||
}
|
||||
|
||||
public void setFilenamePattern(String filenamePattern) {
|
||||
this.filenamePattern = filenamePattern;
|
||||
}
|
||||
|
||||
public void setFilter(SftpFileListFilter filter) {
|
||||
this.filter = filter;
|
||||
}
|
||||
|
||||
public void setUsername(final String username) {
|
||||
this.username = username;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected SftpMessageSource createInstance() throws Exception {
|
||||
try {
|
||||
if ((localWorkingDirectory == null) || !StringUtils.hasText(localWorkingDirectory)) {
|
||||
File tmp = SystemUtils.getJavaIoTmpDir();
|
||||
File sftpTmp = new File(tmp, "sftpInbound");
|
||||
this.localWorkingDirectory = "file://" + sftpTmp.getAbsolutePath();
|
||||
}
|
||||
|
||||
// resource for local directory
|
||||
ResourceEditor editor = new ResourceEditor(this.resourceLoader);
|
||||
editor.setAsText(this.localWorkingDirectory);
|
||||
this.localDirectoryResource = (Resource) editor.getValue();
|
||||
|
||||
fileReadingMessageSource = new FileReadingMessageSource();
|
||||
|
||||
synchronizer = new SftpInboundSynchronizer();
|
||||
|
||||
CompositeFtpFileListFilter compositeFtpFileListFilter = new CompositeFtpFileListFilter();
|
||||
|
||||
if (StringUtils.hasText(this.filenamePattern)) {
|
||||
PatternMatchingSftpFileListFilter flp = new PatternMatchingSftpFileListFilter();
|
||||
flp.setPatternExpression(this.filenamePattern);
|
||||
flp.afterPropertiesSet();
|
||||
compositeFtpFileListFilter.addFilter(flp);
|
||||
}
|
||||
|
||||
if (this.filter != null) {
|
||||
compositeFtpFileListFilter.addFilter(this.filter);
|
||||
}
|
||||
|
||||
synchronizer.setFilter(compositeFtpFileListFilter);
|
||||
|
||||
if (null == taskScheduler) {
|
||||
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 ts = new ThreadPoolTaskScheduler();
|
||||
ts.setPoolSize(10);
|
||||
ts.setErrorHandler(new ErrorHandler() {
|
||||
public void handleError(Throwable t) {
|
||||
// todo make this forward a message onto the error channel (how does that work?)
|
||||
logger.debug("error! ", t);
|
||||
}
|
||||
});
|
||||
|
||||
ts.setWaitForTasksToCompleteOnShutdown(true);
|
||||
ts.initialize();
|
||||
this.taskScheduler = ts;
|
||||
}
|
||||
|
||||
SftpSessionFactory sessionFactory = SftpSessionUtils.buildSftpSessionFactory(
|
||||
this.getHost(), this.getPassword(), this.getUsername(), this.getKeyFile(), this.getKeyFilePassword(), this.getPort());
|
||||
|
||||
QueuedSftpSessionPool pool = new QueuedSftpSessionPool(15, sessionFactory);
|
||||
pool.afterPropertiesSet();
|
||||
synchronizer.setRemotePath(this.getRemoteDirectory());
|
||||
synchronizer.setPool(pool);
|
||||
synchronizer.setAutoCreatePath(this.isAutoCreateDirectories());
|
||||
synchronizer.setShouldDeleteDownloadedRemoteFiles(this.isAutoDeleteRemoteFilesOnSync());
|
||||
|
||||
SftpMessageSource sftpMessageSource = new SftpMessageSource(fileReadingMessageSource, synchronizer);
|
||||
|
||||
sftpMessageSource.setTaskScheduler(taskScheduler);
|
||||
|
||||
if (null != this.trigger) {
|
||||
sftpMessageSource.setTrigger(trigger);
|
||||
}
|
||||
|
||||
sftpMessageSource.setLocalDirectory(this.localDirectoryResource);
|
||||
sftpMessageSource.afterPropertiesSet();
|
||||
sftpMessageSource.start();
|
||||
|
||||
return sftpMessageSource;
|
||||
} catch (Throwable thr) {
|
||||
logger.debug("error occurred when trying to configure SFTPmessageSource ", thr);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -19,12 +19,9 @@ import org.springframework.integration.sftp.SftpSessionFactory;
|
||||
|
||||
|
||||
/**
|
||||
*
|
||||
* Provides a single place to handle this tedious chore.
|
||||
*
|
||||
* todo : replace all the ad-hoc definitions of {@link org.springframework.integration.sftp.SftpSessionFactory}
|
||||
*
|
||||
* @author Josh Long
|
||||
* @author Josh Long
|
||||
*/
|
||||
public class SftpSessionUtils {
|
||||
/**
|
||||
@@ -43,7 +40,7 @@ public class SftpSessionUtils {
|
||||
* @throws Exception thrown in case of darned near <em>anything</em>
|
||||
*/
|
||||
public static SftpSessionFactory buildSftpSessionFactory(String host, String pw, String usr, String pvKey, String pvKeyPass, int port)
|
||||
throws Exception {
|
||||
throws Exception {
|
||||
SftpSessionFactory sftpSessionFactory = new SftpSessionFactory();
|
||||
sftpSessionFactory.setPassword(pw);
|
||||
sftpSessionFactory.setPort(port);
|
||||
|
||||
@@ -24,8 +24,7 @@ import java.io.File;
|
||||
*
|
||||
* @author Josh Long
|
||||
*/
|
||||
public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends
|
||||
AbstractFactoryBean<SftpInboundRemoteFileSystemSynchronizingMessageSource> implements ResourceLoaderAware {
|
||||
public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends AbstractFactoryBean<SftpInboundRemoteFileSystemSynchronizingMessageSource> implements ResourceLoaderAware {
|
||||
/**
|
||||
* injected by the container
|
||||
*/
|
||||
@@ -37,71 +36,78 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends
|
||||
private volatile String filenamePattern;
|
||||
private volatile EntryListFilter<ChannelSftp.LsEntry> filter;
|
||||
private int port = 22;
|
||||
private String host;
|
||||
private String keyFile;
|
||||
private String keyFilePassword;
|
||||
private String password;
|
||||
private String remoteDirectory;
|
||||
private String username;
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setLocalDirectoryResource(Resource localDirectoryResource) {
|
||||
this.localDirectoryResource = localDirectoryResource;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setLocalDirectoryPath(String localDirectoryPath) {
|
||||
this.localDirectoryPath = localDirectoryPath;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setAutoCreateDirectories(String autoCreateDirectories) {
|
||||
this.autoCreateDirectories = autoCreateDirectories;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setAutoDeleteRemoteFilesOnSync(String autoDeleteRemoteFilesOnSync) {
|
||||
this.autoDeleteRemoteFilesOnSync = autoDeleteRemoteFilesOnSync;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setFilenamePattern(String filenamePattern) {
|
||||
this.filenamePattern = filenamePattern;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setFilter(EntryListFilter<ChannelSftp.LsEntry> filter) {
|
||||
this.filter = filter;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setPort(int port) {
|
||||
this.port = port;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setHost(String host) {
|
||||
this.host = host;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setKeyFile(String keyFile) {
|
||||
this.keyFile = keyFile;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setKeyFilePassword(String keyFilePassword) {
|
||||
this.keyFilePassword = keyFilePassword;
|
||||
}
|
||||
|
||||
public void setLocalWorkingDirectory(String localWorkingDirectory) {
|
||||
this.localWorkingDirectory = localWorkingDirectory;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setPassword(String password) {
|
||||
this.password = password;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setRemoteDirectory(String remoteDirectory) {
|
||||
this.remoteDirectory = remoteDirectory;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public void setUsername(String username) {
|
||||
this.username = username;
|
||||
}
|
||||
|
||||
private String host;
|
||||
private String keyFile;
|
||||
private String keyFilePassword;
|
||||
private String localWorkingDirectory;
|
||||
private String password;
|
||||
private String remoteDirectory;
|
||||
private String username;
|
||||
|
||||
public void setResourceLoader(ResourceLoader resourceLoader) {
|
||||
this.resourceLoader = resourceLoader;
|
||||
}
|
||||
@@ -145,8 +151,7 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends
|
||||
this.filter = compositeFtpFileListFilter;
|
||||
|
||||
// pools
|
||||
SftpSessionFactory sessionFactory = SftpSessionUtils.buildSftpSessionFactory(
|
||||
this.host, this.password, this.username, this.keyFile, this.keyFilePassword, this.port);
|
||||
SftpSessionFactory sessionFactory = SftpSessionUtils.buildSftpSessionFactory(this.host, this.password, this.username, this.keyFile, this.keyFilePassword, this.port);
|
||||
|
||||
QueuedSftpSessionPool pool = new QueuedSftpSessionPool(15, sessionFactory);
|
||||
pool.afterPropertiesSet();
|
||||
@@ -158,8 +163,8 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends
|
||||
sftpSync.setFilter(compositeFtpFileListFilter);
|
||||
sftpSync.setBeanFactory(this.getBeanFactory());
|
||||
sftpSync.setRemotePath(this.remoteDirectory);
|
||||
sftpSync.afterPropertiesSet();// todo is this correct ?
|
||||
sftpSync.start();//todo
|
||||
sftpSync.afterPropertiesSet(); // todo is this correct ?
|
||||
sftpSync.start(); //todo
|
||||
|
||||
sftpMsgSrc.setRemotePredicate(compositeFtpFileListFilter);
|
||||
sftpMsgSrc.setSynchronizer(sftpSync);
|
||||
@@ -177,7 +182,7 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends
|
||||
private Resource fromText(String path) {
|
||||
ResourceEditor resourceEditor = new ResourceEditor(this.resourceLoader);
|
||||
resourceEditor.setAsText(path);
|
||||
|
||||
return (Resource) resourceEditor.getValue();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,108 +0,0 @@
|
||||
package org.springframework.integration.sftp;
|
||||
|
||||
import org.apache.commons.lang.SystemUtils;
|
||||
import org.apache.commons.lang.exception.ExceptionUtils;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.core.io.FileSystemResource;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
import org.springframework.util.ErrorHandler;
|
||||
|
||||
import java.io.File;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
|
||||
/**
|
||||
* This tests the API, more than the end-to-end XML to Java approach.
|
||||
*
|
||||
* @author Josh Long
|
||||
*/
|
||||
public class TestSftpReceipt {
|
||||
/*
|
||||
private static final Logger logger = Logger.getLogger(TestSftpReceipt.class.getName());
|
||||
private SftpSessionFactory sftpSessionFactory;
|
||||
private String host;
|
||||
private String password;
|
||||
private String user;
|
||||
private String privateKeyPath;
|
||||
private String privateKeyPassword;
|
||||
private int port;
|
||||
|
||||
@Before
|
||||
public void before() throws Throwable {
|
||||
this.sftpSessionFactory = buildSFTPSessionFactory(this.host, this.password, this.user, this.privateKeyPath, this.privateKeyPassword, this.port);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testReceive() throws Throwable {
|
||||
String localMount = SystemUtils.getUserHome() + "/local_mount";
|
||||
String remoteMount = "remote_mount";
|
||||
|
||||
// local path
|
||||
File local = new File(localMount); // obviously this is just for test. Do what you need to do in your own
|
||||
|
||||
// we are testing, after all
|
||||
if (local.exists() && (local.list().length > 0)) {
|
||||
for (File f : local.listFiles()) {
|
||||
if (!f.delete()) {
|
||||
logger.fine("couldn't delete " + f.getAbsolutePath());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Resource localDirectory = new FileSystemResource(local);
|
||||
|
||||
// pool
|
||||
QueuedSftpSessionPool queuedSFTPSessionPool = new QueuedSftpSessionPool(sftpSessionFactory);
|
||||
queuedSFTPSessionPool.afterPropertiesSet();
|
||||
|
||||
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
|
||||
taskScheduler.setPoolSize(10);
|
||||
taskScheduler.setErrorHandler(new ErrorHandler() {
|
||||
public void handleError(Throwable t) {
|
||||
System.out.println("Error occurred: " + ExceptionUtils.getFullStackTrace(t));
|
||||
}
|
||||
});
|
||||
|
||||
taskScheduler.setWaitForTasksToCompleteOnShutdown(true);
|
||||
taskScheduler.initialize();
|
||||
|
||||
// synchronizer
|
||||
final SftpInboundSynchronizer sftpInboundSynchronizer = new SftpInboundSynchronizer();
|
||||
sftpInboundSynchronizer.setLocalDirectory(localDirectory);
|
||||
sftpInboundSynchronizer.setRemotePath(remoteMount);
|
||||
sftpInboundSynchronizer.setAutoCreatePath(true);
|
||||
sftpInboundSynchronizer.setPool(queuedSFTPSessionPool);
|
||||
sftpInboundSynchronizer.setShouldDeleteDownloadedRemoteFiles(false);
|
||||
sftpInboundSynchronizer.setTaskScheduler(taskScheduler);
|
||||
sftpInboundSynchronizer.afterPropertiesSet();
|
||||
sftpInboundSynchronizer.start();
|
||||
|
||||
new Thread(new Runnable() {
|
||||
public void run() {
|
||||
try {
|
||||
Thread.sleep(60 * 1000); // 1 minute
|
||||
|
||||
sftpInboundSynchronizer.stop();
|
||||
} catch (InterruptedException e) {
|
||||
// don't care
|
||||
}
|
||||
}
|
||||
}).start();
|
||||
}
|
||||
|
||||
private SftpSessionFactory buildSFTPSessionFactory(String host, String pw, String usr, String pvKey, String pvKeyPass, int port)
|
||||
throws Throwable {
|
||||
SftpSessionFactory sftpSessionFactory = new SftpSessionFactory();
|
||||
sftpSessionFactory.setPassword(pw);
|
||||
sftpSessionFactory.setPort(port);
|
||||
sftpSessionFactory.setRemoteHost(host);
|
||||
sftpSessionFactory.setUser(usr);
|
||||
sftpSessionFactory.setPrivateKey(pvKey);
|
||||
sftpSessionFactory.setPrivateKeyPassphrase(pvKeyPass);
|
||||
sftpSessionFactory.afterPropertiesSet();
|
||||
|
||||
return sftpSessionFactory;
|
||||
}*/
|
||||
}
|
||||
Reference in New Issue
Block a user