Merge remote-tracking branch 'upstream/master' into 4.0.0-WIP

Conflicts:
	spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java
	spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java
	spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java
	spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java

Resolved.

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.
@@ -65,18 +69,22 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 	public CachingSessionFactory(SessionFactory<F> sessionFactory, int sessionCacheSize) {
 		this.sessionFactory = sessionFactory;
 		this.pool = new SimplePool<Session<F>>(sessionCacheSize, new SimplePool.PoolItemCallback<Session<F>>() {
+			@Override
 			public Session<F> createForPool() {
 				return CachingSessionFactory.this.sessionFactory.getSession();
 			}
 
+			@Override
 			public boolean isStale(Session<F> session) {
 				return !session.isOpen();
 			}
 
+			@Override
 			public void removedFromPool(Session<F> session) {
 				session.close();
 			}
 		});
+		this.isSharedSessionCapable = sessionFactory instanceof SharedSessionCapable;
 	}
 
 
@@ -100,78 +108,138 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 	/**
 	 * Get a session from the pool (or block if none available).
 	 */
+	@Override
 	public Session<F> getSession() {
-		return new CachedSession(this.pool.getItem());
+		return new CachedSession(this.pool.getItem(), this.sharedSessionEpoch);
 	}
 
 	/**
 	 * Remove (close) any unused sessions in the pool.
 	 */
+	@Override
 	public void destroy() {
 		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;
 			}
 		}
 
+		@Override
 		public boolean remove(String path) throws IOException{
 			return this.targetSession.remove(path);
 		}
 
+		@Override
 		public F[] list(String path) throws IOException{
 			return this.targetSession.list(path);
 		}
 
+		@Override
 		public void read(String source, OutputStream os) throws IOException{
 			this.targetSession.read(source, os);
 		}
 
+		@Override
 		public void write(InputStream inputStream, String destination) throws IOException{
 			this.targetSession.write(inputStream, destination);
 		}
 
+		@Override
 		public boolean isOpen() {
 			return this.targetSession.isOpen();
 		}
 
+		@Override
 		public void rename(String pathFrom, String pathTo) throws IOException {
 			this.targetSession.rename(pathFrom, pathTo);
 		}
 
+		@Override
 		public boolean mkdir(String directory) throws IOException {
 			return this.targetSession.mkdir(directory);
 		}
 
+		@Override
 		public boolean exists(String path) throws IOException{
 			return this.targetSession.exists(path);
 		}
 
+		@Override
 		public String[] listNames(String path) throws IOException {
 			return this.targetSession.listNames(path);
 		}
+
+		@Override
+		public InputStream readRaw(String source) throws IOException {
+			return this.targetSession.readRaw(source);
+		}
+
+		@Override
+		public boolean finalizeRaw() throws IOException {
+			return this.targetSession.finalizeRaw();
+		}
+
 	}
 
 }

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.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.
@@ -35,21 +35,36 @@ public interface Session<T> {
 	boolean remove(String path) throws IOException;
 
 	T[] list(String path) throws IOException;
-	
+
 	void read(String source, OutputStream outputStream) throws IOException;
 
 	void write(InputStream inputStream, String destination) throws IOException;
-	
+
 	boolean mkdir(String directory) throws IOException;
-	
+
 	void rename(String pathFrom, String pathTo) throws IOException;
 
 	void close();
-	
+
 	boolean isOpen();
-	
+
 	boolean exists(String path) throws IOException;
 
 	String[] listNames(String path) throws IOException;
 
+	/**
+	 * Retrieve a remote file as a raw {@link InputStream}.
+	 * @param source The path of the remote file
+	 * @return The raw inputStream.
+	 */
+	InputStream readRaw(String source) throws IOException;
+
+	/**
+	 * Invoke after closing the InputStream from {@link #readRaw(String)}.
+	 * Required by some session providers.
+	 * @return true if successful.
+	 * @throws IOException
+	 */
+	boolean finalizeRaw() throws IOException;
+
 }

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/gateway/RemoteFileOutboundGatewayTests.java

@@ -55,7 +55,6 @@ import org.springframework.messaging.MessagingException;
 /**
  * @author Gary Russell
  * @since 2.1
- *
  */
 @SuppressWarnings("rawtypes")
 public class RemoteFileOutboundGatewayTests {
@@ -63,11 +62,11 @@ public class RemoteFileOutboundGatewayTests {
 	private final String tmpDir = System.getProperty("java.io.tmpdir");
 
 
-	@Test(expected=IllegalArgumentException.class)
+	@Test(expected = IllegalArgumentException.class)
 	public void testBad() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "bad", "payload");
+				(sessionFactory, "bad", "payload");
 		gw.afterPropertiesSet();
 	}
 
@@ -106,7 +105,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
 		TestLsEntry[] files = fileList();
@@ -128,6 +127,7 @@ public class RemoteFileOutboundGatewayTests {
 
 	/**
 	 * Test a wildcard mget where the full path is returned for each file
+	 *
 	 * @throws Exception
 	 */
 	@Test
@@ -138,19 +138,25 @@ public class RemoteFileOutboundGatewayTests {
 	private void testMGetWildGuts(final String path1, final String path2) {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "mget", "payload");
-		gw.setLocalDirectory(new File(this.tmpDir ));
+				(sessionFactory, "mget", "payload");
+		gw.setLocalDirectory(new File(this.tmpDir));
 		gw.afterPropertiesSet();
 		new File(this.tmpDir + "/f1").delete();
 		new File(this.tmpDir + "/f2").delete();
 		when(sessionFactory.getSession()).thenReturn(new Session() {
 			int n;
+
+			@Override
 			public boolean remove(String path) throws IOException {
 				return false;
 			}
+
+			@Override
 			public Object[] list(String path) throws IOException {
 				return null;
 			}
+
+			@Override
 			public void read(String source, OutputStream outputStream)
 					throws IOException {
 				if (n++ == 0) {
@@ -161,25 +167,49 @@ public class RemoteFileOutboundGatewayTests {
 				}
 				outputStream.write("testData".getBytes());
 			}
+
+			@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() {
 			}
+
+			@Override
 			public boolean isOpen() {
 				return false;
 			}
+
+			@Override
 			public boolean exists(String path) throws IOException {
 				return false;
 			}
+
+			@Override
 			public String[] listNames(String path) throws IOException {
-				return new String[] {path1, path2};
+				return new String[]{path1, path2};
+			}
+
+			@Override
+			public InputStream readRaw(String source) throws IOException {
+				return null;
+			}
+
+			@Override
+			public boolean finalizeRaw() throws IOException {
+				return false;
 			}
 		});
 		@SuppressWarnings("unchecked")
@@ -196,40 +226,69 @@ public class RemoteFileOutboundGatewayTests {
 	public void testMGetSingle() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "mget", "payload");
-		gw.setLocalDirectory(new File(this.tmpDir ));
+				(sessionFactory, "mget", "payload");
+		gw.setLocalDirectory(new File(this.tmpDir));
 		gw.afterPropertiesSet();
 		new File(this.tmpDir + "/f1").delete();
 		when(sessionFactory.getSession()).thenReturn(new Session() {
+			@Override
 			public boolean remove(String path) throws IOException {
 				return false;
 			}
+
+			@Override
 			public Object[] list(String path) throws IOException {
 				return null;
 			}
+
+			@Override
 			public void read(String source, OutputStream outputStream)
 					throws IOException {
 				outputStream.write("testData".getBytes());
 			}
+
+			@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() {
 			}
+
+			@Override
 			public boolean isOpen() {
 				return false;
 			}
+
+			@Override
 			public boolean exists(String path) throws IOException {
 				return false;
 			}
+
+			@Override
 			public String[] listNames(String path) throws IOException {
-				return new String[] {"f1"};
+				return new String[]{"f1"};
+			}
+
+			@Override
+			public InputStream readRaw(String source) throws IOException {
+				return null;
+			}
+
+			@Override
+			public boolean finalizeRaw() throws IOException {
+				return false;
 			}
 		});
 		@SuppressWarnings("unchecked")
@@ -241,47 +300,76 @@ public class RemoteFileOutboundGatewayTests {
 				out.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
 	}
 
-	@Test(expected=MessagingException.class)
+	@Test(expected = MessagingException.class)
 	public void testMGetEmpty() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "mget", "payload");
-		gw.setLocalDirectory(new File(this.tmpDir ));
+				(sessionFactory, "mget", "payload");
+		gw.setLocalDirectory(new File(this.tmpDir));
 		gw.setOptions("   -x   ");
 		gw.afterPropertiesSet();
 		new File(this.tmpDir + "/f1").delete();
 		new File(this.tmpDir + "/f2").delete();
 		when(sessionFactory.getSession()).thenReturn(new Session() {
+			@Override
 			public boolean remove(String path) throws IOException {
 				return false;
 			}
+
+			@Override
 			public Object[] list(String path) throws IOException {
 				return null;
 			}
+
+			@Override
 			public void read(String source, OutputStream outputStream)
 					throws IOException {
 				outputStream.write("testData".getBytes());
 			}
+
+			@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() {
 			}
+
+			@Override
 			public boolean isOpen() {
 				return false;
 			}
+
+			@Override
 			public boolean exists(String path) throws IOException {
 				return false;
 			}
+
+			@Override
 			public String[] listNames(String path) throws IOException {
 				return new String[0];
 			}
+
+			@Override
+			public InputStream readRaw(String source) throws IOException {
+				return null;
+			}
+
+			@Override
+			public boolean finalizeRaw() throws IOException {
+				return false;
+			}
 		});
 		gw.handleRequestMessage(new GenericMessage<String>("testremote/*"));
 	}
@@ -290,7 +378,7 @@ public class RemoteFileOutboundGatewayTests {
 	public void testMove() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "mv", "payload");
+				(sessionFactory, "mv", "payload");
 		gw.afterPropertiesSet();
 		Session<?> session = mock(Session.class);
 		final AtomicReference<String> args = new AtomicReference<String>();
@@ -302,7 +390,7 @@ public class RemoteFileOutboundGatewayTests {
 				return null;
 			}
 		}).when(session).rename(anyString(), anyString());
