如何使用Spring integration 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 dynamic registration注册这些流，如下所示

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())

？