INT-4015: Streaming Remote File Inbound Adapter

JIRA: https://jira.spring.io/browse/INT-4015
      https://jira.spring.io/browse/INT-3854

Initial commit.

Reworked to emit an input stream and use the file splitter.

Add StreamTransformer.

Add CLOSABLE_RESOURCE header so we can close the session automatically.

Implement INT-3854, FTP, SFTP

(S)FTP Namespace Changes

Docs - also fixes a PDF overflow

Polishing - PR Comments

checkstyle fixes

Polishing - Add Namespace for StreamParser

Polishing - PR Comments

spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java

@@ -35,6 +35,10 @@ public abstract class FileHeaders {
 
 	public static final String REMOTE_FILE = PREFIX + "remoteFile";
 
+	/**
+	 * @deprecated - use {@code IntegrationMessageHeaderAccessor#CLOSEABLE_RESOURCE}.
+	 */
+	@Deprecated
 	public static final String REMOTE_SESSION = PREFIX + "remoteSession";
 
 	public static final String RENAME_TO = PREFIX + "renameTo";

spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java

@@ -25,6 +25,8 @@ import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.ExpressionFactoryBean;
 import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
 import org.springframework.util.StringUtils;
 
 /**
@@ -41,7 +43,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 	@Override
 	protected final BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-				this.getInboundFileSynchronizerClassname());
+				this.getInboundFileSynchronizerClass());
 
 		synchronizerBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
 
@@ -57,7 +59,8 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 		String remoteFileSeparator = element.getAttribute("remote-file-separator");
 		synchronizerBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "temporary-file-suffix");
-		this.configureFilter(synchronizerBuilder, element, parserContext);
+		FileParserUtils.configureFilter(synchronizerBuilder, element, parserContext,
+				getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass());
 
 		// build the MessageSource
 		BeanDefinitionBuilder messageSourceBuilder =
@@ -82,53 +85,12 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 		return messageSourceBuilder.getBeanDefinition();
 	}
 
-	private void configureFilter(BeanDefinitionBuilder synchronizerBuilder, Element element,
-	                             ParserContext parserContext) {
-		String filter = element.getAttribute("filter");
-		String fileNamePattern = element.getAttribute("filename-pattern");
-		String fileNameRegex = element.getAttribute("filename-regex");
-		boolean hasFilter = StringUtils.hasText(filter);
-		boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern);
-		boolean hasFileNameRegex = StringUtils.hasText(fileNameRegex);
-		if (hasFilter || hasFileNamePattern || hasFileNameRegex) {
-			int count = 0;
-			if (hasFilter) {
-				count++;
-			}
-			if (hasFileNamePattern) {
-				count++;
-			}
-			if (hasFileNameRegex) {
-				count++;
-			}
-			if (count != 1) {
-				parserContext.getReaderContext().error("at most one of 'filename-pattern', " +
-						"'filename-regex', or 'filter' is allowed on remote file inbound adapter", element);
-			}
-			if (hasFilter) {
-				synchronizerBuilder.addPropertyReference("filter", filter);
-			}
-			else if (hasFileNamePattern) {
-				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-						this.getSimplePatternFileListFilterClassname());
-				filterBuilder.addConstructorArgValue(fileNamePattern);
-				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
-			}
-			else if (hasFileNameRegex) {
-				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-						this.getRegexPatternFileListFilterClassname());
-				filterBuilder.addConstructorArgValue(fileNameRegex);
-				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
-			}
-		}
-	}
-
 	protected abstract String getMessageSourceClassname();
 
-	protected abstract String getInboundFileSynchronizerClassname();
+	protected abstract Class<? extends InboundFileSynchronizer> getInboundFileSynchronizerClass();
 
-	protected abstract String getSimplePatternFileListFilterClassname();
+	protected abstract Class<? extends FileListFilter<?>> getSimplePatternFileListFilterClass();
 
-	protected abstract String getRegexPatternFileListFilterClassname();
+	protected abstract Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass();
 
 }

spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2016 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.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.BeanMetadataElement;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.RemoteFileOperations;
+import org.springframework.util.StringUtils;
+
+/**
+ * Abstract base class for parsing remote file streaming inbound channel adapters.
+ *
+ * @author Gary Russell
+ * @since 4.3
+ */
+public abstract class AbstractRemoteFileStreamingInboundChannelAdapterParser
+		extends AbstractPollingInboundChannelAdapterParser {
+
+	@Override
+	protected final BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
+		BeanDefinition templateDefinition = FileParserUtils.parseRemoteFileTemplate(element, parserContext, false,
+				getTemplateClass());
+
+		BeanDefinitionBuilder messageSourceBuilder =
+				BeanDefinitionBuilder.genericBeanDefinition(getMessageSourceClass());
+		messageSourceBuilder.addConstructorArgValue(templateDefinition);
+
+		BeanDefinition expressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression(
+				"remote-directory", "remote-directory-expression", parserContext, element, false);
+		if (expressionDef != null) {
+			messageSourceBuilder.addPropertyValue("remoteDirectoryExpression", expressionDef);
+		}
+
+		String remoteFileSeparator = element.getAttribute("remote-file-separator");
+		messageSourceBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator);
+		FileParserUtils.configureFilter(messageSourceBuilder, element, parserContext,
+				getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass());
+
+		String comparator = element.getAttribute("comparator");
+		if (StringUtils.hasText(comparator)) {
+			messageSourceBuilder.addConstructorArgReference(comparator);
+		}
+		return messageSourceBuilder.getBeanDefinition();
+	}
+
+	protected abstract Class<? extends RemoteFileOperations<?>> getTemplateClass();
+
+	protected abstract Class<? extends MessageSource<?>> getMessageSourceClass();
+
+	protected abstract Class<? extends FileListFilter<?>> getSimplePatternFileListFilterClass();
+
+	protected abstract Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass();
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.file.DefaultFileNameGenerator;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.RemoteFileOperations;
 import org.springframework.util.StringUtils;
 
