From 66b53e749fee06413860ad98924ac1357e276f54 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 23 Feb 2017 17:25:00 -0500 Subject: [PATCH] INT-4233: Prework - Simple JSON Serializer JIRA: https://jira.spring.io/browse/INT-4233 INT-4233: Add (S)FTP FileInfo Header (Streaming) JIRA: https://jira.spring.io/browse/INT-4233 Add the complete file info as JSON (when Jackson or Boon available) to the message headers when streaming inbound. Provide a mechanism to configure Boon to provide similar output to Jackson. Also allow subclasses to provide their own object mapper. Also clean up after the AMQP DSL tests, and don't use a queue `foo`. I often have such a queue with content; this caused tests to fail. Use SimpleJsonSerializer Doc Polishing Polishing - PR Comments More Polishing * Polishing according latest PR comments --- .../json/SimpleJsonSerializer.java | 96 +++++++++++++++++++ .../json/SimpleJsonSerializerTests.java | 88 +++++++++++++++++ .../integration/file/FileHeaders.java | 7 +- .../file/remote/AbstractFileInfo.java | 11 ++- ...tractRemoteFileStreamingMessageSource.java | 20 +++- .../file/remote/StreamingInboundTests.java | 30 +++++- .../FtpStreamingMessageSourceTests.java | 44 +++++++-- .../sftp/session/SftpFileInfo.java | 7 ++ .../SftpStreamingMessageSourceTests.java | 46 +++++++-- src/reference/asciidoc/ftp.adoc | 8 +- src/reference/asciidoc/sftp.adoc | 16 +++- src/reference/asciidoc/whats-new.adoc | 3 + 12 files changed, 347 insertions(+), 29 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/json/SimpleJsonSerializer.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/json/SimpleJsonSerializerTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/json/SimpleJsonSerializer.java b/spring-integration-core/src/main/java/org/springframework/integration/json/SimpleJsonSerializer.java new file mode 100644 index 0000000000..610292d2bb --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/json/SimpleJsonSerializer.java @@ -0,0 +1,96 @@ +/* + * Copyright 2017 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.json; + +import java.beans.PropertyDescriptor; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.BeanUtils; + +/** + * Extremely simple JSON serializer. Only handles top level + * properties accessed by getters. + * + * @author Gary Russell + * @since 5.0 + * + */ +public final class SimpleJsonSerializer { + + private static final Log logger = LogFactory.getLog(SimpleJsonSerializer.class); + + private SimpleJsonSerializer() { + super(); + } + + /** + * Convert the bean to JSON with the provided properties. + * @param bean the object to serialize. + * @param propertiesToExclude the property names to ignore. + * @return the JSON. + */ + public static String toJson(Object bean, String... propertiesToExclude) { + PropertyDescriptor[] propertyDescriptors = BeanUtils.getPropertyDescriptors(bean.getClass()); + Set excluded = new HashSet<>(Arrays.asList(propertiesToExclude)); + excluded.add("class"); + final StringBuilder stringBuilder = new StringBuilder("{"); + final Object[] emptyArgs = new Object[0]; + for (PropertyDescriptor descriptor : propertyDescriptors) { + String propertyName = descriptor.getName(); + Method readMethod = descriptor.getReadMethod(); + if (!excluded.contains(propertyName) && readMethod != null) { + stringBuilder.append(toElement(propertyName)).append(":"); + Object result; + try { + result = readMethod.invoke(bean, emptyArgs); + } + catch (InvocationTargetException | IllegalAccessException | IllegalArgumentException e) { + if (logger.isDebugEnabled()) { + logger.debug("Failed to serialize property " + propertyName, e); + } + result = e.getMessage(); + } + stringBuilder.append(toElement(result)).append(","); + } + } + stringBuilder.setLength(stringBuilder.length() - 1); + stringBuilder.append("}"); + if (stringBuilder.length() == 1) { + return null; + } + else { + return stringBuilder.toString(); + } + } + + private static String toElement(Object result) { + if (result instanceof Number || result instanceof Boolean) { + return result.toString(); + } + else { + return "\"" + result.toString() + "\""; + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/SimpleJsonSerializerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/json/SimpleJsonSerializerTests.java new file mode 100644 index 0000000000..f931b3feec --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/SimpleJsonSerializerTests.java @@ -0,0 +1,88 @@ +/* + * Copyright 2017 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.json; + +import static org.hamcrest.CoreMatchers.equalTo; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; + +import org.junit.Test; + +import org.springframework.integration.support.json.JsonObjectMapperProvider; + +/** + * @author Gary Russell + * @since 5.0 + * + */ +public class SimpleJsonSerializerTests { + + @Test + public void test() throws Exception { + Foo foo = new Foo("foo"); + String json = SimpleJsonSerializer.toJson(foo, "fileInfo"); + Foo fooOut = JsonObjectMapperProvider.newInstance().fromJson(json, Foo.class); + assertThat(fooOut.bool, equalTo(Boolean.TRUE)); + assertThat(fooOut.bar, equalTo(42L)); + assertThat(fooOut.foo, equalTo("bar")); + assertThat(fooOut.dub, equalTo(1.6)); + assertNull(fooOut.fileInfo); + } + + public static class Foo { + + private final String foo = "bar"; + + private final long bar = 42L; + + private final double dub = 1.6; + + private final boolean bool = true; + + private String fileInfo; + + public Foo() { + super(); + } + + public Foo(String info) { + this.fileInfo = "foo"; + } + + public String getFoo() { + return this.foo; + } + + public long getBar() { + return this.bar; + } + + public double getDub() { + return this.dub; + } + + public boolean isBool() { + return this.bool; + } + + public String fileInfo() { + return this.fileInfo; + } + + } + +} 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 36bee26abb..3c75528852 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-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -47,4 +47,9 @@ public abstract class FileHeaders { */ public static final String MARKER = PREFIX + "marker"; + /** + * A remote file information representation + */ + public static final String REMOTE_FILE_INFO = PREFIX + "remoteFileInfo"; + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractFileInfo.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractFileInfo.java index 35365d5cda..a0a7df5ae2 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractFileInfo.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractFileInfo.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -18,6 +18,8 @@ package org.springframework.integration.file.remote; import java.util.Date; +import org.springframework.integration.json.SimpleJsonSerializer; + /** * Abstract implementation of {@link FileInfo}; provides a setter * for the remote directory and a generic toString implementation. @@ -36,10 +38,12 @@ public abstract class AbstractFileInfo implements FileInfo, Comparable o) { return this.getFilename().compareTo(o.getFilename()); } + public String toJson() { + return SimpleJsonSerializer.toJson(this, "fileInfo"); + } + } 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 index 2029397fc0..db2474f14f 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -58,6 +58,8 @@ public abstract class AbstractRemoteFileStreamingMessageSource private final Comparator> comparator; + private boolean fileInfoJson = true; + /** * the path on the remote server. */ @@ -112,6 +114,17 @@ public abstract class AbstractRemoteFileStreamingMessageSource this.filter = filter; } + /** + * Set to false to add the {@link FileHeaders#REMOTE_FILE_INFO} header to the raw {@link FileInfo}. + * Default is true meaning that common file information properties are provided + * in that header as JSON. + * @param fileInfoJson false to set the raw object. + * @since 5.0 + */ + public void setFileInfoJson(boolean fileInfoJson) { + this.fileInfoJson = fileInfoJson; + } + protected RemoteFileTemplate getRemoteFileTemplate() { return this.remoteFileTemplate; } @@ -136,10 +149,13 @@ public abstract class AbstractRemoteFileStreamingMessageSource String remotePath = remotePath(file); Session session = this.remoteFileTemplate.getSession(); try { - return getMessageBuilderFactory().withPayload(session.readRaw(remotePath)) + return getMessageBuilderFactory() + .withPayload(session.readRaw(remotePath)) .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session) .setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory()) .setHeader(FileHeaders.REMOTE_FILE, file.getFilename()) + .setHeader(FileHeaders.REMOTE_FILE_INFO, + this.fileInfoJson ? file.toJson() : file) .build(); } catch (IOException e) { 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 index e90e864f87..cfc5bd13ca 100644 --- 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 @@ -16,8 +16,10 @@ package org.springframework.integration.file.remote; +import static org.hamcrest.CoreMatchers.containsString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willReturn; import static org.mockito.BDDMockito.willThrow; @@ -73,6 +75,14 @@ public class StreamingInboundTests { assertEquals("foo\nbar", new String(received.getPayload())); assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + String fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); + assertThat(fileInfo, containsString("remoteDirectory\":\"/foo")); + assertThat(fileInfo, containsString("permissions\":\"-rw-rw-rw")); + assertThat(fileInfo, containsString("size\":42")); + assertThat(fileInfo, containsString("directory\":false")); + assertThat(fileInfo, containsString("filename\":\"foo")); + assertThat(fileInfo, containsString("modified\":42000")); + assertThat(fileInfo, containsString("link\":false")); // close after list, transform verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(2)).close(); @@ -81,6 +91,14 @@ public class StreamingInboundTests { assertEquals("baz\nqux", new String(received.getPayload())); assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); + assertThat(fileInfo, containsString("remoteDirectory\":\"/foo")); + assertThat(fileInfo, containsString("permissions\":\"-rw-rw-rw")); + assertThat(fileInfo, containsString("size\":42")); + assertThat(fileInfo, containsString("directory\":false")); + assertThat(fileInfo, containsString("filename\":\"bar")); + assertThat(fileInfo, containsString("modified\":42000")); + assertThat(fileInfo, containsString("link\":false")); // close after transform verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(3)).close(); @@ -213,12 +231,12 @@ public class StreamingInboundTests { @Override public long getSize() { - return 0; + return 42; } @Override public long getModified() { - return 0; + return 42_000; } @Override @@ -228,12 +246,16 @@ public class StreamingInboundTests { @Override public String getPermissions() { - return null; + return "-rw-rw-rw"; } @Override public String getFileInfo() { - return null; + return asString(); + } + + private String asString() { + return "StringFileInfo [name=" + this.name + "]"; } } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java index 7954be4db7..f63ad96ee8 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java @@ -16,9 +16,10 @@ package org.springframework.integration.ftp.inbound; +import static org.hamcrest.CoreMatchers.containsString; +import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.Matchers.equalTo; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import java.io.InputStream; @@ -35,15 +36,15 @@ import org.springframework.integration.annotation.Transformer; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessageSource; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.ftp.FtpTestSupport; -import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter; +import org.springframework.integration.ftp.session.FtpFileInfo; import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; -import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.transformer.StreamTransformer; import org.springframework.messaging.Message; -import org.springframework.messaging.PollableChannel; import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -59,7 +60,13 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class FtpStreamingMessageSourceTests extends FtpTestSupport { @Autowired - public PollableChannel data; + private QueueChannel data; + + @Autowired + private FtpStreamingMessageSource source; + + @Autowired + private SourcePollingChannelAdapter adapter; @SuppressWarnings("unchecked") @Test @@ -67,10 +74,34 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { Message received = (Message) this.data.receive(10000); assertNotNull(received); assertThat(new String(received.getPayload()), equalTo("source1")); + String fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); + assertThat(fileInfo, containsString("remoteDirectory\":\"ftpSource")); + assertThat(fileInfo, containsString("permissions\":\"-rw-------")); + assertThat(fileInfo, containsString("size\":7")); + assertThat(fileInfo, containsString("directory\":false")); + assertThat(fileInfo, containsString("filename\":\" ftpSource1.txt")); + assertThat(fileInfo, containsString("modified\":")); + assertThat(fileInfo, containsString("link\":false")); received = (Message) this.data.receive(10000); assertNotNull(received); assertThat(new String(received.getPayload()), equalTo("source2")); - assertNull(this.data.receive(10)); + fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); + assertThat(fileInfo, containsString("remoteDirectory\":\"ftpSource")); + assertThat(fileInfo, containsString("permissions\":\"-rw-------")); + assertThat(fileInfo, containsString("size\":7")); + assertThat(fileInfo, containsString("directory\":false")); + assertThat(fileInfo, containsString("filename\":\"ftpSource2.txt")); + assertThat(fileInfo, containsString("modified\":")); + assertThat(fileInfo, containsString("link\":false")); + + this.adapter.stop(); + this.source.setFileInfoJson(false); + this.data.purge(null); + this.adapter.start(); + received = (Message) this.data.receive(10000); + assertNotNull(received); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO), instanceOf(FtpFileInfo.class)); + this.adapter.stop(); } @Configuration @@ -95,7 +126,6 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { public MessageSource ftpMessageSource() { FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null); messageSource.setRemoteDirectory("ftpSource/"); - messageSource.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming")); return messageSource; } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpFileInfo.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpFileInfo.java index 39c1df434a..45c4f466bb 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpFileInfo.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpFileInfo.java @@ -45,6 +45,7 @@ public class SftpFileInfo extends AbstractFileInfo { /** * @see com.jcraft.jsch.SftpATTRS#isDir() */ + @Override public boolean isDirectory() { return this.attrs.isDir(); } @@ -52,6 +53,7 @@ public class SftpFileInfo extends AbstractFileInfo { /** * @see com.jcraft.jsch.SftpATTRS#isLink() */ + @Override public boolean isLink() { return this.attrs.isLink(); } @@ -59,6 +61,7 @@ public class SftpFileInfo extends AbstractFileInfo { /** * @see com.jcraft.jsch.SftpATTRS#getSize() */ + @Override public long getSize() { return this.attrs.getSize(); } @@ -66,6 +69,7 @@ public class SftpFileInfo extends AbstractFileInfo { /** * @see com.jcraft.jsch.SftpATTRS#getMTime() */ + @Override public long getModified() { return ((long) this.attrs.getMTime()) * 1000; } @@ -73,14 +77,17 @@ public class SftpFileInfo extends AbstractFileInfo { /** * @see com.jcraft.jsch.ChannelSftp.LsEntry#getFilename() */ + @Override public String getFilename() { return this.lsEntry.getFilename(); } + @Override public String getPermissions() { return this.attrs.getPermissionsString(); } + @Override public LsEntry getFileInfo() { return this.lsEntry; } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java index 49dc3d92b4..e774dfcf1e 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java @@ -16,9 +16,10 @@ package org.springframework.integration.sftp.inbound; +import static org.hamcrest.CoreMatchers.containsString; +import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.Matchers.equalTo; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import java.io.InputStream; @@ -34,15 +35,15 @@ import org.springframework.integration.annotation.Transformer; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessageSource; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.remote.session.SessionFactory; -import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.sftp.SftpTestSupport; -import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter; +import org.springframework.integration.sftp.session.SftpFileInfo; import org.springframework.integration.sftp.session.SftpRemoteFileTemplate; import org.springframework.integration.transformer.StreamTransformer; import org.springframework.messaging.Message; -import org.springframework.messaging.PollableChannel; import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -60,7 +61,13 @@ import com.jcraft.jsch.ChannelSftp.LsEntry; public class SftpStreamingMessageSourceTests extends SftpTestSupport { @Autowired - public PollableChannel data; + private QueueChannel data; + + @Autowired + private SftpStreamingMessageSource source; + + @Autowired + private SourcePollingChannelAdapter adapter; @SuppressWarnings("unchecked") @Test @@ -68,10 +75,34 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { Message received = (Message) this.data.receive(10000); assertNotNull(received); assertThat(new String(received.getPayload()), equalTo("source1")); + String fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); + assertThat(fileInfo, containsString("remoteDirectory\":\"sftpSource")); + assertThat(fileInfo, containsString("permissions\":\"-rw-r--r--")); + assertThat(fileInfo, containsString("size\":7")); + assertThat(fileInfo, containsString("directory\":false")); + assertThat(fileInfo, containsString("filename\":\" sftpSource1.txt")); + assertThat(fileInfo, containsString("modified\":")); + assertThat(fileInfo, containsString("link\":false")); received = (Message) this.data.receive(10000); assertNotNull(received); + fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); + assertThat(fileInfo, containsString("remoteDirectory\":\"sftpSource")); + assertThat(fileInfo, containsString("permissions\":\"-rw-r--r--")); + assertThat(fileInfo, containsString("size\":7")); + assertThat(fileInfo, containsString("directory\":false")); + assertThat(fileInfo, containsString("filename\":\"sftpSource2.txt")); + assertThat(fileInfo, containsString("modified\":")); + assertThat(fileInfo, containsString("link\":false")); assertThat(new String(received.getPayload()), equalTo("source2")); - assertNull(this.data.receive(10)); + + this.adapter.stop(); + this.source.setFileInfoJson(false); + this.data.purge(null); + this.adapter.start(); + received = (Message) this.data.receive(10000); + assertNotNull(received); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO), instanceOf(SftpFileInfo.class)); + this.adapter.stop(); } @Configuration @@ -93,10 +124,9 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { @Bean @InboundChannelAdapter(channel = "stream") - public MessageSource ftpMessageSource() { + public MessageSource sftpMessageSource() { SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), null); messageSource.setRemoteDirectory("sftpSource/"); - messageSource.setFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming")); return messageSource; } diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 0ebaad2ec1..eae56661a1 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -534,7 +534,13 @@ If you don't actually want to persist the state, an in-memory `SimpleMetadataSto If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`. The java configuration below shows one technique to remove the remote file after processing. -Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment; +Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment. + +The adapter puts the remote directory and file name in headers `FileHeaders.REMOTE_DIRECTORY` and `FileHeaders.REMOTE_FILE` respectively. +Starting with _version 5.0_, additional remote file information, represented in JSON by default, is provided in the `FileHeaders.REMOTE_FILE_INFO` header. +If you set the `fileInfoJson` property on the `FtpStreamingMessageSource` to `false`, the header will contain an `FtpFileInfo` object. +The `FTPFile` object provided by the underlying Apache Net library can be accessed using the `FtpFileInfo.getFileInfo()` method. +The `fileInfoJson` property is not available when using XML configuration but you can set it by injecting the `FtpStreamingMessageSource` into one of your configuration classes. ==== Configuring with Java Configuration diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 2cf2b4d6fe..cf5cf83769 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -573,7 +573,13 @@ If you don't actually want to persist the state, an in-memory `SimpleMetadataSto If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`. The java configuration below shows one technique to remove the remote file after processing. -Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment; +Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment. + +The adapter puts the remote directory and file name in headers `FileHeaders.REMOTE_DIRECTORY` and `FileHeaders.REMOTE_FILE` respectively. +Starting with _version 5.0_, additional remote file information, in JSON, is provided in the `FileHeaders.REMOTE_FILE_INFO` header. +If you set the `fileInfoJson` property on the `SftpStreamingMessageSource` to `false`, the header will contain an `SftpFileInfo` object. +The `LsEntry` object provided by the underlying Jsch library can be accessed using the `SftpFileInfo.getFileInfo()` method. +The `fileInfoJson` property is not available when using XML configuration but you can set it by injecting the `SftpStreamingMessageSource` into one of your configuration classes. ==== Configuring with Java Configuration @@ -1048,13 +1054,13 @@ public class SftpJavaApplication { } @Bean - public SessionFactory ftpSessionFactory() { + public SessionFactory sftpSessionFactory() { DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory(); sf.setHost("localhost"); sf.setPort(port); sf.setUsername("foo"); sf.setPassword("foo"); - return new CachingSessionFactory(sf); + return new CachingSessionFactory(sf); } @Bean @@ -1066,7 +1072,7 @@ public class SftpJavaApplication { public IntegrationFlow sftpMGetFlow() { return IntegrationFlows.from("sftpMgetInputChannel") .handleWithAdapter(h -> - h.sftpGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET, + h.sftpGateway(sftpSessionFactory(), AbstractRemoteFileOutboundGateway.Command.MGET, "payload") .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE) .regexFileNameFilter("(subSftpSource|.*1.txt)") @@ -1129,7 +1135,7 @@ log4j.category.com.jcraft.jsch=DEBUG === MessageSessionCallback Starting with _Spring Integration version 4.2_, a `MessageSessionCallback` implementation can be used with the -`` (`FtpOutboundGateway`) to perform any operation(s) on the `Session` with +`` (`SftpOutboundGateway`) to perform any operation(s) on the `Session` with the `requestMessage` context. It can be used for any non-standard or low-level FTP operation (or several); for example, allowing access from an integration flow definition, and _functional_ interface (Lambda) implementation injection: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 62abbf0b60..9d7e583ed0 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -85,6 +85,9 @@ See <> and <> for more information. The FTP and SFTP outbound gateways now support the `REPLACE_IF_MODIFIED` `FileExistsMode` when fetching remote files. See <> and <> for more information. +The (S)FTP streaming inbound channel adapters now add remote file information in a message header. +See <> and <> for more information. + ==== Integration Properties Since _version 4.3.2_ a new `spring.integration.readOnly.headers` global property has been added to customize the list of headers which should not be copied to a newly created `Message` by the `MessageBuilder`.