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 6dd36ce8f9..c9057fa8f1 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 @@ -38,7 +38,6 @@ public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChann @Override protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) { String sessionFactoryName = element.getAttribute("session-factory"); - String autoStartup = element.getAttribute("auto-startup"); String fileNamePattern = element.getAttribute("filename-pattern"); String filter = element.getAttribute("filter"); boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern); @@ -49,12 +48,11 @@ public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChann "is allowed on SFTP inbound adapter"); } } - BeanDefinitionBuilder sessionPollBuilder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.sftp.session.QueuedSftpSessionPool"); - sessionPollBuilder.addConstructorArgReference(sessionFactoryName); - sessionPollBuilder.addPropertyValue("autoStartup", autoStartup); + BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition( + "org.springframework.integration.sftp.session.CachingSftpSessionFactory"); + sessionFactoryBuilder.addConstructorArgReference(sessionFactoryName); String sessionPollName = BeanDefinitionReaderUtils.registerWithGeneratedName( - sessionPollBuilder.getBeanDefinition(), parserContext.getRegistry()); + sessionFactoryBuilder.getBeanDefinition(), parserContext.getRegistry()); BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition( "org.springframework.integration.sftp.inbound.SftpInboundSynchronizer"); synchronizerBuilder.addConstructorArgReference(sessionPollName); diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpOutboundChannelAdapterParser.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpOutboundChannelAdapterParser.java index df6c343f23..5e6e975988 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpOutboundChannelAdapterParser.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpOutboundChannelAdapterParser.java @@ -40,7 +40,7 @@ public class SftpOutboundChannelAdapterParser extends AbstractOutboundChannelAda @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { BeanDefinitionBuilder sessionPoolBuilder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.sftp.session.QueuedSftpSessionPool"); + "org.springframework.integration.sftp.session.CachingSftpSessionFactory"); sessionPoolBuilder.addConstructorArgReference(element.getAttribute("session-factory")); String sessionPoolName = BeanDefinitionReaderUtils.registerWithGeneratedName( sessionPoolBuilder.getBeanDefinition(), parserContext.getRegistry()); 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/CachingSftpSessionFactory.java similarity index 51% rename from spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java rename to spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/CachingSftpSessionFactory.java index c5c2641b8c..cbb5a23d56 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/CachingSftpSessionFactory.java @@ -21,69 +21,56 @@ import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.locks.ReentrantLock; import java.util.logging.Logger; -import org.springframework.context.SmartLifecycle; +import org.springframework.beans.factory.DisposableBean; import org.springframework.util.Assert; -import com.jcraft.jsch.Channel; +import com.jcraft.jsch.ChannelSftp; /** * This approach - of having a SessionPool ({@link SftpSessionPool}) that has an - * implementation of a queued SessionPool ({@link QueuedSftpSessionPool}) - was + * implementation of a queued SessionPool ({@link CachingSftpSessionFactory}) - was * taken almost directly from the Spring Integration FTP adapter. * * @author Josh Long * @author Oleg Zhurakousky * @since 2.0 */ -public class QueuedSftpSessionPool implements SftpSessionFactory, SmartLifecycle { +public class CachingSftpSessionFactory implements SftpSessionFactory, DisposableBean { - private static Logger logger = Logger.getLogger(QueuedSftpSessionPool.class.getName()); + private static Logger logger = Logger.getLogger(CachingSftpSessionFactory.class.getName()); public static final int DEFAULT_POOL_SIZE = 10; - private volatile Queue queue; + private final Queue queue; private final SimpleSftpSessionFactory sftpSessionFactory; private final int maxPoolSize; - - private volatile boolean running; - - private volatile boolean autoStartup; - - private volatile int phase = 0; private final ReentrantLock lock = new ReentrantLock(); - public QueuedSftpSessionPool(SimpleSftpSessionFactory factory) { - this(DEFAULT_POOL_SIZE, factory); + public CachingSftpSessionFactory(SimpleSftpSessionFactory sessionFactory) { + this(sessionFactory, DEFAULT_POOL_SIZE); } - public QueuedSftpSessionPool(int maxPoolSize, SimpleSftpSessionFactory sessionFactory) { + public CachingSftpSessionFactory(SimpleSftpSessionFactory sessionFactory, int maxPoolSize) { this.sftpSessionFactory = sessionFactory; this.maxPoolSize = maxPoolSize; + this.queue = new ArrayBlockingQueue(this.maxPoolSize, true); } - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; - } - - public void setPhase(int phase) { - this.phase = phase; - } - public SftpSession getSession() { Assert.notNull(this.queue, "SftpSession is unavailable since the pool component is not started"); this.lock.lock(); try { SftpSession session = this.queue.poll(); if (null == session) { - session = this.sftpSessionFactory.getSession(); + session = sftpSessionFactory.getSession(); } - return session; + return (session != null) ? new PooledSftpSession(session) : null; } finally { this.lock.unlock(); @@ -91,34 +78,17 @@ public class QueuedSftpSessionPool implements SftpSessionFactory, SmartLifecycle } - public void release(SftpSession sftpSession) { - if (this.running) { - this.lock.lock(); - try { - if (queue.size() < maxPoolSize && sftpSession != null) { - queue.add(sftpSession); - } - else { - this.destroySftpSession(sftpSession); - } + public void destroy() { + if (this.queue != null) { + for (SftpSession sftpSession : this.queue) { + this.destroySftpSession(sftpSession); } - finally { - this.lock.unlock(); - } - } - else { - this.destroySftpSession(sftpSession); } } - private void destroySftpSession(SftpSession sftpSession) { try { if (sftpSession != null) { - Channel channel = sftpSession.getChannel(); - if (channel.isConnected()) { - channel.disconnect(); - } sftpSession.disconnect(); } } @@ -129,50 +99,30 @@ public class QueuedSftpSessionPool implements SftpSessionFactory, SmartLifecycle } - // SmartLifeycle implementation + private class PooledSftpSession implements SftpSession { - public boolean isAutoStartup() { - return this.autoStartup; - } + private final SftpSession targetSession; - public int getPhase() { - return this.phase; - } - - public boolean isRunning() { - return this.running; - } - - public void start() { - Assert.isTrue(this.maxPoolSize > 0, "poolSize must be greater than 0"); - this.lock.lock(); - try { - this.queue = new ArrayBlockingQueue(this.maxPoolSize, true); - this.running = true; + private PooledSftpSession(SftpSession targetSession) { + this.targetSession = targetSession; } - finally { - this.lock.unlock(); - } - } - public void stop() { - if (this.queue != null) { - for (SftpSession sftpSession : this.queue) { - this.destroySftpSession(sftpSession); + public ChannelSftp getChannel() { + return targetSession.getChannel(); + } + + public void connect() { + targetSession.connect(); + } + + public void disconnect() { + if (queue.size() < maxPoolSize) { + queue.add(targetSession); + } + else { + targetSession.disconnect(); } } } - public void stop(Runnable callback) { - this.lock.lock(); - try { - this.stop(); - callback.run(); - } - finally { - this.running = false; - this.lock.unlock(); - } - } - } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSession.java index 48565e3945..b144d11e5a 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSession.java @@ -20,6 +20,7 @@ import java.io.InputStream; import org.apache.commons.lang.StringUtils; +import com.jcraft.jsch.Channel; import com.jcraft.jsch.ChannelSftp; import com.jcraft.jsch.JSch; import com.jcraft.jsch.JSchException; @@ -122,6 +123,9 @@ public class DefaultSftpSession implements SftpSession { public void disconnect() { if (targetSession.isConnected()) { targetSession.disconnect(); + if (channel.isConnected()) { + channel.disconnect(); + } } } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapaterParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapaterParserTests.java index 5d0370f832..7d7333299c 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapaterParserTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapaterParserTests.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 static junit.framework.Assert.assertEquals; @@ -30,7 +31,7 @@ import org.springframework.expression.spel.standard.SpelExpression; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.file.FileNameGenerator; import org.springframework.integration.sftp.outbound.SftpSendingMessageHandler; -import org.springframework.integration.sftp.session.QueuedSftpSessionPool; +import org.springframework.integration.sftp.session.CachingSftpSessionFactory; import org.springframework.integration.sftp.session.SimpleSftpSessionFactory; import org.springframework.integration.test.util.TestUtils; @@ -56,7 +57,7 @@ public class OutboundChannelAdapaterParserTests { assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset")); assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolder")); assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolderFile")); - QueuedSftpSessionPool sessionFactory = (QueuedSftpSessionPool) TestUtils.getPropertyValue(handler, "sessionFactory"); + CachingSftpSessionFactory sessionFactory = (CachingSftpSessionFactory) TestUtils.getPropertyValue(handler, "sessionFactory"); SimpleSftpSessionFactory clientFactory = (SimpleSftpSessionFactory) TestUtils.getPropertyValue(sessionFactory, "sftpSessionFactory"); assertEquals("localhost", TestUtils.getPropertyValue(clientFactory, "host")); assertEquals(2222, TestUtils.getPropertyValue(clientFactory, "port"));