From 7bc4fe2d0b2985b0e775f843b7687546825c75d9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 7 Nov 2013 21:09:33 +0200 Subject: [PATCH] INT-3047 SFTP Multiplex Over Single Connection Enable the use of multiple JSch channels over a single connection. JIRA: https://jira.springsource.org/browse/INT-3047 INT-3047: Polishing locks logic INT-3047 SFTP Reset Cache; Quiesce Shared Sessions Support resetCache() on the CachingSessionFactory - reset a shared JSch session so it is reestablished on next use - immediately close idle sessions - close in-use sessions as they are returned - close the channel when an SftpSession is closed and a shared JSch is being used - physically close a shared JSch session only when the last channel is closed Polishing Forgot to save the ftp.xml. Change CachedSession.epoch to created. INT-3047 Polishing - PR Comments - Also add missing docs for `preserve-timestamp` INT-3047 Polish - PR Comments Also change epoch to nanoseconds and use a simple != comparison with the epoch in which a session was created. --- .../integration/util/SimplePool.java | 5 +- .../remote/session/CachingSessionFactory.java | 52 ++++++- .../remote/session/SharedSessionCapable.java | 39 +++++ .../session/CachingSessionFactoryTests.java | 147 ++++++++++++++++++ .../session/DefaultSftpSessionFactory.java | 108 ++++++++++++- .../integration/sftp/session/SftpSession.java | 27 +++- .../sftp/outbound/SftpOutboundTests.java | 129 +++++++++++++++ .../SftpServerOutboundTests-context.xml | 15 ++ .../outbound/SftpServerOutboundTests.java | 77 +++++++++ src/reference/docbook/ftp.xml | 16 +- src/reference/docbook/sftp.xml | 43 ++++- src/reference/docbook/whats-new.xml | 10 ++ 12 files changed, 649 insertions(+), 19 deletions(-) create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java index 2645c21eda..5814efb0bc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java @@ -192,7 +192,7 @@ public class SimplePool implements Pool { } else if (this.callback.isStale(item)) { if (logger.isDebugEnabled()) { - logger.debug("Received a stale item, will attempt to get a new one."); + logger.debug("Received a stale item " + item + ", will attempt to get a new one."); } doRemoveItem(item); item = doGetItem(); @@ -241,6 +241,9 @@ public class SimplePool implements Pool { } private void doRemoveItem(T item) { + if (logger.isDebugEnabled()){ + logger.debug("Removing " + item + " from the pool"); + } this.allocated.remove(item); this.inUse.remove(item); this.callback.removedFromPool(item); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index 77c1363c8f..d28e8762c8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java @@ -45,6 +45,10 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe private final SimplePool> pool; + private final boolean isSharedSessionCapable; + + private volatile long sharedSessionEpoch; + /** * Create a CachingSessionFactory with an unlimited number of sessions. * @param sessionFactory the underlying session factory. @@ -80,6 +84,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe session.close(); } }); + this.isSharedSessionCapable = sessionFactory instanceof SharedSessionCapable; } @@ -105,7 +110,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe */ @Override public Session getSession() { - return new CachedSession(this.pool.getItem()); + return new CachedSession(this.pool.getItem(), this.sharedSessionEpoch); } /** @@ -116,27 +121,64 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe this.pool.removeAllIdleItems(); } + /** + * Clear the cache of sessions; also any in-use sessions will be closed when + * returned to the cache. + */ + public synchronized void resetCache() { + if (logger.isDebugEnabled()) { + logger.debug("Cache reset; idle sessions will be removed, in-use sessions will be closed when returned"); + } + if (this.isSharedSessionCapable && ((SharedSessionCapable) this.sessionFactory).isSharedSession()) { + ((SharedSessionCapable) this.sessionFactory).resetSharedSession(); + } + long sharedSessionEpoch = System.nanoTime(); + /* + * Spin until we get a new value - nano precision but may be lower resolution. + * We reset the epoch AFTER resetting the shared session so there is no possibility + * of an "old" session being created in the new epoch. There is a slight possibility + * that a "new" session might appear in the old epoch and thus be closed when returned to + * the cache. + */ + while (sharedSessionEpoch == this.sharedSessionEpoch) { + sharedSessionEpoch = System.nanoTime(); + } + this.sharedSessionEpoch = sharedSessionEpoch; + this.pool.removeAllIdleItems(); + } private class CachedSession implements Session { private final Session targetSession; - private boolean released; + private volatile boolean released; - private CachedSession(Session targetSession) { + /** + * The epoch in which this session was created. + */ + private final long sharedSessionEpoch; + + private CachedSession(Session targetSession, long sharedSessionEpoch) { this.targetSession = targetSession; + this.sharedSessionEpoch = sharedSessionEpoch; } @Override public synchronized void close() { if (released) { if (logger.isDebugEnabled()){ - logger.debug("Session already released."); + logger.debug("Session " + targetSession + " already released."); } } else { if (logger.isDebugEnabled()){ - logger.debug("Releasing Session back to the pool."); + logger.debug("Releasing Session " + targetSession + " back to the pool."); + } + if (this.sharedSessionEpoch != CachingSessionFactory.this.sharedSessionEpoch) { + if (logger.isDebugEnabled()){ + logger.debug("Closing session " + targetSession + " after reset."); + } + this.targetSession.close(); } pool.releaseItem(targetSession); released = true; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java new file mode 100644 index 0000000000..f9db8f99bd --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java @@ -0,0 +1,39 @@ +/* + * Copyright 2013 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.file.remote.session; + + +/** + * A {@link SessionFactory} that implements this interface is capable of supporting a shared session. + * + * @author Gary Russell + * @since 3.0 + * + */ +public interface SharedSessionCapable { + + /** + * @return true if this factory uses a shared session. + */ + public abstract boolean isSharedSession(); + + /** + * Resets the shared session so the next {@code #getSession()} will return a session + * using a new connection. + */ + public abstract void resetSharedSession(); + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java new file mode 100644 index 0000000000..f0cca26600 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java @@ -0,0 +1,147 @@ +/* + * Copyright 2013 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.file.remote.session; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +import org.junit.Test; + +import org.springframework.integration.test.util.TestUtils; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class CachingSessionFactoryTests { + + @Test + public void testCacheAndReset() { + TestSessionFactory factory = new TestSessionFactory(); + CachingSessionFactory cache = new CachingSessionFactory(factory); + Session sess1 = cache.getSession(); + assertEquals("session:1", TestUtils.getPropertyValue(sess1, "targetSession.id")); + Session sess2 = cache.getSession(); + assertEquals("session:2", TestUtils.getPropertyValue(sess2, "targetSession.id")); + sess1.close(); + // session back to pool; should be open and reused. + assertTrue(sess1.isOpen()); + sess1 = cache.getSession(); + assertEquals("session:1", TestUtils.getPropertyValue(sess1, "targetSession.id")); + sess1.close(); + assertTrue(sess1.isOpen()); + // reset the cache; should close idle (sess1); sess2 should closed later + cache.resetCache(); + assertFalse(sess1.isOpen()); + sess1 = cache.getSession(); + assertEquals("session:3", TestUtils.getPropertyValue(sess1, "targetSession.id")); + sess1.close(); + assertTrue(sess1.isOpen()); + // session from previous epoch is closed on return + sess2.close(); + assertFalse(sess2.isOpen()); + cache.resetCache(); + assertFalse(sess1.isOpen()); + } + + private class TestSessionFactory implements SessionFactory { + + private int n; + + @Override + public Session getSession() { + return new TestSession("session:" + ++n); + } + + } + + private class TestSession implements Session { + + @SuppressWarnings("unused") + private final String id; + + private volatile boolean open = true; + + private TestSession(String id) { + this.id = id; + } + + @Override + public boolean remove(String path) throws IOException { + return false; + } + + @Override + public String[] list(String path) throws IOException { + return null; + } + + @Override + public void read(String source, OutputStream outputStream) throws IOException { + } + + @Override + public void write(InputStream inputStream, String destination) throws IOException { + } + + @Override + public boolean mkdir(String directory) throws IOException { + return false; + } + + @Override + public void rename(String pathFrom, String pathTo) throws IOException { + } + + @Override + public void close() { + this.open = false; + } + + @Override + public boolean isOpen() { + return this.open; + } + + @Override + public boolean exists(String path) throws IOException { + return false; + } + + @Override + public String[] listNames(String path) throws IOException { + return null; + } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } + + } + +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java index 3f33c12e32..9bad196505 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -17,11 +17,14 @@ package org.springframework.integration.sftp.session; import java.util.Properties; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.ReentrantReadWriteLock; import org.springframework.beans.factory.BeanCreationException; import org.springframework.core.io.Resource; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.remote.session.SharedSessionCapable; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -39,10 +42,11 @@ import com.jcraft.jsch.UserInfo; * @author Mario Gray * @author Oleg Zhurakousky * @author Gunnar Hillert + * @author Gary Russell * * @since 2.0 */ -public class DefaultSftpSessionFactory implements SessionFactory { +public class DefaultSftpSessionFactory implements SessionFactory, SharedSessionCapable { private volatile String host; @@ -76,9 +80,35 @@ public class DefaultSftpSessionFactory implements SessionFactory { private volatile Boolean enableDaemonThread; + private final JSch jsch; - private final JSch jsch = new JSch(); + private final boolean isSharedSession; + private volatile JSchSessionWrapper sharedJschSession; + + private final ReentrantReadWriteLock sharedSessionLock = new ReentrantReadWriteLock(); + + + public DefaultSftpSessionFactory() { + this(false); + } + + /** + * @param isSharedSession + */ + public DefaultSftpSessionFactory(boolean isSharedSession) { + this(new JSch(), isSharedSession); + } + + /** + * Intended for use in tests so the jsch can be mocked. + * @param jsch + * @param isSharedSession + */ + public DefaultSftpSessionFactory(JSch jsch, boolean isSharedSession) { + this.jsch = jsch; + this.isSharedSession = isSharedSession; + } /** * The url of the host you want connect to. This is a mandatory property. @@ -257,9 +287,35 @@ public class DefaultSftpSessionFactory implements SessionFactory { Assert.isTrue(StringUtils.hasText(this.password) || this.privateKey != null, "either a password or a private key is required"); try { - com.jcraft.jsch.Session jschSession = this.initJschSession(); + JSchSessionWrapper jschSession; + if (this.isSharedSession) { + this.sharedSessionLock.readLock().lock(); + try { + if (this.sharedJschSession == null || !this.sharedJschSession.isConnected()) { + this.sharedSessionLock.readLock().unlock(); + this.sharedSessionLock.writeLock().lock(); + try { + if (this.sharedJschSession == null || !this.sharedJschSession.isConnected()) { + this.sharedJschSession = new JSchSessionWrapper(initJschSession()); + } + } + finally { + this.sharedSessionLock.readLock().lock(); + this.sharedSessionLock.writeLock().unlock(); + } + } + } + finally { + this.sharedSessionLock.readLock().unlock(); + } + jschSession = this.sharedJschSession; + } + else { + jschSession = new JSchSessionWrapper(initJschSession()); + } SftpSession sftpSession = new SftpSession(jschSession); sftpSession.connect(); + jschSession.addChannel(); return sftpSession; } catch (Exception e) { @@ -327,6 +383,16 @@ public class DefaultSftpSessionFactory implements SessionFactory { return jschSession; } + @Override + public final boolean isSharedSession() { + return this.isSharedSession; + } + + @Override + public void resetSharedSession() { + Assert.state(this.isSharedSession, "Shared sessions are not being used"); + this.sharedJschSession = null; + } /** * this is a simple, optimistic implementation of the UserInfo interface. @@ -370,4 +436,38 @@ public class DefaultSftpSessionFactory implements SessionFactory { } } + /** + * A wrapper for a JSch session that maintains a channel count and + * physically disconnects when the last channel is closed. + * + */ + public class JSchSessionWrapper { + + private final com.jcraft.jsch.Session session; + + private final AtomicInteger channels = new AtomicInteger(); + + JSchSessionWrapper(com.jcraft.jsch.Session session) { + this.session = session; + } + + public void addChannel() { + this.channels.incrementAndGet(); + } + + public void close() { + if (channels.decrementAndGet() <= 0) { + this.session.disconnect(); + } + } + + public final com.jcraft.jsch.Session getSession() { + return session; + } + + public boolean isConnected() { + return session.isConnected(); + } + + } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java index a3d47250af..aadd108ae6 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java @@ -28,6 +28,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.core.NestedIOException; import org.springframework.integration.file.remote.session.Session; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory.JSchSessionWrapper; import org.springframework.util.Assert; import org.springframework.util.FileCopyUtils; @@ -53,14 +54,24 @@ class SftpSession implements Session { private final com.jcraft.jsch.Session jschSession; + private final JSchSessionWrapper wrapper; + private volatile ChannelSftp channel; + private volatile boolean closed; + public SftpSession(com.jcraft.jsch.Session jschSession) { Assert.notNull(jschSession, "jschSession must not be null"); this.jschSession = jschSession; + this.wrapper = null; } + public SftpSession(DefaultSftpSessionFactory.JSchSessionWrapper wrapper) { + Assert.notNull(wrapper, "wrapper must not be null"); + this.jschSession = wrapper.getSession(); + this.wrapper = wrapper; + } @Override public boolean remove(String path) throws IOException { @@ -151,14 +162,21 @@ class SftpSession implements Session { @Override public void close() { - if (this.jschSession.isConnected()) { - this.jschSession.disconnect(); + this.closed = true; + if (this.wrapper != null) { + this.channel.disconnect(); + this.wrapper.close(); + } + else { + if (this.jschSession.isConnected()) { + this.jschSession.disconnect(); + } } } @Override public boolean isOpen() { - return this.jschSession.isConnected(); + return !this.closed && this.jschSession.isConnected(); } @Override @@ -220,8 +238,8 @@ class SftpSession implements Session { try { if (!this.jschSession.isConnected()) { this.jschSession.connect(); - this.channel = (ChannelSftp) this.jschSession.openChannel("sftp"); } + this.channel = (ChannelSftp) this.jschSession.openChannel("sftp"); if (this.channel != null && !this.channel.isConnected()) { this.channel.connect(); } @@ -230,4 +248,5 @@ class SftpSession implements Session { throw new IllegalStateException("failed to connect", e); } } + } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java index aa3f584584..76950c77d3 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java @@ -17,15 +17,21 @@ package org.springframework.integration.sftp.outbound; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.File; import java.io.FileOutputStream; import java.io.InputStream; +import java.lang.reflect.Constructor; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -36,6 +42,7 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -46,16 +53,20 @@ import org.springframework.integration.core.PollableChannel; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.remote.FileInfo; import org.springframework.integration.file.remote.handler.FileTransferringMessageHandler; +import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; import org.springframework.integration.sftp.session.SftpTestSessionFactory; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.util.FileCopyUtils; import com.jcraft.jsch.ChannelSftp; import com.jcraft.jsch.ChannelSftp.LsEntry; +import com.jcraft.jsch.JSch; +import com.jcraft.jsch.JSchException; import com.jcraft.jsch.SftpATTRS; /** @@ -194,6 +205,7 @@ public class SftpOutboundTests { handler.afterPropertiesSet(); final List madeDirs = new ArrayList(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { madeDirs.add((String) invocation.getArguments()[0]); return null; @@ -206,6 +218,121 @@ public class SftpOutboundTests { assertEquals("/foo/bar/baz", madeDirs.get(2)); } + @Test + public void testSharedSession() throws Exception { + JSch jsch = spy(new JSch()); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class); + ctor.setAccessible(true); + com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch)); + com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch)); + new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true); + new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true); + when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2); + ChannelSftp channel1 = spy(new ChannelSftp()); + ChannelSftp channel2 = spy(new ChannelSftp()); + new DirectFieldAccessor(channel1).setPropertyValue("session", jschSession1); + new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); + when(jschSession1.openChannel("sftp")).thenReturn(channel1, channel2); + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, true); + factory.setHost("host"); + factory.setUser("foo"); + factory.setPassword("bar"); + noopConnect(channel1); + noopConnect(channel2); + Session s1 = factory.getSession(); + Session s2 = factory.getSession(); + assertSame(TestUtils.getPropertyValue(s1, "jschSession"), TestUtils.getPropertyValue(s2, "jschSession")); + } + + @Test + public void testNotSharedSession() throws Exception { + JSch jsch = spy(new JSch()); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class); + ctor.setAccessible(true); + com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch)); + com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch)); + new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true); + new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true); + when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2); + ChannelSftp channel1 = spy(new ChannelSftp()); + ChannelSftp channel2 = spy(new ChannelSftp()); + new DirectFieldAccessor(channel1).setPropertyValue("session", jschSession1); + new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); + when(jschSession1.openChannel("sftp")).thenReturn(channel1); + when(jschSession2.openChannel("sftp")).thenReturn(channel2); + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, false); + factory.setHost("host"); + factory.setUser("foo"); + factory.setPassword("bar"); + noopConnect(channel1); + noopConnect(channel2); + Session s1 = factory.getSession(); + Session s2 = factory.getSession(); + assertNotSame(TestUtils.getPropertyValue(s1, "jschSession"), TestUtils.getPropertyValue(s2, "jschSession")); + } + + @Test + public void testSharedSessionCachedReset() throws Exception { + JSch jsch = spy(new JSch()); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class); + ctor.setAccessible(true); + com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch)); + com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch)); + new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true); + new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true); + when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2); + ChannelSftp channel1 = spy(new ChannelSftp()); + ChannelSftp channel2 = spy(new ChannelSftp()); + ChannelSftp channel3 = spy(new ChannelSftp()); + ChannelSftp channel4 = spy(new ChannelSftp()); + new DirectFieldAccessor(channel1).setPropertyValue("session", jschSession1); + new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); + when(jschSession1.openChannel("sftp")).thenReturn(channel1, channel2); + when(jschSession2.openChannel("sftp")).thenReturn(channel3, channel4); + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, true); + factory.setHost("host"); + factory.setUser("foo"); + factory.setPassword("bar"); + CachingSessionFactory cachedFactory = new CachingSessionFactory(factory); + noopConnect(channel1); + noopConnect(channel2); + noopConnect(channel3); + noopConnect(channel4); + Session s1 = cachedFactory.getSession(); + Session s2 = cachedFactory.getSession(); + assertSame(jschSession1, TestUtils.getPropertyValue(s2, "targetSession.jschSession")); + assertSame(TestUtils.getPropertyValue(s1, "targetSession.jschSession"), TestUtils.getPropertyValue(s2, "targetSession.jschSession")); + s1.close(); + Session s3 = cachedFactory.getSession(); + assertSame(TestUtils.getPropertyValue(s1, "targetSession"), TestUtils.getPropertyValue(s3, "targetSession")); + s3.close(); + cachedFactory.resetCache(); + verify(jschSession1, never()).disconnect(); + s3 = cachedFactory.getSession(); + assertSame(jschSession2, TestUtils.getPropertyValue(s3, "targetSession.jschSession")); + assertNotSame(TestUtils.getPropertyValue(s1, "targetSession"), TestUtils.getPropertyValue(s3, "targetSession")); + s2.close(); + verify(jschSession1).disconnect(); + s2 = cachedFactory.getSession(); + assertSame(jschSession2, TestUtils.getPropertyValue(s2, "targetSession.jschSession")); + assertNotSame(TestUtils.getPropertyValue(s3, "targetSession"), TestUtils.getPropertyValue(s2, "targetSession")); + s2.close(); + s3.close(); + verify(jschSession2, never()).disconnect(); + cachedFactory.resetCache(); + verify(jschSession2).disconnect(); + } + + private void noopConnect(ChannelSftp channel1) throws JSchException { + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + return null; + } + }).when(channel1).connect(); + } + public static class TestSftpSessionFactory extends DefaultSftpSessionFactory { @Override @@ -214,6 +341,7 @@ public class SftpOutboundTests { ChannelSftp channel = mock(ChannelSftp.class); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { File file = new File((String)invocation.getArguments()[1]); @@ -225,6 +353,7 @@ public class SftpOutboundTests { }).when(channel).put(Mockito.any(InputStream.class), Mockito.anyString()); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { File file = new File((String) invocation.getArguments()[0]); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml index da0b9d7d55..cf054cdccd 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml @@ -77,4 +77,19 @@ + + + + + + + + + + + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java index 3e5d978499..613b191496 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.sftp.outbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -27,7 +28,12 @@ import static org.mockito.Mockito.when; import java.io.ByteArrayOutputStream; import java.io.File; import java.io.IOException; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.hamcrest.Matchers; import org.junit.After; @@ -44,6 +50,7 @@ import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.sftp.session.SftpFileInfo; +import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.FileCopyUtils; @@ -160,6 +167,11 @@ public class SftpServerOutboundTests { @Test public void testInt2866LocalDirectoryExpressionGET() { + Session session = null; + boolean sharedSession = "realSSHSharedSession".equals(System.getProperty("spring.profiles.active")); + if (sharedSession) { + session = this.sessionFactory.getSession(); + } String dir = "sftpSource/"; this.inboundGet.send(new GenericMessage(dir + "sftpSource1.txt")); Message result = this.output.receive(1000); @@ -175,6 +187,11 @@ public class SftpServerOutboundTests { localFile = (File) result.getPayload(); assertThat(localFile.getPath().replaceAll(java.util.regex.Matcher.quoteReplacement(File.separator), "/"), Matchers.containsString(dir.toUpperCase())); + if (sharedSession) { + Session session2 = this.sessionFactory.getSession(); + assertSame(TestUtils.getPropertyValue(session, "targetSession.jschSession"), + TestUtils.getPropertyValue(session2, "targetSession.jschSession")); + } } @Test @@ -276,5 +293,65 @@ public class SftpServerOutboundTests { } } + @Test + public void testInt3047ConcurrentSharedSession() throws Exception { + if ("realSSHSharedSession".equals(System.getProperty("spring.profiles.active"))) { + final Session session1 = this.sessionFactory.getSession(); + final Session session2 = this.sessionFactory.getSession(); + final PipedInputStream pipe1 = new PipedInputStream(); + PipedOutputStream out1 = new PipedOutputStream(pipe1); + final PipedInputStream pipe2 = new PipedInputStream(); + PipedOutputStream out2 = new PipedOutputStream(pipe2); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + try { + session1.write(pipe1, "foo.txt"); + } + catch (IOException e) { + e.printStackTrace(); + } + latch1.countDown(); + } + }); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + try { + session2.write(pipe2, "bar.txt"); + } + catch (IOException e) { + e.printStackTrace(); + } + latch2.countDown(); + } + }); + + out1.write('a'); + out2.write('b'); + out1.write('c'); + out2.write('d'); + out1.write('e'); + out2.write('f'); + out1.close(); + out2.close(); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + ByteArrayOutputStream bos1 = new ByteArrayOutputStream(); + ByteArrayOutputStream bos2 = new ByteArrayOutputStream(); + session1.read("foo.txt", bos1); + session2.read("bar.txt", bos2); + assertEquals("ace", new String(bos1.toByteArray())); + assertEquals("bdf", new String(bos2.toByteArray())); + session1.remove("foo.txt"); + session2.remove("bar.txt"); + session1.close(); + session2.close(); + } + } } diff --git a/src/reference/docbook/ftp.xml b/src/reference/docbook/ftp.xml index 1e04a9960f..42747dd15d 100644 --- a/src/reference/docbook/ftp.xml +++ b/src/reference/docbook/ftp.xml @@ -154,6 +154,7 @@ protected void postProcessClientBeforeConnect(T client) throws IOException { filename-pattern="*.txt" remote-directory="some/remote/path" remote-file-separator="/" + preserve-timestamp="true" local-filename-generator-expression="#this.toUpperCase() + '.a'" local-filter="myFilter" local-directory="."> @@ -172,7 +173,11 @@ protected void postProcessClientBeforeConnect(T client) throws IOException { that's what it ultimately generates with the transferred file as its payload. So, the root object of the SpEL Evaluation Context is the original name of the remote file (String). - + + Starting with Spring Integration 3.0, you can specify the preserve-timestamp + attribute (default false); when true, the local file's modified timestamp will be set to the value + retrieved from the server; otherwise it will be set to the current time. + Sometimes file filtering based on the simple pattern specified via filename-pattern attribute might not be sufficient. If this is the case, you can use the filename-regex attribute to specify a Regular Expression @@ -500,8 +505,8 @@ protected void postProcessClientBeforeConnect(T client) throws IOException {
FTP Session Caching - Starting with version 3.0, sessions are no longer cached by default; the cache-sessions attribute - is no longer supported on endpoints. You must now use a CachingSessionFactory (see below) if you + Starting with Spring Integration version 3.0, sessions are no longer cached by default; the cache-sessions attribute + is no longer supported on endpoints. You must use a CachingSessionFactory (see below) if you wish to cache sessions. @@ -532,5 +537,10 @@ protected void postProcessClientBeforeConnect(T client) throws IOException { sessionCacheSize set to 10 and the sessionWaitTimeout set to 1 second (its value is in millliseconds). + + Starting with Spring Integration version 3.0, the CachingConnectionFactory + provides a resetCache() method. When invoked, all idle sessions are immediately closed and in-use + sessions are closed when they are returned to the cache. New requests for sessions will establish new sessions as necessary. +
diff --git a/src/reference/docbook/sftp.xml b/src/reference/docbook/sftp.xml index a73cf9c6d6..5111c75d5f 100644 --- a/src/reference/docbook/sftp.xml +++ b/src/reference/docbook/sftp.xml @@ -59,6 +59,27 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp However, Spring Integration also supports the caching of SFTP sessions, please see for more information. + + + JSch supports multiple channels (operations) over a connection to the server. By default, + the Spring Integration session factory uses a separate physical connection for each channel. + Since Spring Integration 3.0, you can configure the session factory + (using a boolean constructor arg - default false) to use a single connection + to the server and create multiple JSch channels on that single connection. + + + When using this feature, you must wrap the session factory in a caching session + factory, as described below, so that the connection is not physically closed when + an operation completes. + + + If the cache is reset, the session is disconnected only when the last channel is closed. + + + The connection will be refreshed if it is found to be disconnected when a new operation + obtains a session. + + If you experience connectivity problems and would like to trace Session creation as well as see which Sessions are polled you may enable it by @@ -79,6 +100,11 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp Below you will find all properties that are exposed by the DefaultSftpSessionFactory. + isSharedSession (constructor argument) + + When true, a single connection will be used and JSch Channels will be multiplexed. + Defaults to false. + clientVersion Allows you to set the client version property. It's default @@ -183,8 +209,8 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
SFTP Session Caching - Starting with version 3.0, sessions are no longer cached by default; the cache-sessions attribute - is no longer supported on endpoints. You must now use a CachingSessionFactory (see below) if you + Starting with Spring Integration version 3.0, sessions are no longer cached by default; the cache-sessions attribute + is no longer supported on endpoints. You must use a CachingSessionFactory (see below) if you wish to cache sessions. @@ -217,6 +243,13 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp sessionCacheSize set to 10 and the sessionWaitTimeout set to 1 second (its value is in millliseconds). + + Starting with Spring Integration version 3.0, the CachingConnectionFactory + provides a resetCache() method. When invoked, all idle sessions are immediately closed and in-use + sessions are closed when they are returned to the cache. When using isSharedSession=true, the channel is + closed, and the shared session is closed only when the last channel is closed. + New requests for sessions will establish new sessions as necessary. +
@@ -230,6 +263,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp channel="requestChannel" filename-pattern="*.txt" remote-directory="/foo/bar" + preserve-timestamp="true" local-directory="file:target/foo" auto-create-local-directory="true" local-filename-generator-expression="#this.toUpperCase() + '.a'" @@ -252,6 +286,11 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp that's what it ultimately generates with the transferred file as its payload. So, the root object of the SpEL Evaluation Context is the original name of the remote file (String). + + Starting with Spring Integration 3.0, you can specify the preserve-timestamp + attribute (default false); when true, the local file's modified timestamp will be set to the value + retrieved from the server; otherwise it will be set to the current time. + Sometimes file filtering based on the simple pattern specified via filename-pattern attribute might not be sufficient. If this is the case, you can use the filename-regex attribute to specify a Regular Expression diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 95186b1deb..c1271ca497 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -243,6 +243,16 @@ For more information, see and . + + The CachingConnectionFactory now provides a new method + resetCache(). This immediately closes idle sessions and causes in-use + sessions to be closed as and when they are returned to the cache. + + + The DefaultSftpSessionFactory (in conjunction with a + CachingSessionFactory) now supports multiplexing channels over + a single SSH connection (SFTP Only). +
FTP, SFTP and FTPS Inbound Adapters