@@ -89,4 +90,43 @@ public final class FileParserUtils {
 		return templateBuilder.getBeanDefinition();
 	}
 
+	static void configureFilter(BeanDefinitionBuilder synchronizerBuilder, Element element, ParserContext parserContext,
+			Class<? extends FileListFilter<?>> patternClass, Class<? extends FileListFilter<?>> regexClass) {
+		String filter = element.getAttribute("filter");
+		String fileNamePattern = element.getAttribute("filename-pattern");
+		String fileNameRegex = element.getAttribute("filename-regex");
+		boolean hasFilter = StringUtils.hasText(filter);
+		boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern);
+		boolean hasFileNameRegex = StringUtils.hasText(fileNameRegex);
+		if (hasFilter || hasFileNamePattern || hasFileNameRegex) {
+			int count = 0;
+			if (hasFilter) {
+				count++;
+			}
+			if (hasFileNamePattern) {
+				count++;
+			}
+			if (hasFileNameRegex) {
+				count++;
+			}
+			if (count != 1) {
+				parserContext.getReaderContext().error("at most one of 'filename-pattern', " +
+						"'filename-regex', or 'filter' is allowed on remote file inbound adapter", element);
+			}
+			if (hasFilter) {
+				synchronizerBuilder.addPropertyReference("filter", filter);
+			}
+			else if (hasFileNamePattern) {
+				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(patternClass);
+				filterBuilder.addConstructorArgValue(fileNamePattern);
+				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
+			}
+			else if (hasFileNameRegex) {
+				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(regexClass);
+				filterBuilder.addConstructorArgValue(fileNameRegex);
+				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
+			}
+		}
+	}
+
 }

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

