Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4233: Prework - Simple JSON Serializer

Browse Source 
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
This commit is contained in:
Gary Russell
2017-02-23 17:25:00 -05:00
committed by Artem Bilan
parent af711a0d30
commit 66b53e749f
12 changed files with 347 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
spring-integration-core/src/main/java/org/springframework/integration/json/SimpleJsonSerializer.java Normal file
View File

@@ -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<String> 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() + "\"";
}
}
}

88
spring-integration-core/src/test/java/org/springframework/integration/json/SimpleJsonSerializerTests.java Normal file
View File

@@ -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;
}
}
}

7
spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java
View File

@@ -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";
}

11
spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractFileInfo.java
View File

@@ -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<F> implements FileInfo<F>, Comparable<Fil
this.remoteDirectory = remoteDirectory;
}
@Override
public String getRemoteDirectory() {
return this.remoteDirectory;
}
@Override
public String toString() {
return "FileInfo [isDirectory=" + isDirectory() + ", isLink=" + isLink()
+ ", Size=" + getSize() + ", ModifiedTime="
@@ -47,8 +51,13 @@ public abstract class AbstractFileInfo<F> implements FileInfo<F>, Comparable<Fil
+ ", RemoteDirectory=" + getRemoteDirectory() + ", Permissions=" + getPermissions() + "]";
}
@Override
public int compareTo(FileInfo<F> o) {
return this.getFilename().compareTo(o.getFilename());
}
public String toJson() {
return SimpleJsonSerializer.toJson(this, "fileInfo");
}
}

20
spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
View File

@@ -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<F>
private final Comparator<AbstractFileInfo<F>> comparator;
private boolean fileInfoJson = true;
/**
* the path on the remote server.
*/
@@ -112,6 +114,17 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
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<F> getRemoteFileTemplate() {
return this.remoteFileTemplate;
}
@@ -136,10 +149,13 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
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) {

30
spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
View File

@@ -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 + "]";
}
}

44
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
View File

@@ -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<byte[]> received = (Message<byte[]>) 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<byte[]>) 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<byte[]>) 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<InputStream> ftpMessageSource() {
FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null);
messageSource.setRemoteDirectory("ftpSource/");
messageSource.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming"));
return messageSource;
}

7
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpFileInfo.java
View File

@@ -45,6 +45,7 @@ public class SftpFileInfo extends AbstractFileInfo<LsEntry> {
/**
* @see com.jcraft.jsch.SftpATTRS#isDir()
*/
@Override
public boolean isDirectory() {
return this.attrs.isDir();
}
@@ -52,6 +53,7 @@ public class SftpFileInfo extends AbstractFileInfo<LsEntry> {
/**
* @see com.jcraft.jsch.SftpATTRS#isLink()
*/
@Override
public boolean isLink() {
return this.attrs.isLink();
}
@@ -59,6 +61,7 @@ public class SftpFileInfo extends AbstractFileInfo<LsEntry> {
/**
* @see com.jcraft.jsch.SftpATTRS#getSize()
*/
@Override
public long getSize() {
return this.attrs.getSize();
}
@@ -66,6 +69,7 @@ public class SftpFileInfo extends AbstractFileInfo<LsEntry> {
/**
* @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<LsEntry> {
/**
* @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;
}

46
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
View File

@@ -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<byte[]> received = (Message<byte[]>) 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<byte[]>) 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<byte[]>) 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<InputStream> ftpMessageSource() {
public MessageSource<InputStream> sftpMessageSource() {
SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), null);
messageSource.setRemoteDirectory("sftpSource/");
messageSource.setFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming"));
return messageSource;
}

8
src/reference/asciidoc/ftp.adoc
View File

@@ -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

16
src/reference/asciidoc/sftp.adoc
View File

@@ -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<FTPFile> ftpSessionFactory() {
public SessionFactory<LsEntry> sftpSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost("localhost");
sf.setPort(port);
sf.setUsername("foo");
sf.setPassword("foo");
return new CachingSessionFactory<FTPFile>(sf);
return new CachingSessionFactory<LsEntry>(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<F, T>` implementation can be used with the
`<int-ftp:outbound-gateway/>` (`FtpOutboundGateway`) to perform any operation(s) on the `Session<FTPFile>` with
`<int-sftp:outbound-gateway/>` (`SftpOutboundGateway`) to perform any operation(s) on the `Session<LsEntry>` 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:

3
src/reference/asciidoc/whats-new.adoc
View File

@@ -85,6 +85,9 @@ See <<ftp-outbound-gateway>> and <<sftp-outbound-gateway>> for more information.
The FTP and SFTP outbound gateways now support the `REPLACE_IF_MODIFIED` `FileExistsMode` when fetching remote files.
See <<ftp-outbound-gateway>> and <<sftp-outbound-gateway>> for more information.
The (S)FTP streaming inbound channel adapters now add remote file information in a message header.
See <<ftp-streaming>> and <<sftp-streaming>> 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`.