From e9f44960ea9b283267790a1287510d35c40557d7 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 27 Apr 2016 15:10:02 -0400 Subject: [PATCH] INT-4121: Backport of Remote File Streaming JIRA: https://jira.spring.io/browse/INT-4121 Backport abstract class only. Introduce temporary `ExtendedRemoteFileOperations`. --- .../IntegrationMessageHeaderAccessor.java | 16 ++ .../transformer/StreamTransformer.java | 70 ++++++ .../integration/file/FileHeaders.java | 4 + ...tractRemoteFileStreamingMessageSource.java | 187 +++++++++++++++ .../remote/ExtendedRemoteFileOperations.java | 46 ++++ .../file/remote/RemoteFileTemplate.java | 20 +- .../AbstractRemoteFileOutboundGateway.java | 18 +- .../file/remote/session/Session.java | 6 +- .../file/splitter/FileSplitter.java | 20 +- .../file/remote/RemoteFileTestSupport.java | 177 ++++++++++++++ .../file/remote/StreamingInboundTests.java | 221 ++++++++++++++++++ .../ftp/outbound/FtpServerOutboundTests.java | 6 +- ...utboundChannelAdapterIntegrationTests.java | 3 +- .../RedisStoreWritingMessageHandlerTests.java | 7 +- .../outbound/SftpServerOutboundTests.java | 4 +- 15 files changed, 787 insertions(+), 18 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/remote/ExtendedRemoteFileOperations.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java index 60f6a9e770..1bf9847224 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java @@ -16,6 +16,7 @@ package org.springframework.integration; +import java.io.Closeable; import java.util.Date; import java.util.Map; @@ -50,6 +51,8 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { public static final String DUPLICATE_MESSAGE = "duplicateMessage"; + public static final String CLOSEABLE_RESOURCE = "closableResource"; + public IntegrationMessageHeaderAccessor(Message message) { super(message); } @@ -76,6 +79,19 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { return this.getHeader(PRIORITY, Integer.class); } + /** + * If the payload was created by a {@link Closeable} that needs to remain + * open until the payload is consumed, the resource will be added to this + * header. After the payload is consumed the {@link Closeable} should be + * closed. Usually this must occur in an endpoint close to the message + * origin in the flow, and in the same JVM. + * @return the {@link Closeable}. + * @since 4.2.10 + */ + public Closeable getCloseableResource() { + return this.getHeader(CLOSEABLE_RESOURCE, Closeable.class); + } + @SuppressWarnings("unchecked") public T getHeader(String key, Class type) { Object value = getHeader(key); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java new file mode 100644 index 0000000000..9be60ad5e3 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java @@ -0,0 +1,70 @@ +/* + * 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.transformer; + +import java.io.ByteArrayOutputStream; +import java.io.Closeable; +import java.io.InputStream; + +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.util.FileCopyUtils; + +/** + * Transforms an InputStream payload to a byte[] or String (if a + * charset is provided). + * + * @author Gary Russell + * @since 4.2.10 + * + */ +public class StreamTransformer extends AbstractTransformer { + + private final String charset; + + /** + * Construct an instance to transform an {@link InputStream} to + * a {@code byte[]}. + */ + public StreamTransformer() { + this(null); + } + + /** + * Construct an instance with the charset to convert the stream to a + * String; if null a {@code byte[]} will be produced instead. + * @param charset the charset. + */ + public StreamTransformer(String charset) { + this.charset = charset; + } + + @Override + protected Object doTransform(Message message) throws Exception { + Assert.isTrue(message.getPayload() instanceof InputStream, "payload must be an InputStream"); + InputStream stream = (InputStream) message.getPayload(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(stream, baos); + Closeable closeableResource = new IntegrationMessageHeaderAccessor(message).getCloseableResource(); + if (closeableResource != null) { + closeableResource.close(); + } + return this.charset == null ? baos.toByteArray() : baos.toString(this.charset); + } + +} 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 0132040e24..6cf47aec0b 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 @@ -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"; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java new file mode 100644 index 0000000000..e3f672b399 --- /dev/null +++ b/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.2.10 + * + */ +public abstract class AbstractRemoteFileStreamingMessageSource extends AbstractMessageSource + implements BeanFactoryAware, InitializingBean { + + private final RemoteFileTemplate remoteFileTemplate; + + private final BlockingQueue> toBeReceived = new LinkedBlockingQueue>(); + + private final Comparator> comparator; + + /** + * the path on the remote server. + */ + private volatile Expression remoteDirectoryExpression; + + private volatile String remoteFileSeparator = "/"; + + /** + * An {@link FileListFilter} that runs against the remote file system view. + */ + private volatile FileListFilter filter; + + protected AbstractRemoteFileStreamingMessageSource(RemoteFileTemplate template, + Comparator> 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 filter) { + this.filter = filter; + } + + protected RemoteFileTemplate 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 file = poll(); + if (file != null) { + String remotePath = remotePath(file); + Session session = this.remoteFileTemplate.getSession(); + 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 poll() { + if (this.toBeReceived.size() == 0) { + listFiles(); + } + return this.toBeReceived.poll(); + } + + protected String remotePath(AbstractFileInfo 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 filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files); + List> fileInfoList = asFileInfoList(filteredFiles); + Iterator> iterator = fileInfoList.iterator(); + while (iterator.hasNext()) { + AbstractFileInfo 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> asFileInfoList(Collection files); + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/ExtendedRemoteFileOperations.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/ExtendedRemoteFileOperations.java new file mode 100644 index 0000000000..cbfae89eeb --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/ExtendedRemoteFileOperations.java @@ -0,0 +1,46 @@ +/* + * 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 org.springframework.integration.file.remote.session.Session; + +/** + * Temporary extension to {@link RemoteFileOperations} (back port). + * Merged into {@link RemoteFileOperations} in 4.3. + * + * @author Gary Russell + * @since 4.2.10 + * + */ +public interface ExtendedRemoteFileOperations extends RemoteFileOperations { + + /** + * List the files at the remote path. + * @param path the path. + * @return the list. + */ + F[] list(String path); + + /** + * Obtain a raw Session object. User must close the session when it is no longer + * needed. + * @return a session. + * @since 4.3 + */ + Session getSession(); + +} 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 52f53471d9..ad363b6218 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 @@ -56,7 +56,7 @@ import org.springframework.util.StringUtils; * @since 3.0 * */ -public class RemoteFileTemplate implements RemoteFileOperations, InitializingBean, BeanFactoryAware { +public class RemoteFileTemplate implements ExtendedRemoteFileOperations, InitializingBean, BeanFactoryAware { private final Log logger = LogFactory.getLog(this.getClass()); @@ -406,6 +406,24 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ }); } + + @Override + public F[] list(final String path) { + return this.execute(new SessionCallback() { + + @Override + public F[] doInSession(Session session) throws IOException { + return session.list(path); + } + + }); + } + + @Override + public Session getSession() { + return this.sessionFactory.getSession(); + } + @SuppressWarnings("rawtypes") @Override public T execute(SessionCallback callback) { 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 a9935de004..784f040719 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 @@ -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; @@ -47,7 +48,6 @@ 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.integration.support.PartialSuccessException; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; @@ -537,13 +537,12 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply } }); } - AbstractIntegrationMessageBuilder builder = this.getMessageBuilderFactory().withPayload(payload) - .setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir) - .setHeader(FileHeaders.REMOTE_FILE, remoteFilename); - if (session != null) { - builder.setHeader(FileHeaders.REMOTE_SESSION, session); - } - return builder.build(); + return getMessageBuilderFactory().withPayload(payload) + .setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir) + .setHeader(FileHeaders.REMOTE_FILE, remoteFilename) + .setHeader("file_remoteSession", session) // TODO: remove in 5.0 + .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session) + .build(); } private Object doMget(final Message requestMessage) { @@ -931,6 +930,9 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply else if (e instanceof IOException) { throw (IOException) e; } + else { + throw new MessagingException("Failed to process MGET on first file", e); + } } return files; } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java index 3da476a5eb..76c0789299 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java +++ b/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 { +public interface Session extends Closeable { boolean remove(String path) throws IOException; @@ -62,6 +63,7 @@ public interface Session { void rename(String pathFrom, String pathTo) throws IOException; + @Override void close(); boolean isOpen(); 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 031b7375da..07ca095314 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 @@ -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; @@ -188,7 +190,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 iterator = new Iterator() { boolean markers = FileSplitter.this.markers; diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java new file mode 100644 index 0000000000..8100112e0f --- /dev/null +++ b/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.2.10 + * + */ +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: + * + * 
+	 *  $ 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/
+	 *
+ * + * The intent is tests retrieve from remoteSource and verify arrival in localTarget or send from localSource and verify + * arrival in remoteTarget. + *

