GH-3026: Fix chmod support for DSL

Fixes #3026

**Cherry-pick to `5.1.x`**

* Populate proper `FileTransferringMessageHandler` impl from DSL spec
implementations.
This way we are able to use a provided `chmod` from Java DSL
* Added `FileTransferringMessageHandlerSpec` ctor TODO
* Update SftpTests
* Code cleanup; `@Ignore` `SftpTests.testSftpOutboundFlowWithChmod()`
since it doesn't work properly on Windows

# Conflicts:
#	spring-integration-sftp/src/test/java/org/springframework/integration/sftp/dsl/SftpTests.java
#	spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java

Author: joaquinjsb

spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileTransferringMessageHandlerSpec.java

@@ -41,6 +41,7 @@
  * @param <S> the target {@link FileTransferringMessageHandlerSpec} implementation type.
  *
  * @author Artem Bilan
+ * @author Joaquin Santana
  *
  * @since 5.0
  */
@@ -52,6 +53,10 @@ public abstract class FileTransferringMessageHandlerSpec<F, S extends FileTransf
 
 	private DefaultFileNameGenerator defaultFileNameGenerator;
 
+	// TODO: should be refactored using generics in next release (breaking change), see PR-3080.
+	protected FileTransferringMessageHandlerSpec() {
+	}
+
 	protected FileTransferringMessageHandlerSpec(SessionFactory<F> sessionFactory) {
 		this.target = new FileTransferringMessageHandler<>(sessionFactory);
 	}
