INT-3088 (S)FTP Outbound Gateway - PUT and MPUT

- Core support in file module
- FTP Parser and Test

https://jira.springsource.org/browse/INT-3088

INT-3088 Add SFTP Support for PUT, MPUT

INT-3088 Polishing - PR Comments

INT-3088 Docbook For PUT/MPUT

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapterParserTests-context-fail.xml

@@ -7,7 +7,7 @@
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
 		http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd">
 
-	<bean id="sftpSessionFactory" class="org.springframework.integration.sftp.session.SimpleSftpSessionFactory">
+	<bean id="sftpSessionFactory" class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
 		<property name="host" value="localhost"/>
 		<property name="knownHosts" value="local, foo.com, bar.foo"/>
 		<property name="privateKey" value="classpath:org/springframework/integration/sftp/config/sftpTest"/>

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapterParserTests.java

@@ -21,11 +21,14 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import java.util.Iterator;
 import java.util.Set;
 
+import org.hamcrest.Matchers;
 import org.junit.Test;
 
 import org.springframework.beans.factory.BeanDefinitionStoreException;
@@ -131,9 +134,15 @@ public class OutboundChannelAdapterParserTests {
 		assertEquals(1, adviceCalled);
 	}
 
-	@Test(expected=BeanDefinitionStoreException.class)
+	@Test
 	public void testFailWithRemoteDirAndExpression(){
-		new ClassPathXmlApplicationContext("OutboundChannelAdapterParserTests-context-fail.xml", this.getClass());
+		try {
+			new ClassPathXmlApplicationContext("OutboundChannelAdapterParserTests-context-fail.xml", this.getClass());
+			fail("Exception expected");
+		}
+		catch (BeanDefinitionStoreException e) {
+			assertThat(e.getMessage(), Matchers.containsString("Only one of 'remote-directory'"));
+		}
 
 	}
 

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests-context.xml

@@ -29,6 +29,7 @@
 		command-options="-1 -f"
 		expression="payload"
 		order="1"
+		mput-regex=".*"
 		/>
 
 	<bean id="fooString" class="java.lang.String">
@@ -49,6 +50,7 @@
 		order="2"
 		requires-reply="false"
 		local-filename-generator-expression="#remoteFileName.toUpperCase() + '.a' + @fooString"
+		mput-pattern="*"
 		/>
 
 	<int-sftp:outbound-gateway id="gateway3"
@@ -61,6 +63,30 @@
 		order="1"
 		/>
 
+	<int-sftp:outbound-gateway id="gateway4"
+		session-factory="csf"
+		request-channel="inbound1"
+		reply-channel="outbound"
+		command="mput"
+		expression="payload"
+		remote-directory="/foo"
+		remote-file-separator="X"
+		auto-create-directory="true"
+		remote-filename-generator="fileNameGenerator"
+		temporary-remote-directory="/bar"
+		rename-expression="'foo'"
+		order="1"
+		mput-filter="mputFilter"
+		/>
+
+	<bean id="fileNameGenerator" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.integration.file.FileNameGenerator"/>
+	</bean>
+
+	<bean id="mputFilter" class="org.springframework.integration.file.filters.RegexPatternFileListFilter">
+		<constructor-arg value="(.*1.txt|sub*)"/>
+	</bean>
+
 	<int-sftp:outbound-gateway id="advised"
 		local-directory="local-test-dir"
 		session-factory="sf"

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java

@@ -18,18 +18,25 @@ package org.springframework.integration.sftp.config;
 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 java.lang.reflect.Method;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.expression.Expression;
 import org.springframework.integration.Message;
 import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.file.FileNameGenerator;
+import org.springframework.integration.file.filters.RegexPatternFileListFilter;
+import org.springframework.integration.file.filters.SimplePatternFileListFilter;
 import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.Command;
 import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.Option;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
