FTP 流入站通道适配器
版本 4.3 引入了流入站通道适配器。
此适配器生成 payload 类型的 messageInputStream，让文件在不写入
local 文件系统。
由于会话保持打开状态，因此使用应用程序负责在文件
消耗。
该会话在closeableResource标头 （IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE).
标准框架组件（例如FileSplitter和StreamTransformer，则会自动关闭会话。
有关这些组件的更多信息，请参见File Splitter 和 Stream Transformer。
以下示例说明如何配置inbound-streaming-channel-adapter:
<int-ftp:inbound-streaming-channel-adapter id="ftpInbound"
channel="ftpChannel"
session-factory="sessionFactory"
filename-pattern="*.txt"
filename-regex=".*\.txt"
filter="filter"
filter-expression="@myFilterBean.check(#root)"
remote-file-separator="/"
comparator="comparator"
max-fetch-size="1"
remote-directory-expression="'foo/bar'">
<int:poller fixed-rate="1000" />
</int-ftp:inbound-streaming-channel-adapter>只有其中之一filename-pattern,filename-regex,filter或filter-expression是允许的。
从版本 5.0 开始，默认情况下，FtpStreamingMessageSource适配器可防止远程文件重复FtpPersistentAcceptOnceFileListFilter基于 In-MemorySimpleMetadataStore.
默认情况下，此过滤器也与文件名模式（或 regex）一起应用。
如果需要允许重复项，可以使用AcceptAllFileListFilter.
任何其他用例都可以由CompositeFileListFilter（或ChainFileListFilter).
Java 配置（本文档后面部分）显示了一种在处理后删除远程文件以避免重复的技术。 |
有关FtpPersistentAcceptOnceFileListFilter及其使用方式，请参阅远程持久性文件列表过滤器。
使用max-fetch-size属性来限制在需要提取时每次轮询时提取的文件数。
将其设置为1并在集群环境中运行时使用持久性过滤器。
有关更多信息，请参见入站通道适配器：控制远程文件获取。
适配器将远程目录和文件名放在FileHeaders.REMOTE_DIRECTORY和FileHeaders.REMOTE_FILE标头。
从版本 5.0 开始，FileHeaders.REMOTE_FILE_INFOheader 提供额外的远程文件信息（默认以 JSON 表示）。
如果将fileInfoJson属性FtpStreamingMessageSource自false，标头包含一个FtpFileInfo对象。
这FTPFile对象，可以使用FtpFileInfo.getFileInfo()方法。
这fileInfoJson属性在使用 XML 配置时不可用，但您可以通过注入FtpStreamingMessageSource添加到您的配置类之一中。
另请参阅远程文件信息。
从版本 5.1 开始，comparator是FTPFile.
以前，它是AbstractFileInfo<FTPFile>.
这是因为排序现在在筛选和应用之前在处理的早期执行maxFetch.
使用 Java 配置进行配置
以下 Spring Boot 应用程序显示了如何使用 Java 配置配置入站适配器的示例：
@SpringBootApplication
public class FtpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
@InboundChannelAdapter(channel = "stream")
public MessageSource<InputStream> ftpMessageSource() {
FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template());
messageSource.setRemoteDirectory("ftpSource/");
messageSource.setFilter(new AcceptAllFileListFilter<>());
messageSource.setMaxFetchSize(1);
return messageSource;
}
@Bean
@Transformer(inputChannel = "stream", outputChannel = "data")
public org.springframework.integration.transformer.Transformer transformer() {
return new StreamTransformer("UTF-8");
}
@Bean
public FtpRemoteFileTemplate template() {
return new FtpRemoteFileTemplate(ftpSessionFactory());
}
@ServiceActivator(inputChannel = "data", adviceChain = "after")
@Bean
public MessageHandler handle() {
return System.out::println;
}
@Bean
public ExpressionEvaluatingRequestHandlerAdvice after() {
ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
advice.setOnSuccessExpression(
"@template.remove(headers['file_remoteDirectory'] + headers['file_remoteFile'])");
advice.setPropagateEvaluationFailures(true);
return advice;
}
}请注意，在此示例中，转换器下游的消息处理程序具有advice，这将在处理后删除远程文件。