@@ -0,0 +1,187 @@
+/*
+ * Copyright 2016 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;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.expression.Expression;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.endpoint.AbstractMessageSource;
+import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.session.Session;
+import org.springframework.messaging.MessagingException;
+import org.springframework.util.Assert;
+
+/**
+ * A message source that produces a message with an {@link InputStream} payload
+ * referencing a remote file.
+ *
+ * @author Gary Russell
+ * @since 4.3
+ *
+ */
+public abstract class AbstractRemoteFileStreamingMessageSource<F> extends AbstractMessageSource<InputStream>
+		implements BeanFactoryAware, InitializingBean {
+
+	private final RemoteFileTemplate<F> remoteFileTemplate;
+
+	private final BlockingQueue<AbstractFileInfo<F>> toBeReceived = new LinkedBlockingQueue<AbstractFileInfo<F>>();
+
+	private final Comparator<AbstractFileInfo<F>> comparator;
+
+	/**
+	 * the path on the remote server.
+	 */
+	private volatile Expression remoteDirectoryExpression;
+
+	private volatile String remoteFileSeparator = "/";
+
+	/**
+	 * An {@link FileListFilter} that runs against the <em>remote</em> file system view.
+	 */
+	private volatile FileListFilter<F> filter;
+
+	protected AbstractRemoteFileStreamingMessageSource(RemoteFileTemplate<F> template,
+			Comparator<AbstractFileInfo<F>> comparator) {
+		this.remoteFileTemplate = template;
+		this.comparator = comparator;
+	}
+
+	/**
+	 * Specify the full path to the remote directory.
+	 *
+	 * @param remoteDirectory The remote directory.
+	 */
+	public void setRemoteDirectory(String remoteDirectory) {
+		this.remoteDirectoryExpression = new LiteralExpression(remoteDirectory);
+	}
+
+	/**
+	 * Specify an expression that evaluates to the full path to the remote directory.
+	 *
+	 * @param remoteDirectoryExpression The remote directory expression.
+	 */
+	public void setRemoteDirectoryExpression(Expression remoteDirectoryExpression) {
+		Assert.notNull(remoteDirectoryExpression, "'remoteDirectoryExpression' must not be null");
+		this.remoteDirectoryExpression = remoteDirectoryExpression;
+	}
+
+	/**
+	 * Set the remote file separator; default '/'
+	 * @param remoteFileSeparator the remote file separator.
+	 */
+	public void setRemoteFileSeparator(String remoteFileSeparator) {
+		Assert.notNull(remoteFileSeparator, "'remoteFileSeparator' must not be null");
+		this.remoteFileSeparator = remoteFileSeparator;
+	}
+
+	/**
+	 * Set the filter to be applied to the remote files before transferring.
+	 * @param filter the file list filter.
+	 */
+	public void setFilter(FileListFilter<F> filter) {
+		this.filter = filter;
+	}
+
+	protected RemoteFileTemplate<F> getRemoteFileTemplate() {
+		return this.remoteFileTemplate;
+	}
+
+	@Override
+	public final void afterPropertiesSet() {
+		Assert.state(this.remoteDirectoryExpression != null, "'remoteDirectoryExpression' must not be null");
+		doInit();
+	}
+
+	/**
+	 * Subclasses can override to perform initialization - called from
+	 * {@link InitializingBean#afterPropertiesSet()}.
+	 */
+	protected void doInit() {
+	}
+
+	@Override
+	protected Object doReceive() {
+		AbstractFileInfo<F> file = poll();
+		if (file != null) {
+			String remotePath = remotePath(file);
+			Session<?> session = this.remoteFileTemplate.getSesssion();
+			try {
+				return getMessageBuilderFactory().withPayload(session.readRaw(remotePath))
+						.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
+						.setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory())
+						.setHeader(FileHeaders.REMOTE_FILE, file.getFilename())
+						.build();
+			}
+			catch (IOException e) {
+				return new MessagingException("IOException when retrieving " + remotePath, e);
+			}
+		}
+		return null;
+	}
+
+	protected AbstractFileInfo<F> poll() {
+		if (this.toBeReceived.size() == 0) {
+			listFiles();
+		}
+		return this.toBeReceived.poll();
+	}
+
+	protected String remotePath(AbstractFileInfo<F> file) {
+		String remotePath = file.getRemoteDirectory().endsWith(this.remoteFileSeparator)
+				? file.getRemoteDirectory() + file.getFilename()
+				: file.getRemoteDirectory() + this.remoteFileSeparator + file.getFilename();
+		return remotePath;
+	}
+
+	private void listFiles() {
+		String remoteDirectory = this.remoteDirectoryExpression.getValue(getEvaluationContext(), String.class);
+		F[] files = this.remoteFileTemplate.list(remoteDirectory);
+		List<F> filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files);
+		List<AbstractFileInfo<F>> fileInfoList = asFileInfoList(filteredFiles);
+		Iterator<AbstractFileInfo<F>> iterator = fileInfoList.iterator();
+		while (iterator.hasNext()) {
+			AbstractFileInfo<F> next = iterator.next();
+			if (next.isDirectory()) {
+				iterator.remove();
+			}
+			else {
+				next.setRemoteDirectory(remoteDirectory);
+			}
+		}
+		if (this.comparator != null) {
+			Collections.sort(fileInfoList, this.comparator);
+		}
+		this.toBeReceived.addAll(fileInfoList);
+	}
+
+	abstract protected List<AbstractFileInfo<F>> asFileInfoList(Collection<F> files);
+
+}

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.file.remote;
 