@@ -62,6 +67,7 @@ protected FileTransferringMessageHandlerSpec(RemoteFileTemplate<F> remoteFileTem
 
 	protected FileTransferringMessageHandlerSpec(RemoteFileTemplate<F> remoteFileTemplate,
 			FileExistsMode fileExistsMode) {
+
 		this.target = new FileTransferringMessageHandler<>(remoteFileTemplate, fileExistsMode);
 	}
 

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpMessageHandlerSpec.java

@@ -22,25 +22,28 @@
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.integration.ftp.outbound.FtpMessageHandler;
 
 /**
  * A {@link FileTransferringMessageHandlerSpec} for FTP.
  *
  * @author Artem Bilan
+ * @author Joaquin Santana
+ *
  * @since 5.0
  */
 public class FtpMessageHandlerSpec extends FileTransferringMessageHandlerSpec<FTPFile, FtpMessageHandlerSpec> {
 
 	FtpMessageHandlerSpec(SessionFactory<FTPFile> sessionFactory) {
-		super(sessionFactory);
+		this.target = new FtpMessageHandler(sessionFactory);
 	}
 
 	FtpMessageHandlerSpec(RemoteFileTemplate<FTPFile> remoteFileTemplate) {
-		super(remoteFileTemplate);
+		this.target = new FtpMessageHandler(remoteFileTemplate.getSessionFactory());
 	}
 
 	FtpMessageHandlerSpec(RemoteFileTemplate<FTPFile> remoteFileTemplate, FileExistsMode fileExistsMode) {
-		super(remoteFileTemplate, fileExistsMode);
+		this.target = new FtpMessageHandler(remoteFileTemplate, fileExistsMode);
 	}
 
 }

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java

@@ -74,6 +74,7 @@
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Joaquin Santana
  *
  * @since 5.0
  */
@@ -197,6 +198,30 @@ public void testFtpOutboundFlow() {
 		registration.destroy();
 	}
 
+	@Test
+	public void testFtpOutboundFlowWithChmod() {
+		IntegrationFlow flow = f -> f
+				.handle(Ftp.outboundAdapter(sessionFactory(), FileExistsMode.FAIL)
+						.useTemporaryFileName(false)
+						.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
+						.chmod(0644)
+						.remoteDirectory("ftpTarget"));
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		String fileName = "foo.file";
+		Message<ByteArrayInputStream> message = MessageBuilder
+				.withPayload(new ByteArrayInputStream("foo".getBytes(StandardCharsets.UTF_8)))
+				.setHeader(FileHeaders.FILENAME, fileName)
+				.build();
+		registration.getInputChannel().send(message);
+		RemoteFileTemplate<FTPFile> template = new RemoteFileTemplate<>(sessionFactory());
+		FTPFile[] files = template.execute(session ->
+				session.list(getTargetRemoteDirectory().getName() + "/" + fileName));
+		assertEquals(1, files.length);
+		assertEquals(3, files[0].getSize());
+
+		registration.destroy();
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testFtpMgetFlow() {

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpMessageHandlerSpec.java

@@ -20,26 +20,30 @@
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.integration.sftp.outbound.SftpMessageHandler;
+import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
 
 import com.jcraft.jsch.ChannelSftp;
 
 /**
  * @author Artem Bilan
+ * @author Joaquin Santana
+ *
  * @since 5.0
  */
 public class SftpMessageHandlerSpec
 		extends FileTransferringMessageHandlerSpec<ChannelSftp.LsEntry, SftpMessageHandlerSpec> {
 
 	SftpMessageHandlerSpec(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
-		super(sessionFactory);
+		this.target = new SftpMessageHandler(sessionFactory);
 	}
 
 	SftpMessageHandlerSpec(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate) {
-		super(remoteFileTemplate);
+		this.target = new SftpMessageHandler(remoteFileTemplate.getSessionFactory());
 	}
 
 	SftpMessageHandlerSpec(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate, FileExistsMode fileExistsMode) {
-		super(remoteFileTemplate, fileExistsMode);
+		this.target = new SftpMessageHandler(new SftpRemoteFileTemplate(remoteFileTemplate.getSessionFactory()), fileExistsMode);
 	}
 
 }

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/dsl/SftpTests.java

@@ -30,6 +30,7 @@
 import java.util.regex.Matcher;
 
 import org.hamcrest.Matchers;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -61,6 +62,7 @@
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Joaquin Santana
  *
  * @since 5.0
  *
@@ -152,6 +154,34 @@ public void testSftpOutboundFlow() {
 		registration.destroy();
 	}
 
+
+	@Test
+	@Ignore("Doesn't work as expected on Windows")
+	public void testSftpOutboundFlowWithChmod() {
+		IntegrationFlow flow = f -> f.handle(Sftp.outboundAdapter(sessionFactory(), FileExistsMode.FAIL)
+				.useTemporaryFileName(false)
+				.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
+				.chmod(0644)
+				.remoteDirectory("sftpTarget"));
+		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
+		String fileName = "foo.file";
+		registration.getInputChannel().send(MessageBuilder.withPayload("foo")
+				.setHeader(FileHeaders.FILENAME, fileName)
+				.build());
+
+		RemoteFileTemplate<ChannelSftp.LsEntry> template = new RemoteFileTemplate<>(sessionFactory());
+		ChannelSftp.LsEntry[] files = template.execute(session ->
+				session.list(getTargetRemoteDirectory().getName() + "/" + fileName));
+		assertEquals(1, files.length);
+		assertEquals(3, files[0].getAttrs().getSize());
+		String[] permissions = files[0].getAttrs().getPermissionsString().substring(1).replaceAll("--", "-").split("-");
+		assertEquals("rw", permissions[0]);
+		assertEquals("r", permissions[1]);
+		assertEquals("r", permissions[2]);
+
+		registration.destroy();
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testSftpMgetFlow() {

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java

@@ -64,6 +64,8 @@
  * @author Gunnar Hillert
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Joaquin Santana
+ *
  * @since 2.0
  */
 public class SftpInboundRemoteFileSystemSynchronizerTests {
@@ -170,7 +172,8 @@ private void init() {
 
 				Calendar calendar = Calendar.getInstance();
 				calendar.add(Calendar.DATE, 1);
-				when(lsEntry.getAttrs().getMTime()).thenReturn(new Long(calendar.getTimeInMillis() / 1000).intValue());
+				when(lsEntry.getAttrs().getMTime())
+						.thenReturn(Long.valueOf(calendar.getTimeInMillis() / 1000).intValue());
 				when(lsEntry.getFilename()).thenReturn(fileName);
 				when(lsEntry.getLongname()).thenReturn(fileName);
 				sftpEntries.add(lsEntry);