java - 如何使用 Spring 集成 DSL 进行 Ftp 轮询

Question

我正在尝试为下面描述的流程创建动态 ftp 入站适配器

<int-ftp:inbound-channel-adapter id="ftpInbound"
    channel="ftpChannel" session-factory="ftpClientFactory"
    filename-pattern="*.txt" auto-create-local-directory="true"
    delete-remote-files="false" remote-directory="${remotedir}"
    local-directory="/mock/test" auto-startup="true">
    <int:poller fixed-rate="1000">
        <int:transactional synchronization-factory="syncFactory" />
    </int:poller>
</int-ftp:inbound-channel-adapter>

为此，我使用 SI java dsl 创建了以下流程，我使用 SI 动态注册进行注册，如下所示

public void createSubFlows(FtpConfig config) {

    DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
    Connection connection = config.getConnection();
    Feed feed = config.getFeed();
    sf.setHost(connection.getHost());
    sf.setPort(connection.getPort());
    sf.setUsername(connection.getUsername());
    sf.setPassword(connection.getPassword());

    IntegrationFlow flow = IntegrationFlows
            .from(Ftp.inboundAdapter(csf).preserveTimestamp(true)
                    .remoteDirectory(feed.getRemoteDirectory())
                    .regexFilter(".*\\.txt$")                   
                    .localDirectory(new File(feed.getLocalDirectory())),
                    e -> e.id("ftpInboundAdapter" + connection.getId()
                            + feed.getId())//
            ).handle(m -> System.out.println(m.getPayload())).get();
    this.flowContext.registration(flow)
    .id("ftp.flow" + connection.getId() + feed.getId()).autoStartup(true)
            .register();
}

同样的效果很好，但我还想配置一个类似于上面 xml 中的动态轮询器。是否可以对 SI Java DSL 执行相同的操作

更新

看起来它可以工作，但不确定为什么它没有出现在内容辅助中

e -> e.poller(Pollers.fixedRate(100).maxMessagesPerPoll(1))

Answer 1

您确定 .from() 中的 e 没有 .poller()`:

   from(Ftp.inboundAdapter(csf)
          .preserveTimestamp(true)
                .remoteDirectory(feed.getRemoteDirectory())
                .regexFilter(".*\\.txt$")                   
                .localDirectory(new File(feed.getLocalDirectory())),
                e -> e.id("ftpInboundAdapter" + connection.getId()
                        + feed.getId()
              .poller())

？