INT-3737: (S)FTP Outbound Gateway Streaming GET

JIRA: https://jira.spring.io/browse/INT-3737

Support returning an `InputStream` from a GET operation.

This can be used in conjuction with a `<file:splitter/>` to stream a text file.

The user is responsible for releasing the session when the download is complete.

The session object is stored in the message header and `RemoteFileUtils.closeSession()` should be called
to clean up and close the session.

INT-3737: Polishing and Docs

Fix `CachedSession.close()` bug

Preventing `Caused by: java.io.IOException: Previous raw read was not finalized`
when `Session` is returned to the pool, but `readingRaw` hasn't been finalized
because the real `close()` isn't invoked in case of normal return to pool.

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java

@@ -144,6 +144,9 @@ public class FtpSession implements Session<FTPFile> {
 	@Override
 	public void close() {
 		try {
+			if (this.readingRaw.get()) {
+				finalizeRaw();
+			}
 			this.client.disconnect();
 		}
 		catch (Exception e) {

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

@@ -24,6 +24,7 @@ import java.util.Arrays;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 
+import org.apache.commons.net.ftp.FTPFile;
 import org.apache.ftpserver.FtpServer;
 import org.apache.ftpserver.FtpServerFactory;
 import org.apache.ftpserver.ftplet.Authentication;
@@ -40,6 +41,8 @@ import org.junit.rules.TemporaryFolder;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.file.remote.session.CachingSessionFactory;
+import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
 import org.springframework.integration.test.util.SocketUtils;
 
@@ -149,13 +152,14 @@ public class TestFtpServer {
 	}
 
 	@Bean
-	public DefaultFtpSessionFactory ftpSessionFactory() {
+	public SessionFactory<FTPFile> ftpSessionFactory() {
 		DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory();
 		factory.setHost("localhost");
 		factory.setPort(this.ftpPort);
 		factory.setUsername("foo");
 		factory.setPassword("foo");
-		return factory;
+
+		return new CachingSessionFactory<FTPFile>(factory);
 	}
 
 	@PostConstruct

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

@@ -1,12 +1,13 @@
 <?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:int-ftp="http://www.springframework.org/schema/integration/ftp"
-	   xmlns:int="http://www.springframework.org/schema/integration"
-	   xsi:schemaLocation="http://www.springframework.org/schema/integration/ftp
-	 http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd
-		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
-		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int-ftp="http://www.springframework.org/schema/integration/ftp"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-file="http://www.springframework.org/schema/integration/file"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd
+		http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
 
 	<bean id="ftpServer" class="org.springframework.integration.ftp.TestFtpServer">
 		<constructor-arg value="FtpServerOutboundTests"/>
@@ -104,6 +105,25 @@
 							  remote-directory="ftpTarget"
 							  reply-channel="output"/>
 
+	<int-ftp:outbound-gateway session-factory="ftpSessionFactory"
+							  request-channel="inboundGetStream"
+							  command="get"
+							  command-options="-stream"
+							  expression="payload"
+							  remote-directory="ftpTarget"
+							  reply-channel="stream" />
+
+	<int:chain input-channel="stream">
+		<int-file:splitter markers="true" />
+		<int:payload-type-router resolution-required="false" default-output-channel="output">
+			<int:mapping type="org.springframework.integration.file.splitter.FileSplitter$FileMarker"
+						 channel="markers" />
+		</int:payload-type-router>
+	</int:chain>
+
+	<int:service-activator input-channel="markers"
+	  expression="payload.mark.toString().equals('END') ? headers['file_remoteSession'].close() : null"/>
+
 	<int:channel id="appending" />
 
 	<int-ftp:outbound-channel-adapter id="appender"

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-2015 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.
@@ -21,7 +21,9 @@ import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 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;
@@ -32,6 +34,7 @@ import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.net.ftp.FTPFile;
 import org.hamcrest.Matchers;
@@ -48,10 +51,11 @@ import org.springframework.integration.file.remote.InputStreamCallback;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.SessionCallback;
 import org.springframework.integration.file.remote.session.Session;
+import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.TestFtpServer;
-import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
 import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
@@ -74,7 +78,7 @@ public class FtpServerOutboundTests {
 	private TestFtpServer ftpServer;
 
 	@Autowired
-	private DefaultFtpSessionFactory ftpSessionFactory;
+	private SessionFactory<FTPFile> ftpSessionFactory;
 
 	@Autowired
 	private PollableChannel output;
@@ -112,6 +116,9 @@ public class FtpServerOutboundTests {
 	@Autowired
 	private DirectChannel failing;
 
+	@Autowired
+	private DirectChannel inboundGetStream;
+
 	@Before
 	public void setup() {
 		this.ftpServer.recursiveDelete(ftpServer.getTargetLocalDirectory());
@@ -342,6 +349,33 @@ public class FtpServerOutboundTests {
 
 	}
 
+	@Test
+	public void testStream() {
+		String dir = "ftpSource/";
+		this.inboundGetStream.send(new GenericMessage<Object>(dir + "ftpSource1.txt"));
+		Message<?> result = this.output.receive(1000);
+		assertNotNull(result);
+		assertEquals("source1", result.getPayload());
+		assertEquals("ftpSource/", result.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+		assertEquals("ftpSource1.txt", result.getHeaders().get(FileHeaders.REMOTE_FILE));
+
+		Session<?> session = (Session<?>) result.getHeaders().get(FileHeaders.REMOTE_SESSION);
+		// Returned to cache
+		assertTrue(session.isOpen());
+		// Raw reading is finished
+		assertFalse(TestUtils.getPropertyValue(session, "targetSession.readingRaw", AtomicBoolean.class).get());
+
+		// Check that we can use the same session from cache to read another remote InputStream
+		this.inboundGetStream.send(new GenericMessage<Object>(dir + "ftpSource2.txt"));
+		result = this.output.receive(1000);
+		assertNotNull(result);
+		assertEquals("source2", result.getPayload());
+		assertEquals("ftpSource/", result.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+		assertEquals("ftpSource2.txt", result.getHeaders().get(FileHeaders.REMOTE_FILE));
+		assertSame(TestUtils.getPropertyValue(session, "targetSession"),
+				TestUtils.getPropertyValue(result.getHeaders().get(FileHeaders.REMOTE_SESSION), "targetSession"));
+	}
+
 	private void assertLength6(FtpRemoteFileTemplate template) {
 		FTPFile[] files = template.execute(new SessionCallback<FTPFile, FTPFile[]>() {
 

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java

@@ -61,7 +61,7 @@ public class FtpRemoteFileTemplateTests {
 	private TestFtpServer ftpServer;
 
 	@Autowired
-	private DefaultFtpSessionFactory sessionFactory;
+	private SessionFactory<FTPFile> sessionFactory;
 
 	@Before
 	@After