-		when (sessionFactory.getSession()).thenReturn(session);
+		when(sessionFactory.getSession()).thenReturn(session);
 		Message<String> requestMessage = MessageBuilder.withPayload("foo")
 				.setHeader(FileHeaders.RENAME_TO, "bar")
 				.build();
@@ -316,7 +404,7 @@ public class RemoteFileOutboundGatewayTests {
 	public void testMoveWithExpression() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "mv", "payload");
+				(sessionFactory, "mv", "payload");
 		gw.setRenameExpression("payload.substring(1)");
 		gw.afterPropertiesSet();
 		Session<?> session = mock(Session.class);
@@ -329,7 +417,7 @@ public class RemoteFileOutboundGatewayTests {
 				return null;
 			}
 		}).when(session).rename(anyString(), anyString());
-		when (sessionFactory.getSession()).thenReturn(session);
+		when(sessionFactory.getSession()).thenReturn(session);
 		Message<?> out = (Message<?>) gw.handleRequestMessage(new GenericMessage<String>("foo"));
 		assertEquals("oo", out.getHeaders().get(FileHeaders.RENAME_TO));
 		assertEquals("foo", out.getHeaders().get(FileHeaders.REMOTE_FILE));
@@ -341,7 +429,7 @@ public class RemoteFileOutboundGatewayTests {
 	public void testMoveWithMkDirs() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "mv", "payload");
+				(sessionFactory, "mv", "payload");
 		gw.setRenameExpression("'foo/bar/baz'");
 		gw.afterPropertiesSet();
 		Session<?> session = mock(Session.class);
@@ -356,12 +444,13 @@ public class RemoteFileOutboundGatewayTests {
 		}).when(session).rename(anyString(), anyString());
 		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;
 			}
 		}).when(session).mkdir(anyString());
-		when (sessionFactory.getSession()).thenReturn(session);
+		when(sessionFactory.getSession()).thenReturn(session);
 		Message<String> requestMessage = MessageBuilder.withPayload("foo")
 				.setHeader(FileHeaders.RENAME_TO, "bar")
 				.build();