@@ -62,16 +69,22 @@ public class SftpOutboundGatewayParserTests {
 	@Autowired
 	AbstractEndpoint gateway3;
 
+	@Autowired
+	AbstractEndpoint gateway4;
+
 	@Autowired
 	AbstractEndpoint advised;
 
+	@Autowired
+	FileNameGenerator generator;
+
 	private static volatile int adviceCalled;
 
 	@Test
 	public void testGateway1() {
 		SftpOutboundGateway gateway = TestUtils.getPropertyValue(gateway1,
 				"handler", SftpOutboundGateway.class);
-		assertEquals("X", TestUtils.getPropertyValue(gateway, "remoteFileSeparator"));
+		assertEquals("X", TestUtils.getPropertyValue(gateway, "remoteFileTemplate.remoteFileSeparator"));
 		assertNotNull(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory"));
 		assertNotNull(TestUtils.getPropertyValue(gateway, "outputChannel"));
 		assertEquals("local-test-dir", TestUtils.getPropertyValue(gateway, "localDirectoryExpression.literalValue"));
@@ -86,13 +99,14 @@ public class SftpOutboundGatewayParserTests {
 
 		Long sendTimeout = TestUtils.getPropertyValue(gateway, "messagingTemplate.sendTimeout", Long.class);
 		assertEquals(Long.valueOf(777), sendTimeout);
+		assertThat(TestUtils.getPropertyValue(gateway, "mputFilter"), Matchers.instanceOf(RegexPatternFileListFilter.class));
 	}
 
 	@Test
 	public void testGateway2() throws Exception {
 		SftpOutboundGateway gateway = TestUtils.getPropertyValue(gateway2,
 				"handler", SftpOutboundGateway.class);
-		assertEquals("X", TestUtils.getPropertyValue(gateway, "remoteFileSeparator"));
+		assertEquals("X", TestUtils.getPropertyValue(gateway, "remoteFileTemplate.remoteFileSeparator"));
 		assertNotNull(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory"));
 		assertTrue(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory") instanceof CachingSessionFactory);
 		assertNotNull(TestUtils.getPropertyValue(gateway, "outputChannel"));
@@ -118,6 +132,7 @@ public class SftpOutboundGatewayParserTests {
 			}
 		});
 		assertEquals("FOO.afoo", genMethod.get().invoke(gateway, new GenericMessage<String>(""), "foo"));
+		assertThat(TestUtils.getPropertyValue(gateway, "mputFilter"), Matchers.instanceOf(SimplePatternFileListFilter.class));
 	}
 
 	@Test
@@ -130,6 +145,24 @@ public class SftpOutboundGatewayParserTests {
 		assertEquals("'foo'", TestUtils.getPropertyValue(gateway, "renameProcessor.expression.expression"));
 	}
 
+	@Test
+	public void testGatewayMPut() {
+		SftpOutboundGateway gateway = TestUtils.getPropertyValue(gateway4,
+				"handler", SftpOutboundGateway.class);
+		assertNotNull(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory"));
+		assertNotNull(TestUtils.getPropertyValue(gateway, "outputChannel"));
+		assertEquals(Command.MPUT, TestUtils.getPropertyValue(gateway, "command"));
+		assertEquals("'foo'", TestUtils.getPropertyValue(gateway, "renameProcessor.expression.expression"));
+		assertThat(TestUtils.getPropertyValue(gateway, "mputFilter"), Matchers.instanceOf(RegexPatternFileListFilter.class));
+		assertSame(generator, TestUtils.getPropertyValue(gateway, "remoteFileTemplate.fileNameGenerator"));
+		assertEquals("/foo",
+				TestUtils.getPropertyValue(gateway, "remoteFileTemplate.directoryExpressionProcessor.expression", Expression.class)
+						.getExpressionString());
+		assertEquals("/bar",
+				TestUtils.getPropertyValue(gateway, "remoteFileTemplate.temporaryDirectoryExpressionProcessor.expression", Expression.class)
+						.getExpressionString());
+	}
+
 	@Test
 	public void advised() {
 		SftpOutboundGateway gateway = TestUtils.getPropertyValue(advised,

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

@@ -64,6 +64,42 @@
 							  local-filename-generator-expression="#remoteFileName.replaceFirst('ftpSource', 'localTarget')"
 							  reply-channel="output"/>
 
+	<int:channel id="inboundMPut"/>
+
+	<int-sftp:outbound-gateway session-factory="ftpSessionFactory"
+							  request-channel="inboundMPut"
+							  command="mput"
+							  auto-create-directory="true"
+							  filename-pattern="*.txt"
+							  expression="payload"
+							  remote-directory="sftpTarget"
+							  reply-channel="output"/>
+
+	<int:channel id="inboundMPutRecursive"/>
+
+	<int-sftp:outbound-gateway session-factory="ftpSessionFactory"
+							  request-channel="inboundMPutRecursive"
+							  command="mput"
+							  command-options="-R"
+							  auto-create-directory="true"
+							  filename-pattern="*.txt"
+							  expression="payload"
+							  remote-directory="sftpTarget"
+							  reply-channel="output"/>
+
+	<int:channel id="inboundMPutRecursiveFiltered"/>
+
+	<int-sftp:outbound-gateway session-factory="ftpSessionFactory"
+							  request-channel="inboundMPutRecursiveFiltered"
+							  command="mput"
+							  command-options="-R"
+							  mput-regex="(.*1.txt|sub.*)"
+							  auto-create-directory="true"
+							  filename-pattern="*.txt"
+							  expression="payload"
+							  remote-directory="sftpTarget"
+							  reply-channel="output"/>
+
 	<bean id="ftpSessionFactory" class="org.mockito.Mockito" factory-method="mock">
 		<constructor-arg value="org.springframework.integration.file.remote.session.SessionFactory" />
 	</bean>

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

@@ -16,6 +16,9 @@
 
 package org.springframework.integration.sftp.outbound;
 
+import static org.hamcrest.Matchers.anyOf;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
@@ -51,6 +54,8 @@ import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.sftp.session.SftpFileInfo;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.test.annotation.IfProfileValue;
+import org.springframework.test.annotation.ProfileValueUtils;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.FileCopyUtils;
@@ -98,6 +103,15 @@ public class SftpServerOutboundTests {
 	@Autowired
 	private DirectChannel inboundMGetRecursiveFiltered;
 
+	@Autowired
+	private DirectChannel inboundMPut;
+
+	@Autowired
+	private DirectChannel inboundMPutRecursive;
+
+	@Autowired
+	private DirectChannel inboundMPutRecursiveFiltered;
+
 	@Autowired
 	private SessionFactory<SftpFileInfo> sessionFactory;
 
@@ -109,7 +123,9 @@ public class SftpServerOutboundTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private void setUpMocksIfNeeded() throws IOException {
-		if (sessionFactory.toString().startsWith("Mock for")) {
+		String profile = ProfileValueUtils.retrieveProfileValueSource(this.getClass()).get("spring.profiles.active");
+		boolean usingMocks = profile == null || ! profile.startsWith("realSSH");
+		if (usingMocks) {
 			Session session = mock(Session.class);
 			when(sessionFactory.getSession()).thenReturn(session);
 			LsEntry entry1 = mock(LsEntry.class);
@@ -276,82 +292,151 @@ public class SftpServerOutboundTests {
 	 * Only runs with a real server (see class javadocs).
 	 */
 	@Test
+	@IfProfileValue(name="spring.profiles.active", value="realSSH")
 	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()));
+		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()));
+		baos = new ByteArrayOutputStream();
+		FileCopyUtils.copy(session.readRaw("sftpSource/sftpSource2.txt"), baos);
+		assertTrue(session.finalizeRaw());
+		assertEquals("source2", new String(baos.toByteArray()));
 
-			session.close();
-		}
+		session.close();
 	}
 
 	@Test
+	@IfProfileValue(name="spring.profiles.active", value="realSSHSharedSession")
 	public void testInt3047ConcurrentSharedSession() throws Exception {
-		if ("realSSHSharedSession".equals(System.getProperty("spring.profiles.active"))) {
-			final Session<?> session1 = this.sessionFactory.getSession();
-			final Session<?> session2 = this.sessionFactory.getSession();
-			final PipedInputStream pipe1 = new PipedInputStream();
-			PipedOutputStream out1 = new PipedOutputStream(pipe1);
-			final PipedInputStream pipe2 = new PipedInputStream();
-			PipedOutputStream out2 = new PipedOutputStream(pipe2);
-			final CountDownLatch latch1 = new CountDownLatch(1);
-			final CountDownLatch latch2 = new CountDownLatch(1);
-			Executors.newSingleThreadExecutor().execute(new Runnable() {
+		final Session<?> session1 = this.sessionFactory.getSession();
+		final Session<?> session2 = this.sessionFactory.getSession();
+		final PipedInputStream pipe1 = new PipedInputStream();
+		PipedOutputStream out1 = new PipedOutputStream(pipe1);
+		final PipedInputStream pipe2 = new PipedInputStream();
+		PipedOutputStream out2 = new PipedOutputStream(pipe2);
+		final CountDownLatch latch1 = new CountDownLatch(1);
+		final CountDownLatch latch2 = new CountDownLatch(1);
+		Executors.newSingleThreadExecutor().execute(new Runnable() {
 
-				@Override
-				public void run() {
-					try {
-						session1.write(pipe1, "foo.txt");
-					}
-					catch (IOException e) {
-						e.printStackTrace();
-					}
-					latch1.countDown();
+			@Override
+			public void run() {
+				try {
+					session1.write(pipe1, "foo.txt");
 				}
-			});
-			Executors.newSingleThreadExecutor().execute(new Runnable() {
-
-				@Override
-				public void run() {
-					try {
-						session2.write(pipe2, "bar.txt");
-					}
-					catch (IOException e) {
-						e.printStackTrace();
-					}
-					latch2.countDown();
+				catch (IOException e) {
+					e.printStackTrace();
 				}
-			});
+				latch1.countDown();
+			}
+		});
+		Executors.newSingleThreadExecutor().execute(new Runnable() {
 
-			out1.write('a');
-			out2.write('b');
-			out1.write('c');
-			out2.write('d');
-			out1.write('e');
-			out2.write('f');
-			out1.close();
-			out2.close();
-			assertTrue(latch1.await(10, TimeUnit.SECONDS));
-			assertTrue(latch2.await(10, TimeUnit.SECONDS));
-			ByteArrayOutputStream bos1 = new ByteArrayOutputStream();
-			ByteArrayOutputStream bos2 = new ByteArrayOutputStream();
-			session1.read("foo.txt", bos1);
-			session2.read("bar.txt", bos2);
-			assertEquals("ace", new String(bos1.toByteArray()));
-			assertEquals("bdf", new String(bos2.toByteArray()));
-			session1.remove("foo.txt");
-			session2.remove("bar.txt");
-			session1.close();
-			session2.close();
-		}
+			@Override
+			public void run() {
+				try {
+					session2.write(pipe2, "bar.txt");
+				}
+				catch (IOException e) {
+					e.printStackTrace();
+				}
+				latch2.countDown();
+			}
+		});
+
+		out1.write('a');
+		out2.write('b');
+		out1.write('c');
+		out2.write('d');
+		out1.write('e');
+		out2.write('f');
+		out1.close();
+		out2.close();
+		assertTrue(latch1.await(10, TimeUnit.SECONDS));
+		assertTrue(latch2.await(10, TimeUnit.SECONDS));
+		ByteArrayOutputStream bos1 = new ByteArrayOutputStream();
+		ByteArrayOutputStream bos2 = new ByteArrayOutputStream();
+		session1.read("foo.txt", bos1);
+		session2.read("bar.txt", bos2);
+		assertEquals("ace", new String(bos1.toByteArray()));
+		assertEquals("bdf", new String(bos2.toByteArray()));
+		session1.remove("foo.txt");
+		session2.remove("bar.txt");
+		session1.close();
+		session2.close();
+	}
+
+	@Test
+	@IfProfileValue(name="spring.profiles.active", value="realSSH")
+	public void testInt3088MPutNotRecursive() {
+		String dir = "sftpSource/";
+		this.inboundMGetRecursive.send(new GenericMessage<Object>(dir + "*"));
+		while (output.receive(0) != null) { }
+		this.inboundMPut.send(new GenericMessage<File>(new File("/tmp/sftpOutboundTests/sftpSource")));
+		@SuppressWarnings("unchecked")
+		Message<List<String>> out = (Message<List<String>>) this.output.receive(1000);
+		assertNotNull(out);
+		assertEquals(2, out.getPayload().size());
+		assertThat(out.getPayload().get(0),
+				not(equalTo(out.getPayload().get(1))));
+		assertThat(
+				out.getPayload().get(0),
+				anyOf(equalTo("sftpTarget/slocalTarget1.txt"), equalTo("sftpTarget/slocalTarget2.txt")));
+		assertThat(
+				out.getPayload().get(1),
+				anyOf(equalTo("sftpTarget/slocalTarget1.txt"), equalTo("sftpTarget/slocalTarget2.txt")));
+	}
+
+	@Test
+	@IfProfileValue(name="spring.profiles.active", value="realSSH")
+	public void testInt3088MPutRecursive() {
+		String dir = "sftpSource/";
+		this.inboundMGetRecursive.send(new GenericMessage<Object>(dir + "*"));
+		while (output.receive(0) != null) { }
+		this.inboundMPutRecursive.send(new GenericMessage<File>(new File("/tmp/sftpOutboundTests/sftpSource")));
+		@SuppressWarnings("unchecked")
+		Message<List<String>> out = (Message<List<String>>) this.output.receive(1000);
+		assertNotNull(out);
+		assertEquals(3, out.getPayload().size());
+		assertThat(out.getPayload().get(0),
+				not(equalTo(out.getPayload().get(1))));
+		assertThat(
+				out.getPayload().get(0),
+				anyOf(equalTo("sftpTarget/slocalTarget1.txt"), equalTo("sftpTarget/slocalTarget2.txt"),
+						equalTo("sftpTarget/subSftpSource/subSlocalTarget1.txt")));
+		assertThat(
+				out.getPayload().get(1),
+				anyOf(equalTo("sftpTarget/slocalTarget1.txt"), equalTo("sftpTarget/slocalTarget2.txt"),
+						equalTo("sftpTarget/subSftpSource/subSlocalTarget1.txt")));
+		assertThat(
+				out.getPayload().get(2),
+				anyOf(equalTo("sftpTarget/slocalTarget1.txt"), equalTo("sftpTarget/slocalTarget2.txt"),
+						equalTo("sftpTarget/subSftpSource/subSlocalTarget1.txt")));
+	}
+
+	@Test
+	@IfProfileValue(name="spring.profiles.active", value="realSSH")
+	public void testInt3088MPutRecursiveFiltered() {
+		String dir = "sftpSource/";
+		this.inboundMGetRecursive.send(new GenericMessage<Object>(dir + "*"));
+		while (output.receive(0) != null) { }
+		this.inboundMPutRecursiveFiltered.send(new GenericMessage<File>(new File("/tmp/sftpOutboundTests/sftpSource")));
+		@SuppressWarnings("unchecked")
+		Message<List<String>> out = (Message<List<String>>) this.output.receive(1000);
+		assertNotNull(out);
+		assertEquals(2, out.getPayload().size());
+		assertThat(out.getPayload().get(0),
+				not(equalTo(out.getPayload().get(1))));
+		assertThat(
+				out.getPayload().get(0),
+				anyOf(equalTo("sftpTarget/slocalTarget1.txt"), equalTo("sftpTarget/slocalTarget2.txt"),
+						equalTo("sftpTarget/subSftpSource/subSlocalTarget1.txt")));
+		assertThat(
+				out.getPayload().get(1),
+				anyOf(equalTo("sftpTarget/slocalTarget1.txt"), equalTo("sftpTarget/slocalTarget2.txt"),
+						equalTo("sftpTarget/subSftpSource/subSlocalTarget1.txt")));
 	}
 
 }