+import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.messaging.Message;
 
@@ -117,6 +118,13 @@ public interface RemoteFileOperations<F> {
 	 */
 	void rename(String fromPath, String toPath);
 
+	/**
+	 * List the files at the remote path.
+	 * @param path the path.
+	 * @return the list.
+	 */
+	F[] list(String path);
+
 	/**
 	 * Execute the callback's doInSession method after obtaining a session.
 	 * Reliably closes the session when the method exits.
@@ -141,4 +149,12 @@ public interface RemoteFileOperations<F> {
 	 */
 	<T, C> T executeWithClient(ClientCallback<C, T> callback);
 
+	/**
+	 * Obtain a raw Session object. User must close the session when it is no longer
+	 * needed.
+	 * @return a session.
+	 * @since 4.3
+	 */
+	Session<F> getSesssion();
+
 }

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

@@ -401,6 +401,24 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations<F>, Initializ
 		});
 	}
 
+
+	@Override
+	public F[] list(final String path) {
+		return this.execute(new SessionCallback<F, F[]>() {
+
+			@Override
+			public F[] doInSession(Session<F> session) throws IOException {
+				return session.list(path);
+			}
+
+		});
+	}
+
+	@Override
+	public Session<F> getSesssion() {
+		return this.sessionFactory.getSession();
+	}
+
 	@SuppressWarnings("rawtypes")
 	@Override
 	public <T> T execute(SessionCallback<F, T> callback) {

spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java

@@ -35,6 +35,7 @@ import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.FileListFilter;
@@ -586,7 +587,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		return getMessageBuilderFactory().withPayload(payload)
 				.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
 				.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
-				.setHeader(FileHeaders.REMOTE_SESSION, session)
+				.setHeader("file_remoteSession", session) // TODO: remove in 5.0
+				.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
 				.build();
 	}
 
@@ -1013,6 +1015,9 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			else if (e instanceof IOException) {
 				throw (IOException) e;
 			}
+			else {
+				throw new MessagingException("Failed to process MGET on first file", e);
+			}
 		}
 		return files;
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.file.remote.session;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -30,7 +31,7 @@ import java.io.OutputStream;
  * @author Gary Russell
  * @since 2.0
  */
