INT-2492 Add (S)FTP Support for mget

No options supported; send a payload of /* or * to
mget all files in the root directory.

Payload is a list of File objects representing files
copied to the 'local-directory'.

INT-2492 Polishing

Fix path issues; suppress LIST when GETing files from NLST.

INT-2492 Polishing

Add ExtendedSession so CachingConnectionFactory works.

INT-2492 SFTP, Docs

INT-2492 Polishing

INT-2492 Polishing - olegz PR Review

INT-2492 Polishing - EOL Note for ExtendedSession

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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,6 +35,7 @@ import org.springframework.integration.MessagingException;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.AbstractFileInfo;
+import org.springframework.integration.file.remote.session.ExtendedSession;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -42,6 +43,7 @@ import org.springframework.integration.handler.ExpressionEvaluatingMessageProces
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Base class for Outbound Gateways that perform remote file operations.
@@ -61,7 +63,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 
 	public static final String COMMAND_RM = "rm";
 
-	protected Set<String> options = new HashSet<String>();
+	public static final String COMMAND_MGET = "mget";
 
 	public static final String OPTION_NAME_ONLY = "-1";
 
@@ -75,8 +77,15 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 
 	public static final String OPTION_PRESERVE_TIMESTAMP = "-P";
 
+	public static final String OPTION_EXCEPTION_WHEN_EMPTY = "-x";
+
+	private final Set<String> supportedCommands = new HashSet<String>(Arrays.asList(
+			COMMAND_LS, COMMAND_GET, COMMAND_RM, COMMAND_MGET));
+
 	private final ExpressionEvaluatingMessageProcessor<String> processor;
 
+	protected volatile Set<String> options = new HashSet<String>();
+
 	private volatile String remoteFileSeparator = "/";
 
 	private volatile File localDirectory;
@@ -149,12 +158,17 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	protected void onInit() {
 		super.onInit();
 		Assert.notNull(this.command, "command must not be null");
-		Assert.isTrue(COMMAND_LS.equals(this.command) || COMMAND_GET.equals(this.command) ||
-					  COMMAND_RM.equals(this.command),
-				"command must be one of ls, get, rm");
-		if (COMMAND_RM.equals(this.command)) {
-			Assert.isNull(this.filter, "Filters are not supported with the rm command");
-		} else if (COMMAND_GET.equals(this.command)) {
+		Assert.isTrue(
+				this.supportedCommands.contains(this.command),
+				"command must be one of "
+						+ StringUtils
+								.collectionToCommaDelimitedString(this.supportedCommands));
+		// NOTE: Filter also not used on GET, need to correct in 2.2.
+		if (COMMAND_RM.equals(this.command) || COMMAND_MGET.equals(this.command)) {
+			Assert.isNull(this.filter, "Filters are not supported with the rm and mget commands");
+		}
+		if (COMMAND_GET.equals(this.command)
+				|| COMMAND_MGET.equals(this.command)) {
 			Assert.notNull(this.localDirectory, "localDirectory must not be null");
 			try {
 				if (!this.localDirectory.exists()) {
@@ -187,35 +201,54 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		try {
 			if (COMMAND_LS.equals(this.command)) {
 				String dir = this.processor.processMessage(requestMessage);
-				if (!dir.endsWith("/")) {
-					dir += "/";
+				if (!dir.endsWith(this.remoteFileSeparator)) {
+					dir += this.remoteFileSeparator;
 				}
-				return MessageBuilder.withPayload(ls(session, dir))
+				List<?> payload = ls(session, dir);
+				return MessageBuilder.withPayload(payload)
 					.setHeader(FileHeaders.REMOTE_DIRECTORY, dir)
 					.build();
-			} else if (COMMAND_GET.equals(this.command)) {
+			}
+			else if (COMMAND_GET.equals(this.command)) {
 				String remoteFilePath =  this.processor.processMessage(requestMessage);
 				String remoteFilename = getRemoteFilename(remoteFilePath);
 				String remoteDir = remoteFilePath.substring(0, remoteFilePath.indexOf(remoteFilename));
 				if (remoteDir.length() == 0) {
-					remoteDir = "/";
+					remoteDir = this.remoteFileSeparator;
 				}
-				return MessageBuilder.withPayload(get(session, remoteFilePath, remoteFilename))
+				File payload = get(session, remoteFilePath, remoteFilename, true);
+				return MessageBuilder.withPayload(payload)
 					.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
 					.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
 					.build();
-			} else if (COMMAND_RM.equals(this.command)) {
+			}
+			else if (COMMAND_MGET.equals(this.command)) {
 				String remoteFilePath =  this.processor.processMessage(requestMessage);
 				String remoteFilename = getRemoteFilename(remoteFilePath);
 				String remoteDir = remoteFilePath.substring(0, remoteFilePath.indexOf(remoteFilename));
 				if (remoteDir.length() == 0) {
-					remoteDir = "/";
+					remoteDir = this.remoteFileSeparator;
 				}
-				return MessageBuilder.withPayload(rm(session, remoteFilePath))
+				List<File> payload = mGet(session, remoteDir, remoteFilename);
+				return MessageBuilder.withPayload(payload)
 					.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
 					.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
 					.build();
-			} else {
+			}
+			else if (COMMAND_RM.equals(this.command)) {
+				String remoteFilePath =  this.processor.processMessage(requestMessage);
+				String remoteFilename = getRemoteFilename(remoteFilePath);
+				String remoteDir = remoteFilePath.substring(0, remoteFilePath.indexOf(remoteFilename));
+				if (remoteDir.length() == 0) {
+					remoteDir = this.remoteFileSeparator;
+				}
+				boolean payload = rm(session, remoteFilePath);
+				return MessageBuilder.withPayload(payload)
+					.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
+					.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
+					.build();
+			}
+			else {
 				return null;
 			}
 		} catch (IOException e) {
@@ -297,11 +330,14 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	 * @param remoteFilePath
 	 * @throws IOException
 	 */
