diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java index e47d9ff31e..c88f2a84f5 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-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. @@ -35,6 +35,8 @@ public abstract class FileHeaders { public static final String REMOTE_FILE = PREFIX + "remoteFile"; + public static final String REMOTE_SESSION = PREFIX + "remoteSession"; + public static final String RENAME_TO = PREFIX + "renameTo"; } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java index 763baa2576..05f15b46a4 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2014 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. @@ -98,6 +98,14 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ this.sessionFactory = sessionFactory; } + /** + * @return this template's {@link SessionFactory}. + * @since 4.2 + */ + public SessionFactory getSessionFactory() { + return sessionFactory; + } + /** * Determine whether the remote directory should automatically be created when * sending files to the remote system. diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index 850d9e3ba7..1549aea010 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -40,12 +40,14 @@ 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.RemoteFileTemplate; +import org.springframework.integration.file.remote.RemoteFileUtils; 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.file.support.FileExistsMode; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; @@ -170,7 +172,12 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply /** * Recursive (ls, mget) */ - RECURSIVE("-R"); + RECURSIVE("-R"), + + /** + * Streaming 'get' (returns InputStream); user must call {@link RemoteFileUtils#closeSession(Session)}. + */ + STREAM("-stream"); private String option; @@ -364,7 +371,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply Command.GET.equals(this.command)) { Assert.isNull(this.filter, "Filters are not supported with the rm and get commands"); } - if (Command.GET.equals(this.command) + if ((Command.GET.equals(this.command) && !options.contains(Option.STREAM)) || Command.MGET.equals(this.command)) { Assert.notNull(this.localDirectoryExpression, "localDirectory must not be null"); if (this.localDirectoryExpression instanceof LiteralExpression) { @@ -449,19 +456,37 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage); final String remoteFilename = this.getRemoteFilename(remoteFilePath); final String remoteDir = this.getRemoteDirectory(remoteFilePath, remoteFilename); - File payload = this.remoteFileTemplate.execute(new SessionCallback() { - - @Override - public File doInSession(Session session) throws IOException { - return AbstractRemoteFileOutboundGateway.this.get(requestMessage, session, remoteDir, remoteFilePath, - remoteFilename, true); - + Session session = null; + Object payload; + if (this.options.contains(Option.STREAM)) { + session = this.remoteFileTemplate.getSessionFactory().getSession(); + try { + payload = session.readRaw(remoteFilePath); } - }); - return this.getMessageBuilderFactory().withPayload(payload) + catch (IOException e) { + throw new MessageHandlingException(requestMessage, "Failed to get the remote file [" + + remoteFilePath + + "] as a stream", e); + } + } + else { + payload = this.remoteFileTemplate.execute(new SessionCallback() { + + @Override + public File doInSession(Session session) throws IOException { + return AbstractRemoteFileOutboundGateway.this.get(requestMessage, session, remoteDir, remoteFilePath, + remoteFilename, true); + + } + }); + } + AbstractIntegrationMessageBuilder builder = this.getMessageBuilderFactory().withPayload(payload) .setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir) - .setHeader(FileHeaders.REMOTE_FILE, remoteFilename) - .build(); + .setHeader(FileHeaders.REMOTE_FILE, remoteFilename); + if (session != null) { + builder.setHeader(FileHeaders.REMOTE_SESSION, session); + } + return builder.build(); } private Object doMget(final Message requestMessage) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index 2863028cdf..9314a0177f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java @@ -190,7 +190,15 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe else if (this.dirty) { this.targetSession.close(); } - pool.releaseItem(targetSession); + if (this.targetSession.isOpen()) { + try { + this.targetSession.finalizeRaw(); + } + catch (IOException e) { + //No-op in this context + } + } + pool.releaseItem(this.targetSession); released = true; } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java index 402e7d4ed1..d2bd488153 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java @@ -38,6 +38,7 @@ import org.springframework.integration.file.splitter.FileSplitter.FileMarker.Mar import org.springframework.integration.splitter.AbstractMessageSplitter; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; +import org.springframework.util.StringUtils; /** * The {@link AbstractMessageSplitter} implementation to split the {@link File} @@ -147,11 +148,11 @@ public class FileSplitter extends AbstractMessageSplitter { else { reader = new InputStreamReader((InputStream) payload, this.charset); } - filePath = ":stream:"; + filePath = buildPathFromMessage(message, ":stream:"); } else if (payload instanceof Reader) { reader = (Reader) payload; - filePath = ":reader:"; + filePath = buildPathFromMessage(message, ":reader:"); } else { return message; @@ -242,7 +243,6 @@ public class FileSplitter extends AbstractMessageSplitter { } } - @Override protected boolean willAddHeaders(Message message) { Object payload = message.getPayload(); @@ -268,6 +268,17 @@ public class FileSplitter extends AbstractMessageSplitter { } } + private String buildPathFromMessage(Message message, String defaultPath) { + String remoteDir = (String) message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY); + String remoteFile = (String) message.getHeaders().get(FileHeaders.REMOTE_FILE); + if (StringUtils.hasText(remoteDir) && StringUtils.hasText(remoteFile)) { + return remoteDir + remoteFile; + } + else { + return defaultPath; + } + } + public static class FileMarker implements Serializable { private static final long serialVersionUID = 8514605438145748406L; diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java index c2f85dc419..3b9ffc51a8 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java @@ -144,6 +144,9 @@ public class FtpSession implements Session { @Override public void close() { try { + if (this.readingRaw.get()) { + finalizeRaw(); + } this.client.disconnect(); } catch (Exception e) { diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TestFtpServer.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TestFtpServer.java index 3859943c76..7a27818553 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TestFtpServer.java +++ b/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 ftpSessionFactory() { DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory(); factory.setHost("localhost"); factory.setPort(this.ftpPort); factory.setUsername("foo"); factory.setPassword("foo"); - return factory; + + return new CachingSessionFactory(factory); } @PostConstruct diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml index d1e42ed8f8..722f69d352 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml @@ -1,12 +1,13 @@ + 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"> @@ -104,6 +105,25 @@ remote-directory="ftpTarget" reply-channel="output"/> + + + + + + + + + + + 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(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(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() { diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java index e8f4dd96af..db98f3ea9e 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java +++ b/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 sessionFactory; @Before @After diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml index ad8633a758..10a1d2b843 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml @@ -1,12 +1,13 @@ + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp" + 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/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd + 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 http://www.springframework.org/schema/integration/spring-integration.xsd"> @@ -133,4 +134,23 @@ auto-create-directory="true" remote-file-separator="/" /> + + + + + + + + + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java index f63a44e8c0..fe9c59d169 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2014 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,6 +21,7 @@ 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; @@ -125,6 +126,9 @@ public class SftpServerOutboundTests { @Autowired private TestSftpServer sftpServer; + @Autowired + private DirectChannel inboundGetStream; + @Before @After public void setup() { @@ -404,6 +408,18 @@ public class SftpServerOutboundTests { } + @Test + public void testStream() { + String dir = "sftpSource/"; + this.inboundGetStream.send(new GenericMessage(dir + "sftpSource1.txt")); + Message result = this.output.receive(1000); + assertNotNull(result); + assertEquals("source1", result.getPayload()); + assertEquals("sftpSource/", result.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); + assertEquals("sftpSource1.txt", result.getHeaders().get(FileHeaders.REMOTE_FILE)); + assertFalse(((Session) result.getHeaders().get(FileHeaders.REMOTE_SESSION)).isOpen()); + } + private void assertLength6(SftpRemoteFileTemplate template) { LsEntry[] files = template.execute(new SessionCallback() { diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 2613e7d34d..73ad5dd683 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -328,10 +328,44 @@ _get_ retrieves a remote file and supports the following option: * -P - preserve the timestamp of the remote file -The message payload resulting from a _get_ operation is a `File` object representing the retrieved file. +* -stream - retrieve the remote file as a stream. The remote directory is provided in the `file_remoteDirectory` header, and the filename is provided in the `file_remoteFile` header. +The message payload resulting from a _get_ operation is a `File` object representing the retrieved file, or +an `InputStream` when the `-stream` option is provided. +This option allows retrieving the file as a stream. +For text files, a common use case is to combine this operation with a <>. +When consuming remote files as streams, the user is responsible for closing the `Session` after the stream is +consumed. +For convenience, the `Session` is provided in the `file_remoteSession` header. + +The following shows an example of consuming a file as a stream: + +[source, xml] +---- + + + + + + + + + + +---- + +The file lines are sent to the channel `output`. + *mget* _mget_ retrieves multiple remote files based on a pattern and supports the following option: diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 7c718e89df..cd572d64ef 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -391,10 +391,44 @@ _get_ retrieves a remote file and supports the following option: * -P - preserve the timestamp of the remote file -The message payload resulting from a _get_ operation is a `File` object representing the retrieved file. +* -stream - retrieve the remote file as a stream. The remote directory is provided in the `file_remoteDirectory` header, and the filename is provided in the `file_remoteFile` header. +The message payload resulting from a _get_ operation is a `File` object representing the retrieved file, or +an `InputStream` when the `-stream` option is provided. +This option allows retrieving the file as a stream. +For text files, a common use case is to combine this operation with a <>. +When consuming remote files as streams, the user is responsible for closing the `Session` after the stream is +consumed. +For convenience, the `Session` is provided in the `file_remoteSession` header. + +The following shows an example of consuming a file as a stream: + +[source, xml] +---- + + + + + + + + + + +---- + +The file lines are sent to the channel `output`. + *mget* _mget_ retrieves multiple remote files based on a pattern and supports the following option: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 510c2acd60..6c383f7cab 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -160,3 +160,9 @@ metadata store if it implements `Flushable` (e.g. the `PropertiesPersistenMetada When using Java 8, gateway methods can now return `CompletableFuture`. See <> for more information. + +[[x4.2-ftp-changes]] +==== (S)FTP Changes + +The SFTP/FTP outbound gateway `get` operations now support the `-stream` option, allowing files to be retrieved as +a stream. See <> and <> for more information.