-public interface Session<F> {
+public interface Session<F> extends Closeable {
 
 	boolean remove(String path) throws IOException;
 
@@ -62,6 +63,7 @@ public interface Session<F> {
 
 	void rename(String pathFrom, String pathTo) throws IOException;
 
+	@Override
 	void close();
 
 	boolean isOpen();

spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.file.splitter;
 
 import java.io.BufferedReader;
+import java.io.Closeable;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
@@ -33,6 +34,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.splitter.FileSplitter.FileMarker.Mark;
 import org.springframework.integration.splitter.AbstractMessageSplitter;
@@ -158,7 +160,23 @@ public class FileSplitter extends AbstractMessageSplitter {
 			return message;
 		}
 
-		final BufferedReader bufferedReader = new BufferedReader(reader);
+		final BufferedReader bufferedReader = new BufferedReader(reader) {
+
+			@Override
+			public void close() throws IOException {
+				try {
+					super.close();
+				}
+				finally {
+					Closeable closeableResource = new IntegrationMessageHeaderAccessor(message).getCloseableResource();
+					if (closeableResource != null) {
+						closeableResource.close();
+					}
+				}
+			}
+
+		};
+
 		Iterator<Object> iterator = new Iterator<Object>() {
 
 			boolean markers = FileSplitter.this.markers;

spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd

@@ -833,15 +833,11 @@ Only files matching this regular expression will be picked up by this adapter.
 	</xsd:simpleType>
 
 	<xsd:attributeGroup name="remoteOutboundAttributeGroup">
-		<xsd:attribute name="remote-directory-expression"
-			type="xsd:string">
+		<xsd:attribute name="charset" type="xsd:string" default="UTF-8">
 			<xsd:annotation>
 				<xsd:documentation>
-					Specify a SpEL expression which
-					will be used to evaluate the directory
-					path to where the files will be transferred
-					(e.g., "headers.['remote_dir'] +
-					'/myTransfers'");
+					Allows you to specify Charset (e.g., US-ASCII, ISO-8859-1, UTF-8). [UTF-8] is default -
+					used when converting String payloads to bytes.
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>

spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java

@@ -0,0 +1,177 @@
+/*
+ * Copyright 2016 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;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Abstract base class for tests requiring remote file servers, e.g. (S)FTP.
+ *
+ * @author Gary Russell
+ * @since 4.3
+ *
+ */
+public abstract class RemoteFileTestSupport {
+
+	protected static int port;
+
+	@ClassRule
+	public static final TemporaryFolder remoteTemporaryFolder = new TemporaryFolder();
+
+	@ClassRule
+	public static final TemporaryFolder localTemporaryFolder = new TemporaryFolder();
+
+	protected volatile File sourceRemoteDirectory;
+
+	protected volatile File targetRemoteDirectory;
+
+	protected volatile File sourceLocalDirectory;
+
+	protected volatile File targetLocalDirectory;
+
+	public File getSourceRemoteDirectory() {
+		return sourceRemoteDirectory;
+	}
+
+	public File getTargetRemoteDirectory() {
+		return targetRemoteDirectory;
+	}
+
+	public File getSourceLocalDirectory() {
+		return sourceLocalDirectory;
+	}
+
+	public File getTargetLocalDirectory() {
+		return targetLocalDirectory;
+	}
+
+	/**
+	 * Default implementation creates the following folder structures:
+	 *
+	 * <pre class="code">
+	 *  $ tree remoteSource/
+	 *  remoteSource/
+	 *  ├── remoteSource1.txt - contains 'source1'
+	 *  ├── remoteSource2.txt - contains 'source2'
+	 *  ├── subRemoteSource
+	 *      ├── subRemoteSource1.txt - contains 'subSource1'
+	 *  remoteTarget/
+	 *  $ tree localSource/
+	 *  localSource/
+	 *  ├── localSource1.txt - contains 'local1'
+	 *  ├── localSource2.txt - contains 'local2'
+	 *  ├── subLocalSource
+	 *      ├── subLocalSource1.txt - contains 'subLocal1'
+	 *  localTarget/
+	 * </pre>
+	 *
+	 * The intent is tests retrieve from remoteSource and verify arrival in localTarget or send from localSource and verify
+	 * arrival in remoteTarget.
+	 * <p>
+	 * Subclasses can change 'remote' in these names by overriding {@link #prefix()} or override this method completely to
+	 * create a different structure.
+	 * <p>
+	 * While a single server exists for all tests, the directory structure is rebuilt for each test.
+	 * @throws IOException IO Exception.
+	 */
+	@Before
+	public void setupFolders() throws IOException {
+		String prefix = prefix();
+		recursiveDelete(new File(remoteTemporaryFolder.getRoot(), prefix + "Source"));
+		this.sourceRemoteDirectory = remoteTemporaryFolder.newFolder(prefix + "Source");
+		recursiveDelete(new File(remoteTemporaryFolder.getRoot(), prefix + "Target"));
+		this.targetRemoteDirectory = remoteTemporaryFolder.newFolder(prefix + "Target");
+		recursiveDelete(new File(localTemporaryFolder.getRoot(), "localSource"));
+		this.sourceLocalDirectory = localTemporaryFolder.newFolder("localSource");
+		recursiveDelete(new File(localTemporaryFolder.getRoot(), "localTarget"));
+		this.targetLocalDirectory = localTemporaryFolder.newFolder("localTarget");
+
+		File file = new File(this.sourceRemoteDirectory, " " + prefix + "Source1.txt");
+		file.createNewFile();
+		FileOutputStream fos = new FileOutputStream(file);
+		fos.write("source1".getBytes());
+		fos.close();
+		file = new File(this.sourceRemoteDirectory, prefix + "Source2.txt");
+		file.createNewFile();
+		fos = new FileOutputStream(file);
+		fos.write("source2".getBytes());
+		fos.close();
+		String camelCasePrefix = camelCase(prefix);
+		File subSourceDirectory = new File(this.sourceRemoteDirectory, "sub" + camelCasePrefix + "Source");
+		subSourceDirectory.mkdir();
+		file = new File(subSourceDirectory, "sub" + camelCasePrefix + "Source1.txt");
+		file.createNewFile();
+		fos = new FileOutputStream(file);
+		fos.write("subSource1".getBytes());
+		fos.close();
+		file = new File(sourceLocalDirectory, "localSource1.txt");
+		file.createNewFile();
+		fos = new FileOutputStream(file);
+		fos.write("local1".getBytes());
+		fos.close();
+		file = new File(sourceLocalDirectory, "localSource2.txt");
+		file.createNewFile();
+		fos = new FileOutputStream(file);
+		fos.write("local2".getBytes());
+		fos.close();
+		File subSourceLocalDirectory = new File(this.sourceLocalDirectory, "subLocalSource");
+		subSourceLocalDirectory.mkdir();
+		file = new File(subSourceLocalDirectory, "subLocalSource1.txt");
+		file.createNewFile();
+		fos = new FileOutputStream(file);
+		fos.write("subLocal1".getBytes());
+		fos.close();
+	}
+
+	private String camelCase(String prefix) {
+		char[] chars = prefix.toCharArray();
+		chars[0] &= 0xdf;
+		return new String(chars);
+	}
+
+	public void recursiveDelete(File file) {
+		if (file != null && file.exists()) {
+			File[] files = file.listFiles();
+			if (files != null) {
+				for (File fyle : files) {
+					if (fyle.isDirectory()) {
+						recursiveDelete(fyle);
+					}
+					else {
+						fyle.delete();
+					}
+				}
+			}
+			file.delete();
+		}
+	}
+
+	/**
+	 * Prefix for directory/file structure; default 'remote'.
+	 * @return the prefix.
+	 */
+	protected String prefix() {
+		return "remote";
+	}
+
+}

spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java

@@ -0,0 +1,221 @@
+/*
+ * Copyright 2016 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.remote.session.Session;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.splitter.FileSplitter;
+import org.springframework.integration.transformer.StreamTransformer;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Gary Russell
+ * @since 4.3
+ *
+ */
+public class StreamingInboundTests {
+
+	private final StreamTransformer transformer = new StreamTransformer();
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testAllData() throws Exception {
+		Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null);
+		streamer.setBeanFactory(mock(BeanFactory.class));
+		streamer.setRemoteDirectory("/foo");
+		streamer.afterPropertiesSet();
+		Message<byte[]> received = (Message<byte[]>) this.transformer.transform(streamer.receive());
+		assertEquals("foo\nbar", new String(received.getPayload()));
+		assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+		assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE));
+
+		verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource()).close();
+
+		received = (Message<byte[]>) this.transformer.transform(streamer.receive());
+		assertEquals("baz\nqux", new String(received.getPayload()));
+		assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+		assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE));
+
+		verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource()).close();
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testLineByLine() throws Exception {
+		Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null);
+		streamer.setBeanFactory(mock(BeanFactory.class));
+		streamer.setRemoteDirectory("/foo");
+		streamer.afterPropertiesSet();
+		QueueChannel out = new QueueChannel();
+		FileSplitter splitter = new FileSplitter();
+		splitter.setBeanFactory(mock(BeanFactory.class));
+		splitter.setOutputChannel(out);
+		splitter.afterPropertiesSet();
+		Message<InputStream> receivedStream = streamer.receive();
+		splitter.handleMessage(receivedStream);
+		Message<byte[]> received = (Message<byte[]>) out.receive(0);
+		assertEquals("foo", received.getPayload());
+		assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+		assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE));
+		received = (Message<byte[]>) out.receive(0);
+		assertEquals("bar", received.getPayload());
+		assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+		assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE));
+		assertNull(out.receive(0));
+
+		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource()).close();
+
+		receivedStream = streamer.receive();
+		splitter.handleMessage(receivedStream);
+		received = (Message<byte[]>) out.receive(0);
+		assertEquals("baz", received.getPayload());
+		assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+		assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE));
+		received = (Message<byte[]>) out.receive(0);
+		assertEquals("qux", received.getPayload());
+		assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+		assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE));
+		assertNull(out.receive(0));
+
+		verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource()).close();
+	}
+
+	public static class Streamer extends AbstractRemoteFileStreamingMessageSource<String> {
+
+		protected Streamer(RemoteFileTemplate<String> template, Comparator<AbstractFileInfo<String>> comparator) {
+			super(template, comparator);
+		}
+
+		@Override
+		public String getComponentType() {
+			return "Streamer";
+		}
+
+		@Override
+		protected List<AbstractFileInfo<String>> asFileInfoList(Collection<String> files) {
+			List<AbstractFileInfo<String>> infos = new ArrayList<AbstractFileInfo<String>>();
+			for (String file : files) {
+				infos.add(new StringFileInfo(file));
+			}
+			return infos;
+		}
+
+	}
+
+	public static class StringFileInfo extends AbstractFileInfo<String> {
+
+		private final String name;
+
+		private StringFileInfo(String name) {
+			this.name = name;
+		}
+
+		@Override
+		public boolean isDirectory() {
+			return false;
+		}
+
+		@Override
+		public boolean isLink() {
+			return false;
+		}
+
+		@Override
+		public long getSize() {
+			return 0;
+		}
+
+		@Override
+		public long getModified() {
+			return 0;
+		}
+
+		@Override
+		public String getFilename() {
+			return this.name.substring(this.name.lastIndexOf("/") + 1);
+		}
+
+		@Override
+		public String getPermissions() {
+			return null;
+		}
+
+		@Override
+		public String getFileInfo() {
+			return null;
+		}
+
+	}
+
+	public static class StringRemoteFileTemplate extends RemoteFileTemplate<String> {
+
+		public StringRemoteFileTemplate(SessionFactory<String> sessionFactory) {
+			super(sessionFactory);
+		}
+
+	}
+
+	public static class StringSessionFactory implements SessionFactory<String> {
+
+		@SuppressWarnings("unchecked")
+		@Override
+		public Session<String> getSession() {
+			try {
+				Session<String> session = mock(Session.class);
+				willReturn(new String[] { "/foo/foo", "/foo/bar" }).given(session).list("/foo");
+				ByteArrayInputStream foo = new ByteArrayInputStream("foo\nbar".getBytes());
+				ByteArrayInputStream bar = new ByteArrayInputStream("baz\nqux".getBytes());
+				willReturn(foo).given(session).readRaw("/foo/foo");
+				willReturn(bar).given(session).readRaw("/foo/bar");
+
+				willReturn(new String[] { "/bar/foo", "/bar/bar" }).given(session).list("/bar");
+				ByteArrayInputStream foo2 = new ByteArrayInputStream("foo\r\nbar".getBytes());
+				ByteArrayInputStream bar2 = new ByteArrayInputStream("baz\r\nqux".getBytes());
+				willReturn(foo2).given(session).readRaw("/bar/foo");
+				willReturn(bar2).given(session).readRaw("/bar/bar");
+
+				given(session.finalizeRaw()).willReturn(true);
+				return session;
+			}
+			catch (Exception e) {
+				throw new RuntimeException("failed to mock session", e);
+			}
+		}
+
+	}
+
+}