@@ -390,7 +479,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.setOptions("-f");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
@@ -407,23 +496,23 @@ public class RemoteFileOutboundGatewayTests {
 	}
 
 	public TestLsEntry[] level1List() {
-		return new TestLsEntry[] {
-			new TestLsEntry("f1", 123, false, false, 1234, "-r--r--r--"),
-			new TestLsEntry("d1", 0, true, false, 12345, "drw-r--r--"),
-			new TestLsEntry("f2", 12345, false, false, 123456, "-rw-r--r--")
+		return new TestLsEntry[]{
+				new TestLsEntry("f1", 123, false, false, 1234, "-r--r--r--"),
+				new TestLsEntry("d1", 0, true, false, 12345, "drw-r--r--"),
+				new TestLsEntry("f2", 12345, false, false, 123456, "-rw-r--r--")
 		};
 	}
 
 	public TestLsEntry[] level2List() {
-		return new TestLsEntry[] {
-			new TestLsEntry("d2", 0, true, false, 12345, "drw-r--r--"),
-			new TestLsEntry("f3", 12345, false, false, 123456, "-rw-r--r--")
+		return new TestLsEntry[]{
+				new TestLsEntry("d2", 0, true, false, 12345, "drw-r--r--"),
+				new TestLsEntry("f3", 12345, false, false, 123456, "-rw-r--r--")
 		};
 	}
 
 	public TestLsEntry[] level3List() {
-		return new TestLsEntry[] {
-			new TestLsEntry("f4", 12345, false, false, 123456, "-rw-r--r--")
+		return new TestLsEntry[]{
+				new TestLsEntry("f4", 12345, false, false, 123456, "-rw-r--r--")
 		};
 	}
 
@@ -432,7 +521,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.setOptions("-f -R");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
@@ -459,7 +548,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.setOptions("-f -R -dirs");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
@@ -488,7 +577,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
 		TestLsEntry[] files = new TestLsEntry[0];
@@ -504,7 +593,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.setOptions("-1");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
@@ -523,7 +612,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.setOptions("-1 -f");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
@@ -542,7 +631,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.setOptions("-1 -dirs");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
@@ -562,7 +651,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.setOptions("-1 -dirs -links");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
@@ -583,7 +672,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.setOptions("-1 -a -f -dirs -links");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
@@ -606,7 +695,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "ls", "payload");
+				(sessionFactory, "ls", "payload");
 		gw.setOptions("-1 -a -f -dirs -links");
 		gw.setFilter(new TestPatternFilter("*4"));
 		gw.afterPropertiesSet();
@@ -624,45 +713,75 @@ public class RemoteFileOutboundGatewayTests {
 	public void testGet() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "get", "payload");
-		gw.setLocalDirectory(new File(this.tmpDir ));
+				(sessionFactory, "get", "payload");
+		gw.setLocalDirectory(new File(this.tmpDir));
 		gw.afterPropertiesSet();
 		new File(this.tmpDir + "/f1").delete();
-		when(sessionFactory.getSession()).thenReturn(new Session(){
+		when(sessionFactory.getSession()).thenReturn(new Session() {
 			private boolean open = true;
+
+			@Override
 			public boolean remove(String path) throws IOException {
 				return false;
 			}
+
+			@Override
 			public TestLsEntry[] list(String path) throws IOException {
-				return new TestLsEntry[] {
+				return new TestLsEntry[]{
 						new TestLsEntry("f1", 1234, false, false, 12345, "-rw-r--r--")
 				};
 			}
+
+			@Override
 			public void read(String source, OutputStream outputStream)
 					throws IOException {
 				outputStream.write("testfile".getBytes());
 			}
+
+			@Override
 			public void write(InputStream inputStream, String destination)
 					throws IOException {
 			}
+
+			@Override
 			public boolean mkdir(String directory) throws IOException {
 				return true;
 			}
+
+			@Override
 			public void rename(String pathFrom, String pathTo)
 					throws IOException {
 			}
+
+			@Override
 			public void close() {
 				open = false;
 			}
+
+			@Override
 			public boolean isOpen() {
 				return open;
 			}
+
+			@Override
 			public boolean exists(String path) throws IOException {
 				return true;
 			}
+
+			@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;
+			}
 		});
 		@SuppressWarnings("unchecked")
 		Message<File> out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
@@ -680,7 +799,7 @@ public class RemoteFileOutboundGatewayTests {
 	public void testGet_P() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "get", "payload");
+				(sessionFactory, "get", "payload");
 		gw.setLocalDirectory(new File(this.tmpDir));
 		gw.setOptions("-P");
 		gw.afterPropertiesSet();
@@ -688,41 +807,71 @@ public class RemoteFileOutboundGatewayTests {
 		Calendar cal = Calendar.getInstance();
 		cal.add(Calendar.MONTH, -1);
 		final Date modified = new Date(cal.getTime().getTime() / 1000 * 1000);
-		when(sessionFactory.getSession()).thenReturn(new Session(){
+		when(sessionFactory.getSession()).thenReturn(new Session() {
 			private boolean open = true;
+
+			@Override
 			public boolean remove(String path) throws IOException {
 				return false;
 			}
+
+			@Override
 			public TestLsEntry[] list(String path) throws IOException {
-				return new TestLsEntry[] {
+				return new TestLsEntry[]{
 						new TestLsEntry("f1", 1234, false, false, modified.getTime(), "-rw-r--r--")
 				};
 			}
+
+			@Override
 			public void read(String source, OutputStream outputStream)
 					throws IOException {
 				outputStream.write("testfile".getBytes());
 			}
+
+			@Override
 			public void write(InputStream inputStream, String destination)
 					throws IOException {
 			}
+
+			@Override
 			public boolean mkdir(String directory) throws IOException {
 				return true;
 			}
+
+			@Override
 			public void rename(String pathFrom, String pathTo)
 					throws IOException {
 			}
+
+			@Override
 			public void close() {
 				open = false;
 			}
+
+			@Override
 			public boolean isOpen() {
 				return open;
 			}
+
+			@Override
 			public boolean exists(String path) throws IOException {
 				return true;
 			}
+
+			@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;
+			}
 		});
 		@SuppressWarnings("unchecked")
 		Message<File> out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("x/f1"));
@@ -743,44 +892,74 @@ public class RemoteFileOutboundGatewayTests {
 		new File(this.tmpDir + "/x").delete();
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "get", "payload");
+				(sessionFactory, "get", "payload");
 		gw.setLocalDirectory(new File(this.tmpDir + "/x"));
 		gw.afterPropertiesSet();
-		when(sessionFactory.getSession()).thenReturn(new Session(){
+		when(sessionFactory.getSession()).thenReturn(new Session() {
 			private boolean open = true;
+
+			@Override
 			public boolean remove(String path) throws IOException {
 				return false;
 			}
+
+			@Override
 			public TestLsEntry[] list(String path) throws IOException {
-				return new TestLsEntry[] {
+				return new TestLsEntry[]{
 						new TestLsEntry("f1", 1234, false, false, 12345, "-rw-r--r--")
 				};
 			}
+
+			@Override
 			public void read(String source, OutputStream outputStream)
 					throws IOException {
 				outputStream.write("testfile".getBytes());
 			}
+
+			@Override
 			public void write(InputStream inputStream, String destination)
 					throws IOException {
 			}
+
+			@Override
 			public boolean mkdir(String directory) throws IOException {
 				return true;
 			}
+
+			@Override
 			public void rename(String pathFrom, String pathTo)
 					throws IOException {
 			}
+
+			@Override
 			public void close() {
 				open = false;
 			}
+
+			@Override
 			public boolean isOpen() {
 				return open;
 			}
+
+			@Override
 			public boolean exists(String path) throws IOException {
 				return true;
 			}
+
+			@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;
+			}
 		});
 		gw.handleRequestMessage(new GenericMessage<String>("f1"));
 		File out = new File(this.tmpDir + "/x/f1");
@@ -793,7 +972,7 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		Session session = mock(Session.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
-			(sessionFactory, "rm", "payload");
+				(sessionFactory, "rm", "payload");
 		gw.afterPropertiesSet();
 		when(sessionFactory.getSession()).thenReturn(session);
 		when(session.remove("testremote/x/f1")).thenReturn(Boolean.TRUE);
@@ -812,9 +991,9 @@ public class RemoteFileOutboundGatewayTests {
 
 class TestRemoteFileOutboundGateway extends AbstractRemoteFileOutboundGateway<TestLsEntry> {
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@SuppressWarnings({"rawtypes", "unchecked"})
 	public TestRemoteFileOutboundGateway(SessionFactory sessionFactory,
-			String command, String expression) {
+										 String command, String expression) {
 		super(sessionFactory, Command.toCommand(command), expression);
 		this.setBeanFactory(mock(BeanFactory.class));
 	}
@@ -868,7 +1047,7 @@ class TestLsEntry extends AbstractFileInfo<TestLsEntry> {
 	private final String permissions;
 
 	public TestLsEntry(String filename, long size, boolean dir, boolean link,
-			long modified, String permissions) {
+					   long modified, String permissions) {
 		this.filename = filename;
 		this.size = size;
 		this.dir = dir;
@@ -877,30 +1056,37 @@ class TestLsEntry extends AbstractFileInfo<TestLsEntry> {
 		this.permissions = permissions;
 	}
 
+	@Override
 	public boolean isDirectory() {
 		return this.dir;
 	}
 
+	@Override
 	public long getModified() {
 		return this.modified;
 	}
 
+	@Override
 	public String getFilename() {
 		return this.filename;
 	}
 
+	@Override
 	public boolean isLink() {
 		return this.link;
 	}
 
+	@Override
 	public long getSize() {
 		return this.size;
 	}
 
+	@Override
 	public String getPermissions() {
 		return this.permissions;
 	}
 
+	@Override
 	public TestLsEntry getFileInfo() {
 		return this;
 	}
@@ -911,7 +1097,7 @@ class TestLsEntry extends AbstractFileInfo<TestLsEntry> {
 
 }
 
-class TestPatternFilter extends AbstractSimplePatternFileListFilter<TestLsEntry>{
+class TestPatternFilter extends AbstractSimplePatternFileListFilter<TestLsEntry> {
 
 	public TestPatternFilter(String path) {
 		super(path);

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-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.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.
@@ -19,17 +19,20 @@ package org.springframework.integration.ftp.session;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.commons.net.ftp.FTPClient;
 import org.apache.commons.net.ftp.FTPFile;
+import org.apache.commons.net.ftp.FTPReply;
+
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.util.Assert;
 
 /**
  * Implementation of {@link Session} for FTP.
- * 
+ *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
@@ -41,6 +44,7 @@ public class FtpSession implements Session<FTPFile> {
 
 	private final FTPClient client;
 
+	private final AtomicBoolean readingRaw = new AtomicBoolean();
 
 	public FtpSession(FTPClient client) {
 		Assert.notNull(client, "client must not be null");
@@ -48,6 +52,7 @@ public class FtpSession implements Session<FTPFile> {
 	}
 
 
+	@Override
 	public boolean remove(String path) throws IOException {
 		Assert.hasText(path, "path must not be null");
 		boolean completed = this.client.deleteFile(path);
@@ -56,17 +61,20 @@ public class FtpSession implements Session<FTPFile> {
 		}
 		return completed;
 	}
-	
+
+	@Override
 	public FTPFile[] list(String path) throws IOException {
 		Assert.hasText(path, "path must not be null");
 		return this.client.listFiles(path);
 	}
 
+	@Override
 	public String[] listNames(String path) throws IOException {
 		Assert.hasText(path, "path must not be null");
 		return this.client.listNames(path);
 	}
 
+	@Override
 	public void read(String path, OutputStream fos) throws IOException {
 		Assert.hasText(path, "path must not be null");
 		Assert.notNull(fos, "outputStream must not be null");
@@ -78,12 +86,40 @@ public class FtpSession implements Session<FTPFile> {
 		logger.info("File has been successfully transfered from: " + path);
 	}
 
+	@Override
+	public InputStream readRaw(String source) throws IOException {
+		if (!this.readingRaw.compareAndSet(false, true)) {
+			throw new IOException("Previous raw read was not finalized");
+		}
+		InputStream inputStream = this.client.retrieveFileStream(source);
+		if (inputStream == null) {
+			throw new IOException("Failed to obtain InputStream for remote file " + source + ": " + this.client.getReplyCode());
+		}
+		return inputStream;
+	}
+
+	@Override
+	public boolean finalizeRaw() throws IOException {
+		if (!this.readingRaw.compareAndSet(true, false)) {
+			throw new IOException("Raw read is not in process");
+		}
+		if (this.client.completePendingCommand()) {
+			int replyCode = this.client.getReplyCode();
+			if (logger.isDebugEnabled()) {
+				logger.debug(this + " finalizeRaw - reply code: " + replyCode);
+			}
+			return FTPReply.isPositiveCompletion(replyCode);
+		}
+		throw new IOException("completePendingCommandFailed");
+	}
+
+	@Override
 	public void write(InputStream inputStream, String path) throws IOException {
 		Assert.notNull(inputStream, "inputStream must not be null");
 		Assert.hasText(path, "path must not be null");
 		boolean completed = this.client.storeFile(path, inputStream);
 		if (!completed) {
-			throw new IOException("Failed to write to '" + path 
+			throw new IOException("Failed to write to '" + path
 					+ "'. Server replied with: " + this.client.getReplyString());
 		}
 		if (logger.isInfoEnabled()) {
@@ -91,6 +127,7 @@ public class FtpSession implements Session<FTPFile> {
 		}
 	}
 
+	@Override
 	public void close() {
 		try {
 			this.client.disconnect();
@@ -102,6 +139,7 @@ public class FtpSession implements Session<FTPFile> {
 		}
 	}
 
+	@Override
 	public boolean isOpen() {
 		try {
 			this.client.noop();
@@ -112,38 +150,41 @@ public class FtpSession implements Session<FTPFile> {
 		return true;
 	}
 
+	@Override
 	public void rename(String pathFrom, String pathTo) throws IOException{
 		this.client.deleteFile(pathTo);
 		boolean completed = this.client.rename(pathFrom, pathTo);
 		if (!completed) {
-			throw new IOException("Failed to rename '" + pathFrom + 
+			throw new IOException("Failed to rename '" + pathFrom +
 					"' to " + pathTo + "'. Server replied with: " + this.client.getReplyString());
 		}
 		if (logger.isInfoEnabled()) {
 			logger.info("File has been successfully renamed from: " + pathFrom + " to " + pathTo);
 		}
 	}
-	
+
+	@Override
 	public boolean mkdir(String remoteDirectory) throws IOException {
 		return this.client.makeDirectory(remoteDirectory);
 	}
-	
+
+	@Override
 	public boolean exists(String path) throws IOException{
 		Assert.hasText(path, "'path' must not be empty");
-	
+
 		String currentWorkingPath = this.client.printWorkingDirectory();
 		Assert.state(currentWorkingPath != null, "working directory cannot be determined, therefore exists check can not be completed");
 		boolean exists = false;
 
 		try {
-			if (this.client.changeWorkingDirectory(path)){	
+			if (this.client.changeWorkingDirectory(path)) {
 				exists = true;
 			}
-		} 
+		}
 		finally {
 			this.client.changeWorkingDirectory(currentWorkingPath);
 		}
-			
+
 		return exists;
 	}
 }

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TesFtpServer.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.ftp;
 
 import java.io.File;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.util.Arrays;
 
@@ -82,13 +83,22 @@ public class TesFtpServer {
 				sourceFtpDirectory.mkdir();
 				File file = new File(sourceFtpDirectory, "ftpSource1.txt");
 				file.createNewFile();
+				FileOutputStream fos = new FileOutputStream(file);
+				fos.write("source1".getBytes());
+				fos.close();
 				file = new File(sourceFtpDirectory, "ftpSource2.txt");
 				file.createNewFile();
+				fos = new FileOutputStream(file);
+				fos.write("source2".getBytes());
+				fos.close();
 
 				File subSourceFtpDirectory = new File(sourceFtpDirectory, "subFtpSource");
 				subSourceFtpDirectory.mkdir();
 				file = new File(subSourceFtpDirectory, "subFtpSource1.txt");
 				file.createNewFile();
+				fos = new FileOutputStream(file);
+				fos.write("subSource1".getBytes());
+				fos.close();
 
 				targetFtpDirectory = new File(ftpRootFolder, "ftpTarget");
 				targetFtpDirectory.mkdirs();

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java

@@ -19,8 +19,10 @@ package org.springframework.integration.ftp.outbound;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.util.List;
 
@@ -31,12 +33,14 @@ import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.ftp.TesFtpServer;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.util.FileCopyUtils;
 
 /**
  * @author Artem Bilan
@@ -170,4 +174,21 @@ public class FtpServerOutboundTests {
 
 	}
 
+	@Test
+	public void testInt3100RawGET() throws Exception {
+		Session<?> session = this.ftpServer.ftpSessionFactory().getSession();
+		ByteArrayOutputStream baos = new ByteArrayOutputStream();
+		FileCopyUtils.copy(session.readRaw("ftpSource/ftpSource1.txt"), baos);
+		assertTrue(session.finalizeRaw());
+		assertEquals("source1", new String(baos.toByteArray()));
+
+		baos = new ByteArrayOutputStream();
+		FileCopyUtils.copy(session.readRaw("ftpSource/ftpSource2.txt"), baos);
+		assertTrue(session.finalizeRaw());
+		assertEquals("source2", new String(baos.toByteArray()));
+
+		session.close();
+	}
+
+
 }

spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.http.config;
 
 import java.util.List;
+
 import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.BeanDefinitionStoreException;
@@ -74,7 +75,7 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse
 			throws BeanDefinitionStoreException {
 		String id = super.resolveId(element, definition, parserContext);
 
-		if (!element.hasAttribute(getInputChannelAttributeName())) {
+		if (!this.expectReply && !element.hasAttribute("channel")) {
 			// the created channel will get the 'id', so the adapter's bean name includes a suffix
 			id = id + ".adapter";
 		}

spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java

@@ -18,14 +18,17 @@ package org.springframework.integration.jms.config;
 
 import org.w3c.dom.Element;
 
+import org.springframework.beans.factory.BeanDefinitionStoreException;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.parsing.BeanComponentDefinition;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.jms.JmsMessageDrivenEndpoint;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
 import org.springframework.util.StringUtils;
 
 /**
@@ -79,6 +82,22 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
 		return JmsMessageDrivenEndpoint.class.getName();
 	}
 
+	@Override
+	protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
+			throws BeanDefinitionStoreException {
+		String id = super.resolveId(element, definition, parserContext);
+
+		if (!this.expectReply && !element.hasAttribute("channel")) {
+			// the created channel will get the 'id', so the adapter's bean name includes a suffix
+			id = id + ".adapter";
+		}
+		if (!StringUtils.hasText(id)) {
+			id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry());
+		}
+
+		return id;
+	}
+
 	@Override
 	protected boolean shouldGenerateId() {
 		return false;
@@ -100,7 +119,11 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
 	}
 
 	private String parseMessageListenerContainer(Element element, ParserContext parserContext) {
+		String containerClass = element.getAttribute("container-class");
 		if (element.hasAttribute("container")) {
+			if (StringUtils.hasText(containerClass)) {
+				parserContext.getReaderContext().error("Cannot have both 'container' and 'container-class'", element);
+			}
 			for (String containerAttribute : containerAttributes) {
 				if (element.hasAttribute(containerAttribute)) {
 					parserContext.getReaderContext().error("The '" + containerAttribute +
@@ -110,8 +133,13 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
 			return element.getAttribute("container");
 		}
 		// otherwise, we build a DefaultMessageListenerContainer instance
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
-				"org.springframework.jms.listener.DefaultMessageListenerContainer");
+		BeanDefinitionBuilder builder;
+		if (StringUtils.hasText(containerClass)) {
+			builder = BeanDefinitionBuilder.genericBeanDefinition(containerClass);
+		}
+		else {
+			builder = BeanDefinitionBuilder.genericBeanDefinition(DefaultMessageListenerContainer.class);
+		}
 		String destinationAttribute = this.expectReply ? "request-destination" : "destination";
 		String destinationNameAttribute = this.expectReply ? "request-destination-name" : "destination-name";
 		String pubSubDomainAttribute = this.expectReply ? "request-pub-sub-domain" : "pub-sub-domain";
@@ -198,7 +226,11 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
 			IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel");
 		}
 		else {
-			IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "requestChannel");
+			String channelName = element.getAttribute("channel");
+			if (!StringUtils.hasText(channelName)) {
+				channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
+			}
+			builder.addPropertyReference("requestChannel", channelName);
 			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout", "requestTimeout");
 			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload", "extractRequestPayload");
 		}

spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd

@@ -1201,6 +1201,11 @@
 			<xsd:extension base="jmsInboundAdapterType">
 				<xsd:attribute name="container" type="xsd:string">
 					<xsd:annotation>
+						<xsd:documentation><![CDATA[
+			A reference to a custom listener container implementation.
+			Note that a custom container class will typically be a subclass of DefaultMessageListenerContainer.
+			This attribute is mutually exclusive with 'container-class'.
+						]]></xsd:documentation>
 						<xsd:appinfo>
 							<tool:annotation kind="ref">
 								<tool:expected-type type="org.springframework.jms.listener.AbstractMessageListenerContainer"/>
@@ -1208,6 +1213,22 @@
 						</xsd:appinfo>
 					</xsd:annotation>
 				</xsd:attribute>
+				<xsd:attribute name="container-class" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation><![CDATA[
+			A custom listener container implementation class as fully qualified class name.
+			Default is Spring's standard DefaultMessageListenerContainer.
+			Note that a custom container class will typically be a subclass of this
+			standard container class. This attribute is mutually exclusive with 'container'.
+						]]></xsd:documentation>
+						<xsd:appinfo>
+							<tool:annotation>
+								<tool:expected-type type="java.lang.Class"/>
+								<tool:assignable-to type="org.springframework.jms.listener.AbstractMessageListenerContainer"/>
+							</tool:annotation>
+						</xsd:appinfo>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java

@@ -26,18 +26,20 @@ import org.junit.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.PollableChannel;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.jms.JmsMessageDrivenEndpoint;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
 import org.springframework.jms.listener.DefaultMessageListenerContainer;
 import org.springframework.jms.support.destination.JmsDestinationAccessor;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
 
 /**
  * @author Mark Fisher
  * @author Michael Bannister
+ * @author Gary Russell
  */
 public class JmsMessageDrivenChannelAdapterParserTests {
 
@@ -58,6 +60,7 @@ public class JmsMessageDrivenChannelAdapterParserTests {
 		assertNotNull("message should not be null", message);
 		assertEquals("test [with selector: TestProperty = 'foo']", message.getPayload());
 		endpoint.stop();
+		context.close();
 	}
 
 	@Test
@@ -68,6 +71,7 @@ public class JmsMessageDrivenChannelAdapterParserTests {
 		JmsDestinationAccessor container = (JmsDestinationAccessor) new DirectFieldAccessor(endpoint).getPropertyValue("listenerContainer");
 		assertEquals(Boolean.TRUE, container.isPubSubDomain());
 		endpoint.stop();
+		context.close();
 	}
 
 	@Test
@@ -81,66 +85,91 @@ public class JmsMessageDrivenChannelAdapterParserTests {
 		assertEquals("testDurableSubscriptionName", container.getDurableSubscriptionName());
 		assertEquals("testClientId", container.getClientId());
 		endpoint.stop();
+		context.close();
 	}
 
 	@Test
 	public void adapterWithTaskExecutor() {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"jmsInboundWithTaskExecutor.xml", this.getClass());
-		JmsMessageDrivenEndpoint endpoint = context.getBean("messageDrivenAdapter", JmsMessageDrivenEndpoint.class);
+		JmsMessageDrivenEndpoint endpoint = context.getBean("messageDrivenAdapter.adapter", JmsMessageDrivenEndpoint.class);
 		DefaultMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "listenerContainer",
 				DefaultMessageListenerContainer.class);
 		assertSame(context.getBean("exec"), TestUtils.getPropertyValue(container, "taskExecutor"));
 		endpoint.stop();
+		context.close();
 	}
 
 	@Test
-	public void testGatewayWithReceiveTimeout() {
+	public void testAdapterWithReceiveTimeout() {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"jmsInboundWithContainerSettings.xml", this.getClass());
-		JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithReceiveTimeout");
-		gateway.start();
+		JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithReceiveTimeout.adapter");
+		adapter.start();
 		AbstractMessageListenerContainer container = (AbstractMessageListenerContainer)
-				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
+				new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer");
 		assertEquals(1111L, new DirectFieldAccessor(container).getPropertyValue("receiveTimeout"));
-		gateway.stop();
+		adapter.stop();
+		context.close();
 	}
 
 	@Test
-	public void testGatewayWithRecoveryInterval() {
+	public void testAdapterWithRecoveryInterval() {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"jmsInboundWithContainerSettings.xml", this.getClass());
-		JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithRecoveryInterval");
-		gateway.start();
+		JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithRecoveryInterval.adapter");
+		adapter.start();
 		AbstractMessageListenerContainer container = (AbstractMessageListenerContainer)
-				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
+				new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer");
 		assertEquals(2222L, new DirectFieldAccessor(container).getPropertyValue("recoveryInterval"));
-		gateway.stop();
+		adapter.stop();
+		context.close();
 	}
 
 	@Test
-	public void testGatewayWithIdleTaskExecutionLimit() {
+	public void testAdapterWithIdleTaskExecutionLimit() {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"jmsInboundWithContainerSettings.xml", this.getClass());
-		JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleTaskExecutionLimit");
-		gateway.start();
+		JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleTaskExecutionLimit.adapter");
+		adapter.start();
 		AbstractMessageListenerContainer container = (AbstractMessageListenerContainer)
-				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
+				new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer");
 		assertEquals(7, new DirectFieldAccessor(container).getPropertyValue("idleTaskExecutionLimit"));
-		gateway.stop();
+		adapter.stop();
+		context.close();
 	}
 
 	@Test
-	public void testGatewayWithIdleConsumerLimit() {
+	public void testAdapterWithIdleConsumerLimit() {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"jmsInboundWithContainerSettings.xml", this.getClass());
-		JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleConsumerLimit");
-		gateway.start();
+		JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleConsumerLimit.adapter");
+		adapter.start();
 		AbstractMessageListenerContainer container = (AbstractMessageListenerContainer)
-				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
+				new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer");
 		assertEquals(33, new DirectFieldAccessor(container).getPropertyValue("idleConsumerLimit"));
 		assertEquals(3, new DirectFieldAccessor(container).getPropertyValue("cacheLevel"));
-		gateway.stop();
+		adapter.stop();
+		context.close();
+	}
+
+	@Test
+	public void testAdapterWithContainerClass() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"jmsInboundWithContainerClass.xml", this.getClass());
+		JmsMessageDrivenEndpoint adapter = context.getBean("adapterWithIdleConsumerLimit.adapter", JmsMessageDrivenEndpoint.class);
+		MessageChannel channel = context.getBean("adapterWithIdleConsumerLimit", MessageChannel.class);
+		assertSame(channel, TestUtils.getPropertyValue(adapter, "listener.gatewayDelegate.requestChannel"));
+		adapter.start();
+		FooContainer container = TestUtils.getPropertyValue(adapter, "listenerContainer", FooContainer.class);
+		assertEquals(33, new DirectFieldAccessor(container).getPropertyValue("idleConsumerLimit"));
+		assertEquals(3, new DirectFieldAccessor(container).getPropertyValue("cacheLevel"));
+		adapter.stop();
+		context.close();
+	}
+
+	public static final class FooContainer extends DefaultMessageListenerContainer {
+
 	}
 
 }

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithContainerClass.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xmlns:jms="http://www.springframework.org/schema/integration/jms"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/integration/jms
+			http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
+
+	<jms:message-driven-channel-adapter id="adapterWithIdleConsumerLimit"
+						 connection-factory="testConnectionFactory"
+						 destination-name="testQueue"
+						 container-class="org.springframework.integration.jms.config.JmsMessageDrivenChannelAdapterParserTests$FooContainer"
+						 idle-consumer-limit="33"
+						 cache-level="3"
+						 auto-startup="false" />
+
+	<bean id="testConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
+		<constructor-arg>
+			<bean class="org.springframework.integration.jms.StubConnection">
+				<constructor-arg value="message-driven-test"/>
+			</bean>
+		</constructor-arg>
+	</bean>
+
+</beans>

spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java

@@ -45,6 +45,7 @@ public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterPars
 		builder.addConstructorArgReference(connectionFactory);
 		builder.addPropertyReference("outputChannel", channelName);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topics");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topic-patterns");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true);

spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java

@@ -50,6 +50,7 @@ public class RedisQueueInboundChannelAdapterParser extends AbstractChannelAdapte
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval");
 		builder.addPropertyReference("outputChannel", channelName);
 
 		return builder.getBeanDefinition();

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java

@@ -21,7 +21,9 @@ import java.util.List;
 
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.PatternTopic;
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.Topic;
 import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
@@ -35,6 +37,7 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.1
  */
 public class RedisInboundChannelAdapter extends MessageProducerSupport {
@@ -45,6 +48,8 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
 
 	private volatile String[] topics;
 
+	private volatile String[] topicPatterns;
+
 	private volatile RedisSerializer<?> serializer = new StringRedisSerializer();
 
 	public RedisInboundChannelAdapter(RedisConnectionFactory connectionFactory) {
@@ -60,6 +65,10 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
 		this.topics = topics;
 	}
 
+	public void setTopicPatterns(String... topicPatterns) {
+		this.topicPatterns = topicPatterns;
+	}
+
 	public void setMessageConverter(MessageConverter messageConverter) {
 		Assert.notNull(messageConverter, "messageConverter must not be null");
 		this.messageConverter = messageConverter;
@@ -73,13 +82,32 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
 	@Override
 	protected void onInit() {
 		super.onInit();
-		Assert.notEmpty(this.topics, "at least one topis is required for subscription");
+		boolean hasTopics = false;
+		if (this.topics != null) {
+			Assert.noNullElements(this.topics, "'topics' may not contain null elements.");
+			hasTopics = true;
+		}
+		boolean hasPatterns = false;
+		if (this.topicPatterns != null) {
+			Assert.noNullElements(this.topicPatterns, "'topicPatterns' may not contain null elements.");
+			hasPatterns = true;
+
+		}
+		Assert.state(hasTopics || hasPatterns, "at least one topic or topic pattern is required for subscription.");
+
 		MessageListenerDelegate delegate = new MessageListenerDelegate();
 		MessageListenerAdapter adapter = new MessageListenerAdapter(delegate);
 		adapter.setSerializer(this.serializer);
-		List<ChannelTopic> topicList = new ArrayList<ChannelTopic>();
-		for (String topic : this.topics) {
-			topicList.add(new ChannelTopic(topic));
+		List<Topic> topicList = new ArrayList<Topic>();
+		if (hasTopics) {
+			for (String topic : this.topics) {
+				topicList.add(new ChannelTopic(topic));
+			}
+		}
+		if (hasPatterns) {
+			for (String pattern : this.topicPatterns) {
+				topicList.add(new PatternTopic(pattern));
+			}
 		}
 		adapter.afterPropertiesSet();
 		this.container.addMessageListener(adapter, topicList);

spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd

@@ -147,6 +147,13 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="topic-patterns" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+	Redis topic patterns as a comma-delimited list of Strings.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="message-converter" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
@@ -392,6 +399,15 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="recovery-interval" type="xsd:string" default="5000">
+						<xsd:annotation>
+							<xsd:documentation>
+								Specify the time in milliseconds for which the listener task should sleep after catching
+								an Exception on a Redis operation, before restarting the listener task.
+								Default is 5 seconds.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="expect-message" type="xsd:string" default="false">
 						<xsd:annotation>
 							<xsd:documentation>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml

@@ -7,7 +7,7 @@
 		http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd">
 
 	<int-redis:inbound-channel-adapter
-		id="adapter" topics="foo, bar" channel="receiveChannel" error-channel="testErrorChannel"
+		id="adapter" topics="foo" topic-patterns="f*, b*" channel="receiveChannel" error-channel="testErrorChannel"
 		message-converter="testConverter"
 		serializer="serializer"/>
 
@@ -25,14 +25,14 @@
 		class="org.springframework.integration.redis.config.RedisInboundChannelAdapterParserTests$TestMessageConverter" />
 
 	<int-redis:inbound-channel-adapter
-		id="autoChannel" topics="foo, bar" error-channel="testErrorChannel"
-		message-converter="testConverter" />
+		id="autoChannel" topics="foo1, bar1" error-channel="testErrorChannel"
+		message-converter="testConverter" auto-startup="false"/>
 
 	<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
 
 	<int:bridge input-channel="autoChannel" output-channel="nullChannel"/>
 
-	<int-redis:inbound-channel-adapter id="withoutSerializer" topics="foo" serializer=""/>
+	<int-redis:inbound-channel-adapter id="withoutSerializer" topics="foo" auto-startup="false" serializer=""/>
 
 
 </beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java

@@ -20,7 +20,9 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 
+import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -29,12 +31,14 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationContext;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter;
 import org.springframework.integration.redis.rules.RedisAvailable;
 import org.springframework.integration.redis.rules.RedisAvailableTests;
 import org.springframework.integration.support.converter.SimpleMessageConverter;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -59,7 +63,6 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests {
 	private RedisInboundChannelAdapter autoChannelAdapter;
 
 	@Test
-	@RedisAvailable
 	public void validateConfiguration() {
 		RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class);
 		assertEquals("adapter", adapter.getComponentName());
@@ -79,18 +82,24 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests {
 	@Test
 	@RedisAvailable
 	public void testInboundChannelAdapterMessaging() throws Exception {
+		RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class);
+		this.awaitContainerSubscribedWithPatterns(TestUtils.getPropertyValue(adapter, "container", RedisMessageListenerContainer.class));
+
 		RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
 
 		connectionFactory.getConnection().publish("foo".getBytes(), "Hello Redis from foo".getBytes());
+		connectionFactory.getConnection().publish("bar".getBytes(), "Hello Redis from bar".getBytes());
 
 		QueueChannel receiveChannel = context.getBean("receiveChannel", QueueChannel.class);
-		assertEquals("Hello Redis from foo", receiveChannel.receive(2000).getPayload());
-		connectionFactory.getConnection().publish("bar".getBytes(), "Hello Redis from bar".getBytes());
-		assertEquals("Hello Redis from bar", receiveChannel.receive(2000).getPayload());
+		for (int i = 0; i < 3; i++) {
+			Message<?> receive = receiveChannel.receive(2000);
+			assertNotNull(receive);
+			assertThat(receive.getPayload(), Matchers.<Object> isOneOf("Hello Redis from foo", "Hello Redis from bar"));
+		}
+
 	}
 
 	@Test
-	@RedisAvailable
 	public void testAutoChannel() {
 		assertSame(autoChannel, TestUtils.getPropertyValue(autoChannelAdapter, "outputChannel"));
 	}

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml

@@ -28,6 +28,7 @@
 											  serializer="serializer"
 											  error-channel="errorChannel"
 											  receive-timeout="2000"
+											  recovery-interval="3000"
 											  task-executor="executor"
 											  auto-startup="false"
 											  phase="100"/>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java

@@ -46,7 +46,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
-public class RedisMessageDrivenEndpointParserTests {
+public class RedisQueueInboundChannelAdapterParserTests {
 
 	@Autowired
 	@Qualifier("redisConnectionFactory")
@@ -90,6 +90,7 @@ public class RedisMessageDrivenEndpointParserTests {
 		assertEquals("si.test.Int3017.Inbound1", TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.key"));
 		assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "expectMessage", Boolean.class));
 		assertEquals(new Long(1000), TestUtils.getPropertyValue(this.defaultAdapter, "receiveTimeout", Long.class));
+		assertEquals(new Long(5000), TestUtils.getPropertyValue(this.defaultAdapter, "recoveryInterval", Long.class));
 		assertNull(TestUtils.getPropertyValue(this.defaultAdapter, "errorChannel"));
 		assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "taskExecutor"), Matchers.instanceOf(ErrorHandlingTaskExecutor.class));
 		assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "serializer"), Matchers.instanceOf(JdkSerializationRedisSerializer.class));
@@ -103,6 +104,7 @@ public class RedisMessageDrivenEndpointParserTests {
 		assertEquals("si.test.Int3017.Inbound2", TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.key"));
 		assertTrue(TestUtils.getPropertyValue(this.customAdapter, "expectMessage", Boolean.class));
 		assertEquals(new Long(2000), TestUtils.getPropertyValue(this.customAdapter, "receiveTimeout", Long.class));
+		assertEquals(new Long(3000), TestUtils.getPropertyValue(this.customAdapter, "recoveryInterval", Long.class));
 		assertSame(this.errorChannel, TestUtils.getPropertyValue(this.customAdapter, "errorChannel"));
 		assertSame(this.taskExecutor, TestUtils.getPropertyValue(this.customAdapter, "taskExecutor"));
 		assertSame(this.serializer, TestUtils.getPropertyValue(this.customAdapter, "serializer"));

spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java

@@ -20,12 +20,15 @@ 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 java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.hamcrest.Matchers;
 import org.junit.Test;
@@ -213,6 +216,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 
 		final List<ApplicationEvent> exceptionEvents = new ArrayList<ApplicationEvent>();
 
+		final CountDownLatch exceptionsLatch = new CountDownLatch(2);
+
 		RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
 		endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
 		endpoint.setApplicationEventPublisher(new ApplicationEventPublisher() {
@@ -220,6 +225,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 			@Override
 			public void publishEvent(ApplicationEvent event) {
 				exceptionEvents.add(event);
+				exceptionsLatch.countDown();
 			}
 		});
 		endpoint.setOutputChannel(channel);
@@ -228,16 +234,26 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 		endpoint.afterPropertiesSet();
 		endpoint.start();
 
+		int n = 0;
+		do {
+			n++;
+			if (n == 100) {
+				break;
+			}
+			Thread.sleep(100);
+		} while (!endpoint.isListening());
+
+		assertTrue(n < 100);
+
 		((DisposableBean) this.connectionFactory).destroy();
 
-		Thread.sleep(300);
+		assertTrue(exceptionsLatch.await(10, TimeUnit.SECONDS));
 
-		assertThat(exceptionEvents.size(), Matchers.greaterThan(0));
 		for (ApplicationEvent exceptionEvent : exceptionEvents) {
 			assertThat(exceptionEvent, Matchers.instanceOf(RedisExceptionEvent.class));
 			assertSame(endpoint, exceptionEvent.getSource());
 			assertThat(((IntegrationEvent) exceptionEvent).getCause().getClass(),
-					Matchers.isIn(Arrays.<Class<? extends Throwable>> asList(RedisSystemException.class, RedisConnectionFailureException.class)));
+					Matchers.isIn(Arrays.<Class<? extends Throwable>>asList(RedisSystemException.class, RedisConnectionFailureException.class)));
 		}
 
 		((InitializingBean) this.connectionFactory).afterPropertiesSet();

spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java

@@ -68,9 +68,25 @@ public class RedisAvailableTests {
 		while (n++ < 100 && !connection.isSubscribed()) {
 			Thread.sleep(100);
 		}
+		// TODO: remove this additional delay when/if https://jira.springsource.org/browse/DATAREDIS-242 is resolved
+		Thread.sleep(250);
 		assertTrue("RedisMessageListenerContainer Failed to Subscribe", n < 100);
 	}
 
+	protected void awaitContainerSubscribedWithPatterns(RedisMessageListenerContainer container) throws Exception {
+		this.awaitContainerSubscribed(container);
+		RedisConnection connection = TestUtils.getPropertyValue(container, "subscriptionTask.connection",
+				RedisConnection.class);
+
+		int n = 0;
+		while (n++ < 100 && connection.getSubscription().getPatterns().size() == 0) {
+			Thread.sleep(100);
+		}
+		// TODO: remove this additional delay when/if https://jira.springsource.org/browse/DATAREDIS-242 is resolved
+		Thread.sleep(250);
+		assertTrue("RedisMessageListenerContainer Failed to Subscribe with patterns", n < 100);
+	}
+
 	protected void prepareList(RedisConnectionFactory connectionFactory){
 
 		StringRedisTemplate redisTemplate = new StringRedisTemplate();
@@ -119,4 +135,5 @@ public class RedisAvailableTests {
 		ops.add("Abraham Lincoln", 19);
 		ops.add("George Washington", 18);
 	}
+
 }

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

@@ -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.
@@ -25,8 +25,10 @@ import java.util.Vector;
 
 import org.apache.commons.logging.Log;
 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;
 
@@ -52,15 +54,26 @@ 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 {
 		Assert.state(this.channel != null, "session is not connected");
 		try {
@@ -72,6 +85,7 @@ class SftpSession implements Session<LsEntry> {
 		}
 	}
 
+	@Override
 	public LsEntry[] list(String path) throws IOException {
 		Assert.state(this.channel != null, "session is not connected");
 		try {
@@ -92,6 +106,7 @@ class SftpSession implements Session<LsEntry> {
 		return new LsEntry[0];
 	}
 
+	@Override
 	public String[] listNames(String path) throws IOException {
 		LsEntry[] entries = this.list(path);
 		List<String> names = new ArrayList<String>();
@@ -107,6 +122,7 @@ class SftpSession implements Session<LsEntry> {
 	}
 
 
+	@Override
 	public void read(String source, OutputStream os) throws IOException {
 		Assert.state(this.channel != null, "session is not connected");
 		try {
@@ -114,10 +130,26 @@ class SftpSession implements Session<LsEntry> {
 			FileCopyUtils.copy(is, os);
 		}
 		catch (SftpException e) {
-			throw new NestedIOException("failed to read file", e);
+			throw new NestedIOException("failed to read file " + source, e);
 		}
 	}
 
+	@Override
+	public InputStream readRaw(String source) throws IOException {
+		try {
+			return this.channel.get(source);
+		}
+		catch (SftpException e) {
+			throw new NestedIOException("failed to read file " + source, e);
+		}
+	}
+
+	@Override
+	public boolean finalizeRaw() throws IOException {
+		return true;
+	}
+
+	@Override
 	public void write(InputStream inputStream, String destination) throws IOException {
 		Assert.state(this.channel != null, "session is not connected");
 		try {
@@ -128,38 +160,48 @@ 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
 	public void rename(String pathFrom, String pathTo) throws IOException {
-		try {	
+		try {
 			this.channel.rename(pathFrom, pathTo);
-		} 
+		}
 		catch (SftpException sftpex) {
 			if (logger.isDebugEnabled()){
-				logger.debug("Initial File rename failed, possibly because file already exists. Will attempt to delete file: " 
+				logger.debug("Initial File rename failed, possibly because file already exists. Will attempt to delete file: "
 						+ pathTo + " and execute rename again.");
 			}
-			try {			
+			try {
 				this.remove(pathTo);
 				if (logger.isDebugEnabled()) {
 					logger.debug("Delete file: " + pathTo + " succeeded. Will attempt rename again");
-				}		
-			} 
+				}
+			}
 			catch (IOException ioex) {
 				throw new NestedIOException("Failed to delete file " + pathTo, ioex);
 			}
 			try {
 				// attempt to rename again
 				this.channel.rename(pathFrom, pathTo);
-			} 
+			}
 			catch (SftpException sftpex2) {
 				throw new NestedIOException("failed to rename from " + pathFrom + " to " + pathTo, sftpex2);
 			}
@@ -169,8 +211,9 @@ class SftpSession implements Session<LsEntry> {
 		}
 	}
 
+	@Override
 	public boolean mkdir(String remoteDirectory) throws IOException {
-		try {	
+		try {
 			this.channel.mkdir(remoteDirectory);
 		}
 		catch (SftpException e) {
@@ -179,6 +222,7 @@ class SftpSession implements Session<LsEntry> {
 		return true;
 	}
 
+	@Override
 	public boolean exists(String path) {
 		try {
 			this.channel.lstat(path);
@@ -189,13 +233,13 @@ class SftpSession implements Session<LsEntry> {
 		}
 		return false;
 	}
-	
+
 	void connect() {
 		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();
 			}
@@ -204,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.messaging.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.messaging.support.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,14 +18,22 @@ 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;
 import static org.mockito.Mockito.mock;
 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;
@@ -38,11 +46,13 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.util.FileCopyUtils;
 
 import com.jcraft.jsch.ChannelSftp.LsEntry;
 import com.jcraft.jsch.SftpATTRS;
@@ -55,10 +65,10 @@ import com.jcraft.jsch.SftpATTRS;
  * <pre class="code">
  *  $ tree sftpSource/
  *  sftpSource/
- *  ├── sftpSource1.txt
- *  ├── sftpSource2.txt
+ *  ├── sftpSource1.txt - contains 'source1'
+ *  ├── sftpSource2.txt - contains 'source2'
  *  └── subSftpSource
- *      └── subSftpSource1.txt
+ *      └── subSftpSource1.txt - contains 'subSource1'
  * </pre>
  *
  * @author Artem Bilan
@@ -156,6 +166,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);
@@ -171,6 +186,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
@@ -251,4 +271,86 @@ public class SftpServerOutboundTests {
 
 	}
 
+	/**
+	 * Only runs with a real server (see class javadocs).
+	 */
+	@Test
+	public void testInt3100RawGET() throws Exception {
+		if (!sessionFactory.toString().startsWith("Mock for")) {
+			Session<?> session = this.sessionFactory.getSession();
+			ByteArrayOutputStream baos = new ByteArrayOutputStream();
+			FileCopyUtils.copy(session.readRaw("sftpSource/sftpSource1.txt"), baos);
+			assertTrue(session.finalizeRaw());
+			assertEquals("source1", new String(baos.toByteArray()));
+
+			baos = new ByteArrayOutputStream();
+			FileCopyUtils.copy(session.readRaw("sftpSource/sftpSource2.txt"), baos);
+			assertTrue(session.finalizeRaw());
+			assertEquals("source2", new String(baos.toByteArray()));
+
+			session.close();
+		}
+	}
+
+	@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/jms.xml

@@ -87,12 +87,21 @@
       message-driven Channel Adapter with a <classname>Destination</classname> reference.
       <programlisting language="xml"><![CDATA[<int-jms:message-driven-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel"/>]]></programlisting>
       <note>
+       <para>
         The Message-Driven adapter also accepts several properties that pertain to the MessageListener container.
-        These values are only considered if you do not provide an actual 'container' reference. In that case,
+        These values are only considered if you do not provide a <code>container</code> reference. In that case,
         an instance of DefaultMessageListenerContainer will be created and configured based on these properties.
         For example, you can specify the "transaction-manager" reference, the "concurrent-consumers" value, and
         several other property references and values. Refer to the JavaDoc and Spring Integration's JMS Schema
-        (spring-integration-jms.xsd) for more detail.
+        (spring-integration-jms.xsd) for more details.
+       </para>
+       <para>
+        If you have a custom listener container implementation (usually a subclass of
+        <classname>DefaultMessageListenerContainer</classname>), you can either provide a reference to an instance
+        of it using the <code>container</code> attribute, or simply provide its fully qualified class name using
+        the <code>container-class</code> attribute. In that case, the attributes on the adapter
+        are transferred to an instance of your custom container.
+       </para>
       </note>
     </para>
     <para>

src/reference/docbook/redis.xml

@@ -150,8 +150,15 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
       Redis Messages and the Spring Integration Message payloads. The default is a <code>SimpleMessageConverter</code>.
       </para>
 
-      <para>Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the
-      <code>topics</code> attribute.</para>
+      <para>
+       Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the
+       <code>topics</code> attribute.
+	  </para>
+      <para>
+        Since <emphasis>Spring Integration 3.0</emphasis>, the Inbound Adapter, in addition to the existing <code>topics</code> attribute,
+        now has the <code>topic-patterns</code> attribute. This attribute contains a comma-delimited set of Redis topic patterns.
+        For more information regarding Redis publish/subscribe, see <ulink url="http://redis.io/topics/pubsub">Redis Pub/Sub</ulink>.
+      </para>
       <para>
          Inbound adapters can use a <classname>RedisSerializer</classname> to deserialize the body of Redis Messages.
          The <code>serializer</code> attribute of the <code>&lt;int-redis:inbound-channel-adapter&gt;</code> can be set to an
@@ -205,6 +212,7 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
                     error-channel="" ]]><co id="redis-m-d-c-a-errorChannel"/><![CDATA[
                     serializer="" ]]><co id="redis-m-d-c-a-serializer"/><![CDATA[
                     receive-timeout="" ]]><co id="redis-m-d-c-a-receiveTimeout"/><![CDATA[
+                    recovery-interval="" ]]><co id="redis-m-d-c-a-recoveryInterval"/><![CDATA[
                     expect-message="" ]]><co id="redis-m-d-c-a-expectMessage"/><![CDATA[
                     task-executor=""/> ]]><co id="redis-m-d-c-a-task-executor"/>
 		</programlisting>
@@ -247,7 +255,9 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
 			<callout arearefs="redis-m-d-c-a-errorChannel">
 				<para>
 					The <interfacename>MessageChannel</interfacename> to which to send <interfacename>ErrorMessage</interfacename>s with
-					<interfacename>Exception</interfacename>s from the listening task of the Endpoint.
+					<interfacename>Exception</interfacename>s from the listening task of the Endpoint. By default
+					the underlying <classname>MessagePublishingErrorHandler</classname> uses the
+					default <code>errorChannel</code> from the application context.
 				</para>
 			</callout>
 			<callout arearefs="redis-m-d-c-a-serializer">
@@ -262,6 +272,12 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
 					The timeout in milliseconds for 'right pop' operation to wait for a Redis message from the queue. Default is 1 second.
 				</para>
 			</callout>
+			<callout arearefs="redis-m-d-c-a-recoveryInterval">
+				<para>
+					The time in milliseconds for which the listener task should sleep after exceptions on the 'right pop' operation,
+					before restarting the listener task.
+				</para>
+			</callout>
 			<callout arearefs="redis-m-d-c-a-expectMessage">
 				<para>
 				   Specify if this Endpoint expects data from the Redis queue to contain entire <interfacename>Message</interfacename>s.
@@ -344,6 +360,22 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
 			  </calloutlist>
 		  </para>
 	  </section>
+	  <section id="redis-application-events">
+		  <title>Redis Application Events</title>
+		  <para>
+			  Since <emphasis>Spring Integration 3.0</emphasis>, the Redis module provides an implementation
+			  of <classname>IntegrationEvent</classname> - which, in turn, is a
+			  <interfacename>org.springframework.context.ApplicationEvent</interfacename>. The <classname>RedisExceptionEvent</classname>
+			  encapsulates an <classname>Exception</classname>s from Redis operations (with the Endpoint being the <code>source</code>
+			  of the event). For example, the <code>&lt;int-redis:queue-inbound-channel-adapter/&gt;</code>
+			  emits those events after catching <classname>Exception</classname>s from the <code>BoundListOperations.rightPop</code>
+			  operation.
+			  The exception may be any generic <classname>org.springframework.data.redis.RedisSystemException</classname> or
+			  a <classname>org.springframework.data.redis.RedisConnectionFailureException</classname>.
+			  Handling these events using an <code>&lt;int-event:inbound-channel-adapter/&gt;</code> can be useful to determine
+			  problems with background Redis tasks and to take administrative actions.
+		  </para>
+	  </section>
   </section>
 
   <section id="redis-message-store">

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>
@@ -584,6 +594,10 @@
 						The Redis Outbound Channel Adapter now has the <code>topic-expression</code> property to determine
 						the Redis topic against the Message at runtime.
 					</listitem>
+					<listitem>
+						The Redis Inbound Channel Adapter, in addition to the existing <code>topics</code> attribute,
+						now has the <code>topic-patterns</code> attribute.
+					</listitem>
 				</itemizedlist>
 			</para>
 			<para>