From d9eb925235b8041b16f7daa2f0b48b2d263261cd Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 27 Oct 2010 17:54:19 -0400 Subject: [PATCH] INT-1562 formatting --- .../sftp/QueuedSftpSessionPool.java | 43 ++++++----- .../integration/sftp/SftpConstants.java | 4 +- .../integration/sftp/SftpEntryNamer.java | 7 +- .../sftp/SftpSendingMessageHandler.java | 61 ++++++--------- .../integration/sftp/SftpSession.java | 48 ++++++------ .../integration/sftp/SftpSessionFactory.java | 53 +++++++------ .../integration/sftp/SftpSessionPool.java | 17 +++-- ...SftpMessageSendingConsumerFactoryBean.java | 58 ++++++++------ .../sftp/config/SftpNamespaceHandler.java | 19 +++-- ...SynchronizingMessageSourceFactoryBean.java | 48 +++++------- .../sftp/config/SftpSessionUtils.java | 12 +-- ...tpInboundRemoteFileSystemSynchronizer.java | 76 +++++++++---------- 12 files changed, 225 insertions(+), 221 deletions(-) diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/QueuedSftpSessionPool.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/QueuedSftpSessionPool.java index a783b56074..d22d8ca7c5 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/QueuedSftpSessionPool.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/QueuedSftpSessionPool.java @@ -13,27 +13,34 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.sftp; -import org.springframework.beans.factory.InitializingBean; +package org.springframework.integration.sftp; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.Assert; /** * This approach - of having a SessionPool ({@link SftpSessionPool}) that has an * implementation of Queued*SessionPool ({@link QueuedSftpSessionPool}) - was - * taken pretty directly from the incredibly good Spring IntegrationFTP adapter. + * taken almost directly from the Spring Integration FTP adapter. * * @author Josh Long * @since 2.0 */ public class QueuedSftpSessionPool implements SftpSessionPool, InitializingBean { + public static final int DEFAULT_POOL_SIZE = 10; - private Queue queue; + + + private volatile Queue queue; + private final SftpSessionFactory sftpSessionFactory; - private int maxPoolSize; + + private final int maxPoolSize; + public QueuedSftpSessionPool(SftpSessionFactory factory) { this(DEFAULT_POOL_SIZE, factory); @@ -44,34 +51,32 @@ public class QueuedSftpSessionPool implements SftpSessionPool, InitializingBean this.maxPoolSize = maxPoolSize; } + public void afterPropertiesSet() throws Exception { - assert maxPoolSize > 0 : "poolSize must be greater than 0!"; - queue = new ArrayBlockingQueue(maxPoolSize, true); // size, faireness to avoid starvation - assert sftpSessionFactory != null : "sftpSessionFactory must not be null!"; + Assert.notNull(this.sftpSessionFactory, "sftpSessionFactory must not be null"); + Assert.isTrue(this.maxPoolSize > 0, "poolSize must be greater than 0"); + this.queue = new ArrayBlockingQueue(this.maxPoolSize, true); // size, fairness to avoid starvation } public SftpSession getSession() throws Exception { SftpSession session = this.queue.poll(); - if (null == session) { session = this.sftpSessionFactory.getObject(); - - if (queue.size() < maxPoolSize) { - queue.add(session); + if (this.queue.size() < this.maxPoolSize) { + this.queue.add(session); } } - if (null == session) { session = queue.poll(); } - return session; } public void release(SftpSession session) { if (queue.size() < maxPoolSize) { queue.add(session); // somehow one snuck in before session was finished! - } else { + } + else { dispose(session); } } @@ -80,18 +85,16 @@ public class QueuedSftpSessionPool implements SftpSessionPool, InitializingBean if (s == null) { return; } - - if (queue.contains(s)) //this should never happen, but if it does ... - { + if (queue.contains(s)) { + //this should never happen, but if it does... queue.remove(s); } - if ((s.getChannel() != null) && s.getChannel().isConnected()) { s.getChannel().disconnect(); } - if (s.getSession().isConnected()) { s.getSession().disconnect(); } } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpConstants.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpConstants.java index b136974eff..acb45bd6e5 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpConstants.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpConstants.java @@ -13,12 +13,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.sftp; +package org.springframework.integration.sftp; /** * @author Josh Long */ public class SftpConstants { + public static final String SFTP_REMOTE_DIRECTORY_HEADER = "SFTP_REMOTE_DIRECTORY_HEADER"; + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpEntryNamer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpEntryNamer.java index 2c822d439c..fa55db4366 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpEntryNamer.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpEntryNamer.java @@ -13,13 +13,15 @@ * 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.integration.file.entries.EntryNamer; +import com.jcraft.jsch.ChannelSftp; + /** - * Knows how to name a {@link com.jcraft.jsch.ChannelSftp.LsEntry} instance + * Stratgy for naming a {@link com.jcraft.jsch.ChannelSftp.LsEntry} instance. * * @author Josh Long */ @@ -28,4 +30,5 @@ public class SftpEntryNamer implements EntryNamer { public String nameOf(ChannelSftp.LsEntry entry) { return entry.getFilename(); } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSendingMessageHandler.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSendingMessageHandler.java index 3f28376cdb..8bbd05c3d8 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSendingMessageHandler.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSendingMessageHandler.java @@ -41,11 +41,11 @@ import org.springframework.util.Assert; import org.springframework.util.FileCopyUtils; import com.jcraft.jsch.ChannelSftp; +import com.sun.xml.internal.messaging.saaj.packaging.mime.MessagingException; /** - * Sending a message payload to a remote SFTP endpoint. For now, we assume that the payload of the inbound message is of - * type {@link java.io.File}. Perhaps we could support a payload of java.io.InputStream with a Header designating the file - * name? + * Sends message payloads to a remote SFTP endpoint. + * Assumes that the payload of the inbound message is of type {@link java.io.File}. * * @author Josh Long * @since 2.0 @@ -65,7 +65,7 @@ public class SftpSendingMessageHandler implements MessageHandler, InitializingBe private volatile Resource temporaryBufferFolder = new FileSystemResource(SystemUtils.getJavaIoTmpDir()); - private volatile boolean afterPropertiesSetRan; + private volatile boolean initialized; private volatile String charset = Charset.defaultCharset().name(); @@ -88,7 +88,7 @@ public class SftpSendingMessageHandler implements MessageHandler, InitializingBe } public String getRemoteDirectory() { - return remoteDirectory; + return this.remoteDirectory; } public void setCharset(String charset) { @@ -96,19 +96,16 @@ public class SftpSendingMessageHandler implements MessageHandler, InitializingBe } public void afterPropertiesSet() throws Exception { - Assert.state(this.pool != null, "the pool can't be null!"); - temporaryBufferFolderFile = this.temporaryBufferFolder.getFile(); - if (!afterPropertiesSetRan) { + Assert.notNull(this.pool, "the pool must not be null"); + this.temporaryBufferFolderFile = this.temporaryBufferFolder.getFile(); + if (!this.initialized) { if (StringUtils.isEmpty(this.remoteDirectory)) { - remoteDirectory = null; + this.remoteDirectory = null; } - this.afterPropertiesSetRan = true; + this.initialized = true; } } - - /* Ugh this needs to be put in a convenient place accessible for all the file:, sftp:, and ftp:* adapters */ - private File handleFileMessage(File sourceFile, File tempFile, File resultFile) throws IOException { if (sourceFile.renameTo(resultFile)) { return resultFile; @@ -131,13 +128,13 @@ public class SftpSendingMessageHandler implements MessageHandler, InitializingBe return resultFile; } - private File redeemForStorableFile(Message msg) throws MessageDeliveryException { + private File redeemForStorableFile(Message message) throws MessageDeliveryException { try { - Object payload = msg.getPayload(); - String generateFileName = this.fileNameGenerator.generateFileName(msg); - File tempFile = new File(temporaryBufferFolderFile, generateFileName + TEMPORARY_FILE_SUFFIX); - File resultFile = new File(temporaryBufferFolderFile, generateFileName); - File sendableFile; + Object payload = message.getPayload(); + String generateFileName = this.fileNameGenerator.generateFileName(message); + File tempFile = new File(this.temporaryBufferFolderFile, generateFileName + TEMPORARY_FILE_SUFFIX); + File resultFile = new File(this.temporaryBufferFolderFile, generateFileName); + File sendableFile = null; if (payload instanceof String) { sendableFile = this.handleStringMessage((String) payload, tempFile, resultFile, this.charset); } @@ -147,30 +144,23 @@ public class SftpSendingMessageHandler implements MessageHandler, InitializingBe else if (payload instanceof byte[]) { sendableFile = this.handleByteArrayMessage((byte[]) payload, tempFile, resultFile); } - else { - sendableFile = null; - } return sendableFile; } catch (Throwable th) { - throw new MessageDeliveryException(msg); + throw new MessageDeliveryException(message); } } - - /* Ugh this needs to be put in a convenient place accessible for all the file:, sftp:, and ftp:* adapters */ - public void handleMessage(final Message message) { - Assert.state(this.pool != null, "need a working pool"); + Assert.notNull(this.pool, "the pool must not be null"); File inboundFilePayload = this.redeemForStorableFile(message); try { if ((inboundFilePayload != null) && inboundFilePayload.exists()) { sendFileToRemoteEndpoint(message, inboundFilePayload); } } - catch (Throwable thr) { - // logger.debug("recieved an exception.", thr); - throw new MessageDeliveryException(message, "couldn't deliver the message!", thr); + catch (Exception e) { + throw new MessageDeliveryException(message, "failed to deliver the message", e); } finally { if (inboundFilePayload != null && inboundFilePayload.exists()) @@ -178,11 +168,11 @@ public class SftpSendingMessageHandler implements MessageHandler, InitializingBe } } - private boolean sendFileToRemoteEndpoint(Message message, File file) throws Throwable { - assert this.pool != null : "need a working pool"; + private boolean sendFileToRemoteEndpoint(Message message, File file) throws Exception { + Assert.notNull(this.pool, "pool must not be null"); SftpSession session = this.pool.getSession(); if (session == null) { - throw new RuntimeException("the session returned from the pool is null, can't possibly proceed."); + throw new MessagingException("The session returned from the pool is null, cannot proceed."); } session.start(); ChannelSftp sftp = session.getChannel(); @@ -190,7 +180,6 @@ public class SftpSendingMessageHandler implements MessageHandler, InitializingBe try { fileInputStream = new FileInputStream(file); String baseOfRemotePath = StringUtils.isEmpty(this.remoteDirectory) ? StringUtils.EMPTY : remoteDirectory; // the safe default - // logger.debug("going to send " + file.getAbsolutePath() + " to a remote sftp endpoint"); String dynRd = null; MessageHeaders messageHeaders = null; if (message != null) { @@ -210,8 +199,8 @@ public class SftpSendingMessageHandler implements MessageHandler, InitializingBe } finally { IOUtils.closeQuietly(fileInputStream); - if (pool != null) { - pool.release(session); + if (this.pool != null) { + this.pool.release(session); } } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSession.java index 233d5048a9..4a4b8fecec 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSession.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.sftp; import com.jcraft.jsch.ChannelSftp; @@ -23,9 +24,7 @@ import org.apache.commons.lang.StringUtils; import java.io.InputStream; - /** - * c * There are many ways to create a {@link SftpSession} just as there are many ways to SSH into a remote system. * You may use a username and password, you may use a username and private key, you may use a username and a private key with a password, etc. *

@@ -36,10 +35,15 @@ import java.io.InputStream; * @author Mario Gray */ public class SftpSession { + private volatile ChannelSftp channel; + private volatile Session session; + private String privateKey; + private String privateKeyPassphrase; + private volatile UserInfo userInfo; @@ -66,44 +70,41 @@ public class SftpSession { * to surmount that, we need the private key passphrase. Specify that here. * @throws Exception thrown if any of a myriad of scenarios plays out */ - public SftpSession(String userName, String hostName, String userPassword, int port, String knownHostsFile, InputStream knownHostsInputStream, String privateKey, String pvKeyPassPhrase) - throws Exception { - JSch jSch = new JSch(); + public SftpSession(String userName, String hostName, String userPassword, int port, String knownHostsFile, + InputStream knownHostsInputStream, String privateKey, String pvKeyPassPhrase) throws Exception { + JSch jSch = new JSch(); if (port <= 0) { port = 22; } - this.privateKey = privateKey; this.privateKeyPassphrase = pvKeyPassPhrase; - if (!StringUtils.isEmpty(knownHostsFile)) { jSch.setKnownHosts(knownHostsFile); - } else if (null != knownHostsInputStream) { + } + else if (null != knownHostsInputStream) { jSch.setKnownHosts(knownHostsInputStream); } - // private key if (!StringUtils.isEmpty(this.privateKey)) { if (!StringUtils.isEmpty(privateKeyPassphrase)) { jSch.addIdentity(this.privateKey, privateKeyPassphrase); - } else { + } + else { jSch.addIdentity(this.privateKey); } } - - session = jSch.getSession(userName, hostName, port); - + this.session = jSch.getSession(userName, hostName, port); if (!StringUtils.isEmpty(userPassword)) { - session.setPassword(userPassword); + this.session.setPassword(userPassword); } - - userInfo = new OptimisticUserInfoImpl(userPassword); - session.setUserInfo(userInfo); - session.connect(); - channel = (ChannelSftp) session.openChannel("sftp"); + this.userInfo = new OptimisticUserInfoImpl(userPassword); + this.session.setUserInfo(userInfo); + this.session.connect(); + this.channel = (ChannelSftp) this.session.openChannel("sftp"); } + public ChannelSftp getChannel() { return channel; } @@ -118,15 +119,17 @@ public class SftpSession { } } + /** * this is a simple, optimistic implementation of this interface. It simply returns in the positive where possible * and handles interactive authentication (ie, 'Please enter your password: ' prompts are dispatched automatically using this) */ private static class OptimisticUserInfoImpl implements UserInfo { - private String pw; + + private String password; public OptimisticUserInfoImpl(String password) { - this.pw = password; + this.password = password; } public String getPassphrase() { @@ -134,7 +137,7 @@ public class SftpSession { } public String getPassword() { - return pw; + return password; } public boolean promptPassphrase(String string) { @@ -152,4 +155,5 @@ public class SftpSession { public void showMessage(String string) { } } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSessionFactory.java index 022e564553..e990e9b009 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSessionFactory.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSessionFactory.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.sftp; import org.springframework.beans.factory.FactoryBean; @@ -20,43 +21,30 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; import org.springframework.util.StringUtils; - /** - * Factories {@link SftpSession} instances. There are lots of ways to construct a - * {@link SftpSession} instance, and not all of them are obvious. This factory - * does its best to make it work. + * Factory for creating {@link SftpSession} instances. There are lots of ways to construct a + * {@link SftpSession} instance, and not all of them are obvious. This factory should help. * * @author Josh Long * @author Mario Gray */ public class SftpSessionFactory implements FactoryBean, InitializingBean { + private volatile String knownHosts; + private volatile String password; + private volatile String privateKey; + private volatile String privateKeyPassphrase; + private volatile String remoteHost; + private volatile String user; + private volatile int port = 22; // the default - public void afterPropertiesSet() throws Exception { - Assert.hasText(this.remoteHost, "remoteHost can't be empty!"); - Assert.hasText(this.user, "user can't be empty!"); - Assert.state(StringUtils.hasText(this.password) || StringUtils.hasText(this.privateKey) || StringUtils.hasText(this.privateKeyPassphrase), - "you must configure either a password or a private key and/or a private key passphrase!"); - Assert.state(this.port >= 0, "port must be a valid number! "); - } - public SftpSession getObject() throws Exception { - return new SftpSession(this.user, this.remoteHost, this.password, this.port, this.knownHosts, null, this.privateKey, this.privateKeyPassphrase); - } - - public Class getObjectType() { - return SftpSession.class; - } - - public boolean isSingleton() { - return false; - } public void setKnownHosts(String knownHosts) { this.knownHosts = knownHosts; @@ -85,4 +73,25 @@ public class SftpSessionFactory implements FactoryBean, Initializin public void setUser(String user) { this.user = user; } + + public void afterPropertiesSet() throws Exception { + Assert.hasText(this.remoteHost, "remoteHost must not be empty"); + Assert.hasText(this.user, "user mut not be empty"); + Assert.state(StringUtils.hasText(this.password) || StringUtils.hasText(this.privateKey) || StringUtils.hasText(this.privateKeyPassphrase), + "either a password or a private key and/or a private key passphrase is required"); + Assert.state(this.port >= 0, "port must be a positive number"); + } + + public SftpSession getObject() throws Exception { + return new SftpSession(this.user, this.remoteHost, this.password, this.port, this.knownHosts, null, this.privateKey, this.privateKeyPassphrase); + } + + public Class getObjectType() { + return SftpSession.class; + } + + public boolean isSingleton() { + return false; + } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSessionPool.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSessionPool.java index 6e904f6012..0767329f46 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSessionPool.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpSessionPool.java @@ -13,29 +13,30 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.sftp; /** - * Holds instances of {@link SftpSession} since they're stateful - * and might be in use while another run happens. + * Holds instances of {@link SftpSession} since they are stateful + * and might be in use while another operation runs. * * @author Josh Long */ public interface SftpSessionPool { + /** - * this returns a session that can be used to connct to an sftp instance and perform operations + * Returns a session that can be used to connect to an sftp instance and perform operations * * @return the session from the pool ready to be connected to. - * @throws Exception thrown if theres any of the numerous faults possible when trying to connect to the remote - * server + * @throws Exception if any fault occurs when trying to connect to the remote server */ SftpSession getSession() throws Exception; /** - * Frees up the client. Im not sure what the meaningful semantics of this are. Perhaps it just calls (session - * ,channel).disconnect() ? - * + * Releases the session. + * * @param session the session to relinquish / renew */ void release(SftpSession session); + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpMessageSendingConsumerFactoryBean.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpMessageSendingConsumerFactoryBean.java index 56a092ace0..bfd0726c6a 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpMessageSendingConsumerFactoryBean.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpMessageSendingConsumerFactoryBean.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.sftp.config; import org.springframework.beans.factory.FactoryBean; @@ -20,46 +21,32 @@ import org.springframework.integration.sftp.QueuedSftpSessionPool; import org.springframework.integration.sftp.SftpSendingMessageHandler; import org.springframework.integration.sftp.SftpSessionFactory; - /** - * Supports the construction of a MessagHandler that knows how to take inbound #java.io.File objects and send them to a - * remote destination. + * Supports the construction of a MessagHandler that knows how to take inbound File objects + * and send them to a remote destination. * * @author Josh Long */ public class SftpMessageSendingConsumerFactoryBean implements FactoryBean { + private String host; + private String keyFile; + private String keyFilePassword; + private String password; + private String remoteDirectory; + private String username; + private boolean autoCreateDirectories; + private int port; + private String charset; - public SftpSendingMessageHandler getObject() throws Exception { - SftpSessionFactory sessionFactory = SftpSessionUtils.buildSftpSessionFactory( - this.host, this.password, this.username, this.keyFile, this.keyFilePassword, this.port); - - QueuedSftpSessionPool queuedSFTPSessionPool = new QueuedSftpSessionPool(15, sessionFactory); - queuedSFTPSessionPool.afterPropertiesSet(); - - SftpSendingMessageHandler sftpSendingMessageHandler = new SftpSendingMessageHandler(queuedSFTPSessionPool); - sftpSendingMessageHandler.setRemoteDirectory(this.remoteDirectory); - sftpSendingMessageHandler.setCharset(this.charset); - sftpSendingMessageHandler.afterPropertiesSet(); - - return sftpSendingMessageHandler; - } - - public Class getObjectType() { - return SftpSendingMessageHandler.class; - } - - public boolean isSingleton() { - return false; - } public void setAutoCreateDirectories(final boolean autoCreateDirectories) { this.autoCreateDirectories = autoCreateDirectories; @@ -96,4 +83,25 @@ public class SftpMessageSendingConsumerFactoryBean implements FactoryBean getObjectType() { + return SftpSendingMessageHandler.class; + } + + public boolean isSingleton() { + return false; + } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpNamespaceHandler.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpNamespaceHandler.java index e7d663a27b..ef05ab94dd 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpNamespaceHandler.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpNamespaceHandler.java @@ -16,35 +16,36 @@ package org.springframework.integration.sftp.config; +import org.w3c.dom.Element; + import org.springframework.beans.BeanMetadataElement; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; -import org.springframework.beans.factory.xml.NamespaceHandlerSupport; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler; import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; -import org.w3c.dom.Element; /** * Provides namespace support for using SFTP. - * This is very largely based on the FTP support by Iwein Fuld. + * This is largely based on the FTP support by Iwein Fuld. * * @author Josh Long */ -@SuppressWarnings("unused") -public class SftpNamespaceHandler extends NamespaceHandlerSupport { +public class SftpNamespaceHandler extends AbstractIntegrationNamespaceHandler { public void init() { registerBeanDefinitionParser("inbound-channel-adapter", new SFTPMessageSourceBeanDefinitionParser()); registerBeanDefinitionParser("outbound-channel-adapter", new SFTPMessageSendingConsumerBeanDefinitionParser()); } + /** * Configures an object that can take inbound messages and send them. */ private static class SFTPMessageSendingConsumerBeanDefinitionParser extends AbstractOutboundChannelAdapterParser { + @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SftpMessageSendingConsumerFactoryBean.class.getName()); @@ -56,15 +57,13 @@ public class SftpNamespaceHandler extends NamespaceHandlerSupport { } /** - * Configures an object that can recieve files from a remote SFTP endpoint and broadcast their arrival to the - * consumer + * Configures an object that can receive files from a remote SFTP endpoint and broadcast their arrival to the consumer. */ private static class SFTPMessageSourceBeanDefinitionParser extends AbstractPollingInboundChannelAdapterParser { @Override protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean.class.getName()); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean.class.getName()); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "filter"); for (String p : "filename-pattern,auto-create-directories,username,password,host,key-file,key-file-password,remote-directory,local-directory-path,auto-delete-remote-files-on-sync".split(",")) { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, p); diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean.java index 804082a0ae..828536a39d 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.sftp.config; import com.jcraft.jsch.ChannelSftp; @@ -34,75 +35,79 @@ import org.springframework.util.StringUtils; import java.io.File; - /** * Factory bean to hide the fairly complex configuration possibilities for an SFTP endpoint * * @author Josh Long */ -public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends AbstractFactoryBean implements ResourceLoaderAware { +public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean + extends AbstractFactoryBean implements ResourceLoaderAware { private volatile ResourceLoader resourceLoader; + private volatile Resource localDirectoryResource; + private volatile String localDirectoryPath; + private volatile String autoCreateDirectories; + private volatile String autoDeleteRemoteFilesOnSync; + private volatile String filenamePattern; + private volatile EntryListFilter filter; + private int port = 22; + private String host; + private String keyFile; + private String keyFilePassword; + private String remoteDirectory; + private String username; + private String password; - @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 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; } @@ -110,7 +115,6 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends A /** * Set the remote directory to synchronize with */ - @SuppressWarnings("unused") public void setRemoteDirectory(String remoteDirectory) { this.remoteDirectory = remoteDirectory; } @@ -118,7 +122,6 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends A /** * Set the user name to be used for authentication with the remote server */ - @SuppressWarnings("unused") public void setUsername(String username) { this.username = username; } @@ -127,7 +130,6 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends A * Set the password to be used for authentication with the remote server * @param password */ - @SuppressWarnings("unused") public void setPassword(String password) { this.password = password; } @@ -152,11 +154,9 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends A * @return Fully configured SftpInboundRemoteFileSystemSynchronizingMessageSource */ @Override - protected SftpInboundRemoteFileSystemSynchronizingMessageSource createInstance() - throws Exception { + protected SftpInboundRemoteFileSystemSynchronizingMessageSource createInstance() throws Exception { boolean autoCreatDirs = Boolean.parseBoolean(this.autoCreateDirectories); boolean ackRemoteDir = Boolean.parseBoolean(this.autoDeleteRemoteFilesOnSync); - SftpInboundRemoteFileSystemSynchronizingMessageSource sftpMsgSrc = new SftpInboundRemoteFileSystemSynchronizingMessageSource(); sftpMsgSrc.setAutoCreateDirectories(autoCreatDirs); @@ -166,27 +166,22 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends A File sftpTmp = new File(tmp, "sftpInbound"); this.localDirectoryPath = "file://" + sftpTmp.getAbsolutePath(); } - this.localDirectoryResource = this.resourceFromString(localDirectoryPath); // remote predicates SftpEntryNamer sftpEntryNamer = new SftpEntryNamer(); CompositeEntryListFilter compositeFtpFileListFilter = new CompositeEntryListFilter(); - if (StringUtils.hasText(this.filenamePattern)) { PatternMatchingEntryListFilter ftpFilePatternMatchingEntryListFilter = new PatternMatchingEntryListFilter(sftpEntryNamer, filenamePattern); compositeFtpFileListFilter.addFilter(ftpFilePatternMatchingEntryListFilter); } - if (this.filter != null) { compositeFtpFileListFilter.addFilter(this.filter); } - this.filter = compositeFtpFileListFilter; // pools SftpSessionFactory sessionFactory = SftpSessionUtils.buildSftpSessionFactory(this.host, this.password, this.username, this.keyFile, this.keyFilePassword, this.port); - QueuedSftpSessionPool pool = new QueuedSftpSessionPool(15, sessionFactory); pool.afterPropertiesSet(); @@ -197,8 +192,8 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends A sftpSync.setFilter(compositeFtpFileListFilter); sftpSync.setBeanFactory(this.getBeanFactory()); sftpSync.setRemotePath(this.remoteDirectory); - sftpSync.afterPropertiesSet(); // todo is this correct ? - sftpSync.start(); //todo + sftpSync.afterPropertiesSet(); + sftpSync.start(); sftpMsgSrc.setRemotePredicate(compositeFtpFileListFilter); sftpMsgSrc.setSynchronizer(sftpSync); @@ -209,14 +204,13 @@ public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends A sftpMsgSrc.setAutoStartup(true); sftpMsgSrc.afterPropertiesSet(); sftpMsgSrc.start(); - return sftpMsgSrc; } private Resource resourceFromString(String path) { ResourceEditor resourceEditor = new ResourceEditor(this.resourceLoader); resourceEditor.setAsText(path); - return (Resource) resourceEditor.getValue(); } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpSessionUtils.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpSessionUtils.java index 27fbbd7e8e..d9aaeae580 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpSessionUtils.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpSessionUtils.java @@ -13,17 +13,18 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.sftp.config; import org.springframework.integration.sftp.SftpSessionFactory; - /** - * Provides a single place to handle this tedious chore. + * Utility methods for SFTP Session management. * * @author Josh Long */ -public class SftpSessionUtils { +public abstract class SftpSessionUtils { + /** * This method hides the minutae required to build an #SftpSessionFactory. * @@ -39,8 +40,7 @@ public class SftpSessionUtils { * commands against a remote SFTP/SSH filesystem * @throws Exception thrown in case of darned near anything */ - public static SftpSessionFactory buildSftpSessionFactory(String host, String pw, String usr, String pvKey, String pvKeyPass, int port) - throws Exception { + public static SftpSessionFactory buildSftpSessionFactory(String host, String pw, String usr, String pvKey, String pvKeyPass, int port) throws Exception { SftpSessionFactory sftpSessionFactory = new SftpSessionFactory(); sftpSessionFactory.setPassword(pw); sftpSessionFactory.setPort(port); @@ -49,7 +49,7 @@ public class SftpSessionUtils { sftpSessionFactory.setPrivateKey(pvKey); sftpSessionFactory.setPrivateKeyPassphrase(pvKeyPass); sftpSessionFactory.afterPropertiesSet(); - return sftpSessionFactory; } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizer.java index 75b43b7281..2c922976db 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizer.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizer.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.sftp.impl; import com.jcraft.jsch.ChannelSftp; @@ -34,13 +35,13 @@ import java.io.IOException; import java.io.InputStream; import java.util.Collection; - /** - * This handles the synchronization between a remote SFTP endpoint and a local mount + * Gandles the synchronization between a remote SFTP endpoint and a local mount. * * @author Josh Long */ public class SftpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemoteFileSystemSychronizer { + /** * the path on the remote mount */ @@ -51,62 +52,61 @@ public class SftpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemo */ private volatile SftpSessionPool clientPool; + public void setRemotePath(String remotePath) { this.remotePath = remotePath; } - @Override - protected void onInit() throws Exception { - 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(); - } - } - @Required public void setClientPool(SftpSessionPool clientPool) { this.clientPool = clientPool; } - @SuppressWarnings("ignored") - private boolean copyFromRemoteToLocalDirectory(SftpSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir) - throws Exception { + @Override + protected Trigger getTrigger() { + return new PeriodicTrigger(10 * 1000); + } + + @Override + protected void onInit() throws Exception { + Assert.notNull(this.clientPool, "'clientPool' must not be null"); + Assert.notNull(this.remotePath, "'remotePath' must not be null"); + if (this.shouldDeleteSourceFile) { + this.entryAcknowledgmentStrategy = new DeletionEntryAcknowledgmentStrategy(); + } + } + + 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 { + } + finally { IOUtils.closeQuietly(in); IOUtils.closeQuietly(fileOutputStream); } - if (tmpLocalTarget.renameTo(localFile)) { this.acknowledge(sftpSession, entry); } - return true; - } catch (Throwable th) { - logger.error("exception thrown in #copyFromRemoteToLocalDirectory", th); } - } else { + catch (Throwable th) { + logger.error("failure occurred while copying from remote to local directory", th); + } + } + else { return true; } - return false; } @@ -114,47 +114,39 @@ public class SftpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemo @SuppressWarnings("unchecked") protected void syncRemoteToLocalFileSystem() throws Exception { SftpSession session = null; - try { session = clientPool.getSession(); session.start(); - ChannelSftp channelSftp = session.getChannel(); Collection beforeFilter = channelSftp.ls(remotePath); ChannelSftp.LsEntry[] entries = (beforeFilter == null) ? new ChannelSftp.LsEntry[0] : beforeFilter.toArray(new ChannelSftp.LsEntry[beforeFilter.size()]); Collection files = this.filter.filterEntries(entries); - for (ChannelSftp.LsEntry lsEntry : files) { if ((lsEntry != null) && !lsEntry.getAttrs().isDir() && !lsEntry.getAttrs().isLink()) { copyFromRemoteToLocalDirectory(session, lsEntry, this.localDirectory); } } - } catch (IOException e) { + } + catch (IOException e) { throw new MessagingException("couldn't synchronize remote to local directory", e); - } finally { + } + finally { if ((session != null) && (clientPool != null)) { clientPool.release(session); } } } - @Override - protected Trigger getTrigger() { - return new PeriodicTrigger(10 * 1000); - } - - class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy { - public void acknowledge(Object useful, ChannelSftp.LsEntry msg) - throws Exception { + private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy { + + public void acknowledge(Object useful, ChannelSftp.LsEntry msg) throws Exception { SftpSession sftpSession = (SftpSession) useful; - String remoteFqPath = remotePath + "/" + msg.getFilename(); - sftpSession.getChannel().rm(remoteFqPath); - if (logger.isDebugEnabled()) { logger.debug("deleted " + msg.getFilename()); } } } + }