INT-3100 Support Remote File Streaming (Inbound)

- Add readRaw and finalizeRaw methods to Session
- Add tests for SFTP and FTP

Allows retrieval of a remote file as a stream.

JIRA: https://jira.springsource.org/browse/INT-3100

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

@@ -65,14 +65,17 @@ 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();
 			}
@@ -100,6 +103,7 @@ 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());
 	}
@@ -107,6 +111,7 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 	/**
 	 * Remove (close) any unused sessions in the pool.
 	 */
+	@Override
 	public void destroy() {
 		this.pool.removeAllIdleItems();
 	}
@@ -122,6 +127,7 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 			this.targetSession = targetSession;
 		}
 
+		@Override
 		public synchronized void close() {
 			if (released) {
 				if (logger.isDebugEnabled()){
@@ -137,41 +143,61 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 			}
 		}
 
+		@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/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java

@@ -55,7 +55,6 @@ import org.springframework.integration.support.MessageBuilder;
 /**
  * @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);
@@ -922,4 +1108,4 @@ class TestPatternFilter extends AbstractSimplePatternFileListFilter<TestLsEntry>
 		return file.getFilename();
 	}
 
-}
+}

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;
 
@@ -33,10 +35,12 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.Message;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.ftp.TesFtpServer;
 import org.springframework.integration.message.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-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,6 +25,7 @@ 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.util.Assert;
@@ -61,6 +62,7 @@ class SftpSession implements Session<LsEntry> {
 	}
 
 
+	@Override
 	public boolean remove(String path) throws IOException {
 		Assert.state(this.channel != null, "session is not connected");
 		try {
@@ -72,6 +74,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 +95,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 +111,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 +119,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 +149,41 @@ class SftpSession implements Session<LsEntry> {
 		}
 	}
 
+	@Override
 	public void close() {
 		if (this.jschSession.isConnected()) {
 			this.jschSession.disconnect();
 		}
 	}
 
+	@Override
 	public boolean isOpen() {
 		return 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 +193,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 +204,7 @@ class SftpSession implements Session<LsEntry> {
 		return true;
 	}
 
+	@Override
 	public boolean exists(String path) {
 		try {
 			this.channel.lstat(path);
@@ -189,7 +215,7 @@ class SftpSession implements Session<LsEntry> {
 		}
 		return false;
 	}
-	
+
 	void connect() {
 		try {
 			if (!this.jschSession.isConnected()) {

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

@@ -19,10 +19,12 @@ package org.springframework.integration.sftp.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 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.util.List;
@@ -44,6 +46,7 @@ import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.sftp.session.SftpFileInfo;
 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;
@@ -56,10 +59,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
@@ -252,4 +255,26 @@ 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();
+		}
+	}
+
+
 }