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.

spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java

@@ -192,7 +192,7 @@ public class SimplePool<T> implements Pool<T> {
 		}
 		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<T> implements Pool<T> {
 	}
 
 	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);

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java

@@ -45,6 +45,10 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 
 	private final SimplePool<Session<F>> 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<F> implements SessionFactory<F>, DisposableBe
 				session.close();
 			}
 		});
+		this.isSharedSessionCapable = sessionFactory instanceof SharedSessionCapable;
 	}
 
 
@@ -105,7 +110,7 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 	 */
 	@Override
 	public Session<F> getSession() {
-		return new CachedSession(this.pool.getItem());
+		return new CachedSession(this.pool.getItem(), this.sharedSessionEpoch);
 	}
 
 	/**
@@ -116,27 +121,64 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, 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<F> {
 
 		private final Session<F> targetSession;
 
-		private boolean released;
+		private volatile boolean released;
 
-		private CachedSession(Session<F> targetSession) {
+		/**
+		 * The epoch in which this session was created.
+		 */
+		private final long sharedSessionEpoch;
+
+		private CachedSession(Session<F> 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;

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();
+
+}

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<String> cache = new CachingSessionFactory<String>(factory);
+		Session<String> sess1 = cache.getSession();
+		assertEquals("session:1", TestUtils.getPropertyValue(sess1, "targetSession.id"));
+		Session<String> 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<String> {
+
+		private int n;
+
+		@Override
+		public Session<String> getSession() {
+			return new TestSession("session:" + ++n);
+		}
+
+	}
+
+	private class TestSession implements Session<String> {
+
+		@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;
+		}
+
+	}
+
+}

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<LsEntry> {
+public class DefaultSftpSessionFactory implements SessionFactory<LsEntry>, SharedSessionCapable {
 
 	private volatile String host;
 
@@ -76,9 +80,35 @@ public class DefaultSftpSessionFactory implements SessionFactory<LsEntry> {
 
 	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<LsEntry> {
 		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<LsEntry> {
 		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<LsEntry> {
 		}
 	}
 
+	/**
+	 * 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();
+		}
+
+	}
 }

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<LsEntry> {
 
 	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<LsEntry> {
 
 	@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<LsEntry> {
 		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<LsEntry> {
 			throw new IllegalStateException("failed to connect", e);
 		}
 	}
+
 }

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<String> madeDirs = new ArrayList<String>();
 		doAnswer(new Answer<Object>() {
+			@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<com.jcraft.jsch.Session> 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<LsEntry> s1 = factory.getSession();
+		Session<LsEntry> s2 = factory.getSession();
+		assertSame(TestUtils.getPropertyValue(s1, "jschSession"), TestUtils.getPropertyValue(s2, "jschSession"));
+	}
+
+	@Test
+	public void testNotSharedSession() throws Exception {
+		JSch jsch = spy(new JSch());
+		Constructor<com.jcraft.jsch.Session> 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<LsEntry> s1 = factory.getSession();
+		Session<LsEntry> s2 = factory.getSession();
+		assertNotSame(TestUtils.getPropertyValue(s1, "jschSession"), TestUtils.getPropertyValue(s2, "jschSession"));
+	}
+
+	@Test
+	public void testSharedSessionCachedReset() throws Exception {
+		JSch jsch = spy(new JSch());
+		Constructor<com.jcraft.jsch.Session> 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<LsEntry> cachedFactory = new CachingSessionFactory<LsEntry>(factory);
+		noopConnect(channel1);
+		noopConnect(channel2);
+		noopConnect(channel3);
+		noopConnect(channel4);
+		Session<LsEntry> s1 = cachedFactory.getSession();
+		Session<LsEntry> s2 = cachedFactory.getSession();
+		assertSame(jschSession1, TestUtils.getPropertyValue(s2, "targetSession.jschSession"));
+		assertSame(TestUtils.getPropertyValue(s1, "targetSession.jschSession"), TestUtils.getPropertyValue(s2, "targetSession.jschSession"));
+		s1.close();
+		Session<LsEntry> 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<Object>() {
+
+			@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<Object>() {
+					@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<Object>() {
+					@Override
 					public Object answer(InvocationOnMock invocation)
 							throws Throwable {
 						File file = new File((String) invocation.getArguments()[0]);

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml

@@ -77,4 +77,19 @@
 		</bean>
 	</beans>
 
+	<beans profile="realSSHSharedSession">
+		<bean id="ftpSessionFactory" 
+				class="org.springframework.integration.file.remote.session.CachingSessionFactory">
+			<constructor-arg>
+				<bean
+			  		class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
+					<constructor-arg value="true"/>
+					<property name="host" value="localhost"/>
+					<property name="user" value="ftptest"/>
+					<property name="password" value="ftptest"/>
+				</bean>
+			</constructor-arg>
+		</bean>
+	</beans>
+
 </beans>

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<Object>(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();
+		}
+	}
 
 }

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).
 	</para>
-
+	<para>
+	Starting with <emphasis>Spring Integration 3.0</emphasis>, you can specify the <code>preserve-timestamp</code>
+	attribute (default <code>false</code>); when <code>true</code>, 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.
+	</para>
   	<para>
 	Sometimes file filtering based on the simple pattern specified via <code>filename-pattern</code> attribute might not be
 	sufficient. If this is the case, you can use the <code>filename-regex</code> attribute to specify a Regular Expression
@@ -500,8 +505,8 @@ protected void postProcessClientBeforeConnect(T client) throws IOException {
   <section id="ftp-session-caching">
 	<title>FTP Session Caching</title>
 	<important>
-		Starting with version 3.0, sessions are no longer cached by default; the <code>cache-sessions</code> attribute
-		is no longer supported on endpoints. You must now use a <classname>CachingSessionFactory</classname> (see below) if you
+		Starting with <emphasis>Spring Integration version 3.0</emphasis>, sessions are no longer cached by default; the <code>cache-sessions</code> attribute
+		is no longer supported on endpoints. You must use a <classname>CachingSessionFactory</classname> (see below) if you
 		wish to cache sessions.
 	</important>
 	<para>
@@ -532,5 +537,10 @@ protected void postProcessClientBeforeConnect(T client) throws IOException {
 	<code>sessionCacheSize</code> set to 10 and the <code>sessionWaitTimeout</code> set to 1 second (its value is in millliseconds).
 
 	</para>
+	<para>
+		Starting with <emphasis>Spring Integration version 3.0</emphasis>, the <classname>CachingConnectionFactory</classname>
+		provides a <code>resetCache()</code> 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.
+	</para>
   </section>
 </chapter>

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 <xref linkend="sftp-session-caching"/> for more information.
 	</para>
+	<important>
+		<para>
+			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 <emphasis>Spring Integration 3.0</emphasis>, you can configure the session factory
+			(using a boolean constructor arg - default <code>false</code>) to use a single connection
+			to the server and create multiple <code>JSch</code> channels on that single connection.
+		</para>
+		<para>
+			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.
+		</para>
+		<para>
+			If the cache is reset, the session is disconnected only when the last channel is closed.
+		</para>
+		<para>
+			The connection will be refreshed if it is found to be disconnected when a new operation
+			obtains a session.
+		</para>
+	</important>
 	<note>
 		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
 				<classname><ulink url="http://static.springsource.org/spring-integration/api/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.html">DefaultSftpSessionFactory</ulink></classname>.
 			</para>
+			<para><emphasis role="bold">isSharedSession (constructor argument)</emphasis></para>
+			<para>
+				When true, a single connection will be used and <code>JSch Channels</code> will be multiplexed.
+				Defaults to false.
+			</para>
 			<para><emphasis role="bold">clientVersion</emphasis></para>
 			<para>
 				Allows you to set the client version property. It's default
@@ -183,8 +209,8 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
   <section id="sftp-session-caching">
   	<title>SFTP Session Caching</title>
 	<important>
-		Starting with version 3.0, sessions are no longer cached by default; the <code>cache-sessions</code> attribute
-		is no longer supported on endpoints. You must now use a <classname>CachingSessionFactory</classname> (see below) if you
+		Starting with <emphasis>Spring Integration version 3.0</emphasis>, sessions are no longer cached by default; the <code>cache-sessions</code> attribute
+		is no longer supported on endpoints. You must use a <classname>CachingSessionFactory</classname> (see below) if you
 		wish to cache sessions.
 	</important>
 	<para>
@@ -217,6 +243,13 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
 	<code>sessionCacheSize</code> set to 10 and the <code>sessionWaitTimeout</code> set to 1 second (its value is in millliseconds).
 
   	</para>
+	<para>
+		Starting with <emphasis>Spring Integration version 3.0</emphasis>, the <classname>CachingConnectionFactory</classname>
+		provides a <code>resetCache()</code> method. When invoked, all idle sessions are immediately closed and in-use
+		sessions are closed when they are returned to the cache. When using <code>isSharedSession=true</code>, 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.
+  	</para>
   </section>
 
    <section id="sftp-inbound">
@@ -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).
 	</para>
+	<para>
+	Starting with <emphasis>Spring Integration 3.0</emphasis>, you can specify the <code>preserve-timestamp</code>
+	attribute (default <code>false</code>); when <code>true</code>, 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.
+	</para>
   	<para>
 	Sometimes file filtering based on the simple pattern specified via <code>filename-pattern</code> attribute might not be
 	sufficient. If this is the case, you can use the <code>filename-regex</code> attribute to specify a Regular Expression

src/reference/docbook/whats-new.xml

@@ -243,6 +243,16 @@
 			    For more information, see
 			    <xref linkend="ftp-session-caching"/> and <xref linkend="sftp-session-caching"/>.
 			</para>
+			<para>
+				The <classname>CachingConnectionFactory</classname> now provides a new method
+				<code>resetCache()</code>. This immediately closes idle sessions and causes in-use
+				sessions to be closed as and when they are returned to the cache.
+			</para>
+			<para>
+				The <classname>DefaultSftpSessionFactory</classname> (in conjunction with a
+				<classname>CachingSessionFactory</classname>) now supports multiplexing channels over
+				a single SSH connection (SFTP Only).
+			</para>
 		</section>
 		<section id="3.0-xFTP-ib">
 			<title>FTP, SFTP and FTPS Inbound Adapters</title>