java - 使用 Spring Boot 的 Spring Integration SFTP 示例

Question

我们正在为 Spring 应用程序使用最新的 Spring Boot，并为 SFTP 使用最新的 Spring Integration。我去过 Spring Integration SFTP 文档站点，我按原样获取了 Spring Boot 配置:

 @Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
    DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
    factory.setHost("localhost");
    factory.setPort(port);
    factory.setUser("foo");
    factory.setPassword("foo");
    factory.setAllowUnknownKeys(true);
    return new CachingSessionFactory<LsEntry>(factory);
}

@Bean
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
    SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
    fileSynchronizer.setDeleteRemoteFiles(false);
    fileSynchronizer.setRemoteDirectory("/");
    fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter("*.xml"));
    return fileSynchronizer;
}

@Bean
@InboundChannelAdapter(channel = "sftpChannel")
public MessageSource<File> sftpMessageSource() {
    SftpInboundFileSynchronizingMessageSource source =
            new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
    source.setLocalDirectory(new File("ftp-inbound"));
    source.setAutoCreateLocalDirectory(true);
    source.setLocalFilter(new AcceptOnceFileListFilter<File>());
    return source;
}

@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler handler() {
    return new MessageHandler() {

        @Override
        public void handleMessage(Message<?> message) throws MessagingException {
            System.out.println(message.getPayload());
        }

    };
}

让我说清楚，在剪切和粘贴之后，运行了一些单元测试。但是，在加载应用程序上下文时出现错误消息，因为轮询不存在。

当我用谷歌搜索该错误时，StackOverflow 上的其他帖子说我还必须添加以在加载应用程序上下文时删除此错误消息。

@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata defaultPoller() {

    PollerMetadata pollerMetadata = new PollerMetadata();
    pollerMetadata.setTrigger(new PeriodicTrigger(60));
    return pollerMetadata;
}

当我添加此代码时，那么至少我的构建可以运行并且测试可以运行，因为应用程序上下文现在已正确加载。

现在我正在寻找有关如何实现此功能和移动文件的代码示例？ GitHub 上的 Spring Integration SFTP 示例还可以，但不是很好……远非如此。

Basic Spring Integration Example 展示了如何从 SFTP 服务器读取文件(如果数据是使用 application-context.xml 文件配置的)。使用 Spring Boot 配置的示例在哪里，然后是从该服务器读取的代码，以及用于测试的代码？

我明白，无论您是使用 Java 类进行 Spring Boot 配置还是使用 application-context.xml 文件……对于 Autowiring 的 SFTP channel 和一些入站 channel 适配器，工作代码都应该相同。

所以这是代码，我正在努力工作:

@Component
@Profile("sftpInputFetch")
public class SFTPInputFetcher implements InputFetcher
{
    // The PollableChannel seems fine
    @Autowired
    PollableChannel sftpChannel;

    @Autowired
    SourcePollingChannelAdapter sftpChannelAdapter;

@Override
public Stream<String> fetchLatest() throws FileNotFoundException
{
    Stream<String> stream = null;
    sftpChannelAdapter.start();
    Message<?> received = sftpChannel.receive();
    File file = (File)received.getPayload();
    // get Stream<String> from file
    return stream;
}

目前，“sftpChannelAdapter.start();”是我遇到麻烦的部分。此实现未找到“SourcePollingChannelAdapter”类。

如果这是在带有“id”的经典 XML 应用程序上下文中定义的，那么这段代码会 Autowiring 。使用 Spring Boot 配置，您似乎无法为 bean 定义“id”。

这只是因为我不了解如何从在代码中使用带注释的传统应用程序上下文 XML 文件转换为使用完整的 Spring Boot 应用程序上下文配置文件。

非常感谢您对此提供的任何帮助。谢谢!

Answer 1

我不明白这个问题；你说

I had to add ... to make it work

然后

Now I am looking for a code sample on how to make this work?

什么不起作用？

你也可以使用

@InboundChannelAdapter(value = "sftpChannel", poller = @Poller(fixedDelay = "5000"))

而不是添加默认轮询器定义。

我们会fix the docs for the missing poller config .

编辑

我只是将代码复制到一个新的启动应用程序(使用轮询器配置)并且它按预期工作。

@SpringBootApplication
public class SftpJavaApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(SftpJavaApplication.class).web(false).run(args);
    }

    @Bean
    public SessionFactory<LsEntry> sftpSessionFactory() {
        DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
        factory.setHost("...");
        factory.setPort(22);
        factory.setUser("...");
        factory.setPassword("...");
        factory.setAllowUnknownKeys(true);
        return new CachingSessionFactory<LsEntry>(factory);
    }

    @Bean
    public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
        SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
        fileSynchronizer.setDeleteRemoteFiles(false);
        fileSynchronizer.setRemoteDirectory("foo");
        fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter("*.txt"));
        return fileSynchronizer;
    }

    @Bean
    @InboundChannelAdapter(channel = "sftpChannel", poller = @Poller(fixedDelay = "5000"))
    public MessageSource<File> sftpMessageSource() {
        SftpInboundFileSynchronizingMessageSource source = new SftpInboundFileSynchronizingMessageSource(
                sftpInboundFileSynchronizer());
        source.setLocalDirectory(new File("ftp-inbound"));
        source.setAutoCreateLocalDirectory(true);
        source.setLocalFilter(new AcceptOnceFileListFilter<File>());
        return source;
    }

    @Bean
    @ServiceActivator(inputChannel = "sftpChannel")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.out.println(message.getPayload());
            }

        };
    }

}

结果:

16:57:59.697 [task-scheduler-1] WARN  com.jcraft.jsch - Permanently added '10.0.0.3' (RSA) to the list of known hosts.
ftp-inbound/bar.txt
ftp-inbound/baz.txt