-	protected File get(Session<F> session, String remoteFilePath, String remoteFilename)
+	protected File get(Session<F> session, String remoteFilePath, String remoteFilename, boolean lsFirst)
 			throws IOException {
-		F[] files = session.list(remoteFilePath);
-		if (files.length != 1 || isDirectory(files[0]) || isLink(files[0])) {
-			throw new MessagingException(remoteFilePath + " is not a file");
+		F[] files = null;
+		if (lsFirst) {
+			files = session.list(remoteFilePath);
+			if (files.length != 1 || isDirectory(files[0]) || isLink(files[0])) {
+				throw new MessagingException(remoteFilePath + " is not a file");
+			}
 		}
 		File localFile = new File(this.localDirectory, remoteFilename);
 		if (!localFile.exists()) {
@@ -329,7 +365,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			if (!tempFile.renameTo(localFile)) {
 				throw new MessagingException("Failed to rename local file");
 			}
-			if (this.options.contains(OPTION_PRESERVE_TIMESTAMP)) {
+			if (lsFirst && this.options.contains(OPTION_PRESERVE_TIMESTAMP)) {
 				localFile.setLastModified(getModified(files[0]));
 			}
 			return localFile;
@@ -339,6 +375,36 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		}
 	}
 
+	protected List<File> mGet(Session<F> session, String remoteDirectory,
+			String remoteFilename) throws IOException {
+		Assert.isInstanceOf(ExtendedSession.class, session, "mget failed - ");
+		String path = fixPath(remoteDirectory, remoteFilename);
+		String[] fileNames = ((ExtendedSession<F>) session).listNames(path);
+		if (fileNames.length == 0 && this.options.contains(OPTION_EXCEPTION_WHEN_EMPTY)) {
+			throw new MessagingException("No files found at " + remoteDirectory
+					+ " with pattern " + remoteFilename);
+		}
+		List<File> files = new ArrayList<File>();
+		for (String fileName : fileNames) {
+			files.add(this.get(session, fixPath(remoteDirectory, fileName), fileName, false));
+		}
+		return files;
+	}
+
+	private String fixPath(String remoteDirectory, String remoteFilename) {
+		String path;
+		if (this.remoteFileSeparator.equals(remoteDirectory)) {
+			path = remoteFilename;
+		}
+		else if (remoteDirectory.endsWith(this.remoteFileSeparator)) {
+			path = remoteDirectory + remoteFilename;
+		}
+		else {
+			path = remoteDirectory + this.remoteFileSeparator + remoteFilename;
+		}
+		return path;
+	}
+
 	/**
 	 * @param remoteFilePath
 	 */

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

@@ -20,7 +20,6 @@ import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 

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

@@ -25,6 +25,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.integration.util.UpperBound;
+import org.springframework.util.Assert;
 
 /**
  * A {@link SessionFactory} implementation that caches Sessions for reuse without
@@ -34,6 +35,7 @@ import org.springframework.integration.util.UpperBound;
  * @author Josh Long
  * @author Oleg Zhurakousky
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.0
  */
 public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBean {
@@ -115,7 +117,7 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 	}
 
 
-	private class CachedSession implements Session<F> {
+	private class CachedSession implements ExtendedSession<F> {
 
 		private final Session<F> targetSession;
 
@@ -162,6 +164,11 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 		public boolean exists(String path) throws IOException{
 			return this.targetSession.exists(path);
 		}
+
+		public String[] listNames(String path) throws IOException {
+			Assert.isInstanceOf(ExtendedSession.class, this.targetSession, "mget failed - ");
+			return ((ExtendedSession<F>) this.targetSession).listNames(path);
+		}
 	}
 
 }

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.file.remote.session;
+
+import java.io.IOException;
+
+/**
+ * <b>Temporary extension</b> to {@link Session} to avoid a breaking change
+ * in a point release - merge with Session in 2.2.
+ * <p>
+ * <b>**NOTE**</b> This interface will be <b>removed</b> (not deprecated) in 2.2.
+ * It exists purely to avoid existing user implementations of Session from
+ * failing to compile if we had added the new method to that interface.
+ * <p>Any user implementations of ExtendedSession will need to be refactored to
+ * implement Session in 2.2.
+ * @author Gary Russell
+ * @since 2.1.1
+ *
+ */
+public interface ExtendedSession<T> extends Session<T> {
+
+	/**
+	 * Returns an array of Strings containing just the names of
+	 * the remote files at path. Will move to {@link Session}
+	 * in 2.2.
+	 * @param path The path
+	 * @return The list of names
+	 * @throws IOException
+	 */
+	String[] listNames(String path) throws IOException;
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -34,9 +34,11 @@ import java.util.List;
 
 import org.junit.Test;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessagingException;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.AbstractSimplePatternFileListFilter;
 import org.springframework.integration.file.remote.AbstractFileInfo;
+import org.springframework.integration.file.remote.session.ExtendedSession;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.message.GenericMessage;
@@ -53,6 +55,14 @@ public class RemoteFileOutboundGatewayTests {
 	private String tmpDir = System.getProperty("java.io.tmpdir");
 
 	
+	@Test(expected=IllegalArgumentException.class)
+	public void testBad() throws Exception {
+		SessionFactory sessionFactory = mock(SessionFactory.class);
+		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
+			(sessionFactory, "bad", "payload");
+		gw.afterPropertiesSet();
+	}
+
 	@Test
 	public void testLs() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
@@ -72,6 +82,163 @@ public class RemoteFileOutboundGatewayTests {
 				out.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
 	}
 
+	@Test(expected=IllegalArgumentException.class)
+	public void testMGetSession() throws Exception {
+		SessionFactory sessionFactory = mock(SessionFactory.class);
+		Session session = mock(Session.class);
+		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
+			(sessionFactory, "mget", "payload");
+		when(sessionFactory.getSession()).thenReturn(session);
+		gw.setLocalDirectory(new File(this.tmpDir ));
+		gw.afterPropertiesSet();
+		gw.handleRequestMessage(new GenericMessage<String>("testremote/*"));
+	}
+
+	@Test
+	public void testMGetWild() throws Exception {
+		SessionFactory sessionFactory = mock(SessionFactory.class);
+		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
+			(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 ExtendedSession() {
+			public boolean remove(String path) throws IOException {
+				return false;
+			}
+			public Object[] list(String path) throws IOException {
+				return null;
+			}
+			public void read(String source, OutputStream outputStream)
+					throws IOException {
+				outputStream.write("testData".getBytes());
+			}
+			public void write(InputStream inputStream, String destination)
+					throws IOException {
+			}
+			public boolean mkdir(String directory) throws IOException {
+				return false;
+			}
+			public void rename(String pathFrom, String pathTo)
+					throws IOException {
+			}
+			public void close() {
+			}
+			public boolean isOpen() {
+				return false;
+			}
+			public boolean exists(String path) throws IOException {
+				return false;
+			}
+			public String[] listNames(String path) throws IOException {
+				return new String[] {"f1", "f2"};
+			}
+		});
+		@SuppressWarnings("unchecked")
+		Message<List<File>> out = (Message<List<File>>) gw
+				.handleRequestMessage(new GenericMessage<String>("testremote/*"));
+		assertEquals(2, out.getPayload().size());
+		assertEquals("f1", out.getPayload().get(0).getName());
+		assertEquals("f2", out.getPayload().get(1).getName());
+		assertEquals("testremote/",
+				out.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+	}
+
+	@Test
+	public void testMGetSingle() throws Exception {
+		SessionFactory sessionFactory = mock(SessionFactory.class);
+		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
+			(sessionFactory, "mget", "payload");
+		gw.setLocalDirectory(new File(this.tmpDir ));
+		gw.afterPropertiesSet();
+		new File(this.tmpDir + "/f1").delete();
+		when(sessionFactory.getSession()).thenReturn(new ExtendedSession() {
+			public boolean remove(String path) throws IOException {
+				return false;
+			}
+			public Object[] list(String path) throws IOException {
+				return null;
+			}
+			public void read(String source, OutputStream outputStream)
+					throws IOException {
+				outputStream.write("testData".getBytes());
+			}
+			public void write(InputStream inputStream, String destination)
+					throws IOException {
+			}
+			public boolean mkdir(String directory) throws IOException {
+				return false;
+			}
+			public void rename(String pathFrom, String pathTo)
+					throws IOException {
+			}
+			public void close() {
+			}
+			public boolean isOpen() {
+				return false;
+			}
+			public boolean exists(String path) throws IOException {
+				return false;
+			}
+			public String[] listNames(String path) throws IOException {
+				return new String[] {"f1"};
+			}
+		});
+		@SuppressWarnings("unchecked")
+		Message<List<File>> out = (Message<List<File>>) gw
+				.handleRequestMessage(new GenericMessage<String>("testremote/f1"));
+		assertEquals(1, out.getPayload().size());
+		assertEquals("f1", out.getPayload().get(0).getName());
+		assertEquals("testremote/",
+				out.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
+	}
+
+	@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 ));
+		gw.setOptions("   -x   ");
+		gw.afterPropertiesSet();
+		new File(this.tmpDir + "/f1").delete();
+		new File(this.tmpDir + "/f2").delete();
+		when(sessionFactory.getSession()).thenReturn(new ExtendedSession() {
+			public boolean remove(String path) throws IOException {
+				return false;
+			}
+			public Object[] list(String path) throws IOException {
+				return null;
+			}
+			public void read(String source, OutputStream outputStream)
+					throws IOException {
+				outputStream.write("testData".getBytes());
+			}
+			public void write(InputStream inputStream, String destination)
+					throws IOException {
+			}
+			public boolean mkdir(String directory) throws IOException {
+				return false;
+			}
+			public void rename(String pathFrom, String pathTo)
+					throws IOException {
+			}
+			public void close() {
+			}
+			public boolean isOpen() {
+				return false;
+			}
+			public boolean exists(String path) throws IOException {
+				return false;
+			}
+			public String[] listNames(String path) throws IOException {
+				return new String[0];
+			}
+		});
+		gw.handleRequestMessage(new GenericMessage<String>("testremote/*"));
+	}
+
 	/**
 	 * @return
 	 */