From 9f1d9e87ce0e10a9c3ec43326b2a96bfc172aec9 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 17 Nov 2010 23:53:42 -0500 Subject: [PATCH] INT-1631 polishing, moved checkIfRemoteDirExist to synchronier other minor cleanup --- .../SftpInboundChannelAdapterParser.java | 1 - .../sftp/inbound/SftpInboundSynchronizer.java | 52 +++++++++++- ...SftpInboundSynchronizingMessageSource.java | 82 +------------------ .../sftp/session/QueuedSftpSessionPool.java | 3 - ...tpInboundReceiveSample-ignored-context.xml | 6 +- .../sftp/config/SftpInboundReceiveSample.java | 16 +--- 6 files changed, 61 insertions(+), 99 deletions(-) diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java index 814b14635a..4b0225cb60 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java @@ -69,7 +69,6 @@ public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChann BeanDefinitionBuilder messageSourceBuilder = BeanDefinitionBuilder.rootBeanDefinition("org.springframework.integration.sftp.inbound.SftpInboundSynchronizingMessageSource"); - messageSourceBuilder.addConstructorArgReference(sessionPollName); messageSourceBuilder.addPropertyValue("synchronizer", synchronizerBuilder.getBeanDefinition()); if (hasFileNamePattern){ diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java index f6ccb8ca34..8796e54de6 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java @@ -33,6 +33,7 @@ import org.springframework.scheduling.Trigger; import org.springframework.util.Assert; import com.jcraft.jsch.ChannelSftp; +import com.jcraft.jsch.SftpATTRS; /** * Gandles the synchronization between a remote SFTP endpoint and a local mount. @@ -47,6 +48,8 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych * the path on the remote mount */ private volatile String remotePath; + + private volatile boolean autoCreateDirectories; /** * the pool of {@link org.springframework.integration.sftp.session.SftpSessionPool} SFTP sessions @@ -58,6 +61,9 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych this.sessionPool = sessionPool; } + public void setAutoCreateDirectories(boolean autoCreateDirectories) { + this.autoCreateDirectories = autoCreateDirectories; + } public void setRemotePath(String remotePath) { this.remotePath = remotePath; @@ -66,7 +72,6 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych @Override protected Trigger getTrigger() { throw new UnsupportedOperationException("This method curently is not implemented"); - //return new PeriodicTrigger(10 * 1000); } @Override @@ -76,6 +81,48 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych this.entryAcknowledgmentStrategy = new DeletionEntryAcknowledgmentStrategy(); } } + + /** + * 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) { + + ChannelSftp channelSftp = session.getChannel(); + + try { + 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.autoCreateDirectories && (this.sessionPool != null) && (session != null)) { + try { + if (channelSftp != null) { + channelSftp.mkdir(remotePath); + + if (channelSftp.stat(remotePath).isDir()) { + return true; + } + } + } + catch (RuntimeException re) { + throw re; + } + catch (Exception e){ + throw new MessagingException("Failed to auto-create remote directory", e); + } + + } + } + + return false; + } private boolean copyFromRemoteToLocalDirectory(SftpSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir) throws Exception { File fileForLocalDir = localDir.getFile(); @@ -116,7 +163,9 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych SftpSession session = null; try { session = sessionPool.getSession(); + logger.trace("Pooled SftpSession " + this.sessionPool + " from the pool"); session.start(); + this.checkThatRemotePathExists(remotePath, session); ChannelSftp channelSftp = session.getChannel(); Collection beforeFilter = channelSftp.ls(remotePath); ChannelSftp.LsEntry[] entries = (beforeFilter == null) ? new ChannelSftp.LsEntry[0] : @@ -133,6 +182,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych } finally { this.sessionPool.release(session); + logger.trace("Putting SftpSession " + this.sessionPool + " back into the pool"); } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java index da7e3a4cc2..8fad28e563 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java @@ -24,11 +24,8 @@ import org.springframework.integration.MessagingException; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSynchronizingMessageSource; import org.springframework.integration.sftp.filters.SftpPatternMatchingFileListFilter; -import org.springframework.integration.sftp.session.SftpSession; -import org.springframework.integration.sftp.session.SftpSessionPool; import com.jcraft.jsch.ChannelSftp; -import com.jcraft.jsch.SftpATTRS; /** @@ -40,31 +37,13 @@ import com.jcraft.jsch.SftpATTRS; */ public class SftpInboundSynchronizingMessageSource extends AbstractInboundRemoteFileSystemSynchronizingMessageSource { - - /** - * the pool of sessions - */ - private final SftpSessionPool sessionPool; - - /** - * the remote path on the server - */ - private volatile String remoteDirectory; private volatile Pattern filenamePattern; - public SftpInboundSynchronizingMessageSource(SftpSessionPool sessionPool){ - this.sessionPool = sessionPool; - } - public void setFilenamePattern(Pattern filenamePattern) { this.filenamePattern = filenamePattern; } - - public void setRemoteDirectory(String remoteDirectory) { - this.remoteDirectory = remoteDirectory; - } - + public String getComponentType(){ return "sftp:inbound-channel-adapter"; } @@ -77,69 +56,12 @@ public class SftpInboundSynchronizingMessageSource extends */ Message message = this.fileSource.receive(); if (message == null){ - this.checkThatRemotePathExists(this.remoteDirectory); this.synchronizer.syncRemoteToLocalFileSystem(); message = this.fileSource.receive(); } return message; } - /** - * 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 = this.sessionPool.getSession(); - session.start(); - channelSftp = session.getChannel(); - } - catch (RuntimeException re) { - throw re; - } - catch (Exception e){ - throw new MessagingException("Failed to get SftpSession while checking for existance of the remote directory", e); - } - - try { - 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.autoCreateDirectories && (this.sessionPool != null) && (session != null)) { - try { - if (channelSftp != null) { - channelSftp.mkdir(remotePath); - - if (channelSftp.stat(remotePath).isDir()) { - return true; - } - } - } - catch (RuntimeException re) { - throw re; - } - catch (Exception e){ - throw new MessagingException("Failed to auto-create remote directory", e); - } - - } - } - finally { - this.sessionPool.release(session); - } - - return false; - } - @Override protected void onInit() { try { @@ -172,7 +94,9 @@ public class SftpInboundSynchronizingMessageSource extends if (filenamePattern != null) { SftpPatternMatchingFileListFilter sftpFilePatternMatchingEntryListFilter = new SftpPatternMatchingFileListFilter(filenamePattern); + //TODO refactor the design so values don't need to be duplicated this.synchronizer.setFilter(sftpFilePatternMatchingEntryListFilter); + this.synchronizer.setAutoCreateDirectories(this.autoCreateDirectories); } } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java index 17a9d292f4..abc2d25a1b 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java @@ -74,9 +74,6 @@ public class QueuedSftpSessionPool implements SftpSessionPool, SmartLifecycle { SftpSession session = this.queue.poll(); if (null == session) { session = this.sftpSessionFactory.getSession(); - if (this.queue.size() < this.maxPoolSize) { - this.queue.add(session); - } } return session; } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample-ignored-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample-ignored-context.xml index 95f1fff947..538336f8e3 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample-ignored-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample-ignored-context.xml @@ -3,8 +3,10 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration" xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp" + xmlns:task="http://www.springframework.org/schema/task" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd + http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp-2.0.xsd"> @@ -29,7 +31,7 @@ auto-create-directories="false" auto-delete-remote-files-on-sync="false" filename-pattern=".*\.txt$"> - + @@ -40,5 +42,7 @@ + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample.java index ba2ebc6170..512f4be56b 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample.java @@ -29,22 +29,10 @@ public class SftpInboundReceiveSample { @Test @Ignore public void testInbound() throws Exception{ - ClassPathXmlApplicationContext ac = - new ClassPathXmlApplicationContext("SftpInboundReceiveSample-ignored.xml", SftpInboundReceiveSample.class); + new ClassPathXmlApplicationContext("SftpInboundReceiveSample-ignored-context.xml", SftpInboundReceiveSample.class); System.out.println("Done"); - System.in.read(); - ac.stop(); -// File file = new File("/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-sftp/foo.txt"); -// if (file.exists()){ -// Message message = MessageBuilder.withPayload(file).build(); -// MessageChannel inputChannel = ac.getBean("inputChannel", MessageChannel.class); -// inputChannel.send(message); -// Thread.sleep(2000); -// } -// System.out.println("Done"); -// ac.stop(); - + Thread.sleep(3000 * 300); } }