使用Spring集成SftpInboundChannel与Spring Batch结合

Question

问题描述

亲爱的Stack-Overflow-Community,我最近在Spring-Integration-Framework上看到了sfpinbound通道。

是否可以通过批处理作业启动入站通道?

我的选拔赛到现在

我已经创建了一个RunOnceTrigger，并尝试使用以下代码通过作业请求启动它的入站通道:

@Bean
@InboundChannelAdapter(channel = "fromSftpChannel",poller = @Poller(trigger = "fireSftpOnceTrigger"),autoStartup =
"false")
//@InboundChannelAdapter(channel = "fromSftpChannel", poller = @Poller(fixedRate = "1000"))
public MessageSource<InputStream> sftpMessageSource() {
SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template());
messageSource.setRemoteDirectory(sftpRemoteDirectoryDownload);
messageSource.setFilter(new AcceptAllFileListFilter<>());
messageSource.setMaxFetchSize(1);
return messageSource;
}
@Bean
public IntegrationFlow flow() {
return IntegrationFlows.from(Sftp.inboundStreamingAdapter(template())
.filter(new AcceptAllFileListFilter<>())
.remoteDirectory(sftpRemoteDirectoryDownload),
e -> e.id("sftpinboundpoller").autoStartup(false).poller(Pollers.trigger(fireSftpOnceTrigger)))
.channel(MessageChannels.queue("files"))
.get();
}

最后我用这段代码来触发我的sftp轮询器，但它似乎只触发sftp而不是文件读取作业。

SourcePollingChannelAdapterFactoryBean factoryBean =
(SourcePollingChannelAdapterFactoryBean)flow().getIntegrationComponents().keySet().stream().findFirst().get();
fireSftpOnceTrigger.reset();
factoryBean.start();

也许有人知道一个更好的方法来启动一个批处理作业的sftp入站通道?

问好Devron1705

Answer 1

查看我的错误

@Bean
public IntegrationFlow flow() {
return IntegrationFlows.from(Sftp.inboundStreamingAdapter(template())
.filter(new AcceptAllFileListFilter<>())
.remoteDirectory(sftpRemoteDirectoryDownload),
e -> e.id("sftpinboundpoller").autoStartup(false).poller(Pollers.trigger(fireSftpOnceTrigger)))
**.channel(MessageChannels.queue("files"))**
.get();
}

.channel (MessageChannels.queue("files")——比;意思是排队留言。我已经创建了一个名为fromftp的通道，它将把传入消息转换为文件流。可以通过将前面的代码行更改为.channel(messagechannes .direct("from msft"))来安排连接。