+ * Subclasses can change 'remote' in these names by overriding {@link #prefix()} or override this method completely to + * create a different structure. + *

+ * 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"; + } + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java new file mode 100644 index 0000000000..c9e3b0e431 --- /dev/null +++ b/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.2.10 + * + */ +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 received = (Message) 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) 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 receivedStream = streamer.receive(); + splitter.handleMessage(receivedStream); + Message received = (Message) 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) 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) 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) 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 { + + protected Streamer(RemoteFileTemplate template, Comparator> comparator) { + super(template, comparator); + } + + @Override + public String getComponentType() { + return "Streamer"; + } + + @Override + protected List> asFileInfoList(Collection files) { + List> infos = new ArrayList>(); + for (String file : files) { + infos.add(new StringFileInfo(file)); + } + return infos; + } + + } + + public static class StringFileInfo extends AbstractFileInfo { + + 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 { + + public StringRemoteFileTemplate(SessionFactory sessionFactory) { + super(sessionFactory); + } + + } + + public static class StringSessionFactory implements SessionFactory { + + @SuppressWarnings("unchecked") + @Override + public Session getSession() { + try { + Session 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); + } + } + + } + +} diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java index f8eca3ca8d..7be875597f 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java @@ -56,6 +56,7 @@ import org.mockito.stubbing.Answer; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.filters.FileListFilter; @@ -380,7 +381,7 @@ public class FtpServerOutboundTests { 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); + Session session = (Session) result.getHeaders().get(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE); // Returned to cache assertTrue(session.isOpen()); // Raw reading is finished @@ -394,7 +395,8 @@ public class FtpServerOutboundTests { 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")); + TestUtils.getPropertyValue(result.getHeaders().get(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE), + "targetSession")); } @Test diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java index 5f9b5cceee..b0d0ae138c 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java @@ -26,6 +26,7 @@ import java.util.Map; import java.util.Properties; import java.util.Set; +import org.junit.Ignore; import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -85,7 +86,7 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail context.close(); } - @Test + @Test @Ignore @RedisAvailable public void testListWithKeyAsHeaderSimple(){ RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandlerTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandlerTests.java index 5bd662b8b6..75ade7bee1 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandlerTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandlerTests.java @@ -91,7 +91,7 @@ public class RedisStoreWritingMessageHandlerTests extends RedisAvailableTests{ @Test @RedisAvailable - public void testListWithListPayloadParsedAndProvidedKeyAsHeader() { + public void testListWithListPayloadParsedAndProvidedKeyAsHeader() throws Exception { RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); this.deleteKey(jcf, "foo"); String key = "foo"; @@ -112,7 +112,10 @@ public class RedisStoreWritingMessageHandlerTests extends RedisAvailableTests{ list.add("Jack"); Message> message = MessageBuilder.withPayload(list).setHeader("redis_key", key).build(); handler.handleMessage(message); - + int n = 0; + while (n++ < 100 && redisList.size() != 3) { + Thread.sleep(100); + } assertEquals(3, redisList.size()); assertEquals("Manny", redisList.get(0)); assertEquals("Moe", redisList.get(1)); 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 955fb29f0b..a158603c07 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 @@ -45,6 +45,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.remote.MessageSessionCallback; @@ -422,7 +423,8 @@ public class SftpServerOutboundTests { 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()); + assertFalse( + ((Session) result.getHeaders().get(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE)).isOpen()); } @Test