如果 `RotatingServerAdvice` 没有多次获取文件，如何解决？

Question

我使用IntegrationFlow 作为Sftp 入站DSL 配置，我使用CustomTriggerAdvice 来处理手动触发。请参阅下面的代码 sn-p 以供参考。

我还使用RotatingServerAdvice 来处理同一主机中的多个路径。

但是当我启动 Sftp Inbound 时，它会第一次从每个路径中获取文件，但第二次以后就无法正常工作了。 Sftp Inbound 启动但不从路径中获取文件。我无法找出问题所在。有什么我想念的吗？

SftpConfiguration

public IntegrationFlow fileFlow() {
    SftpInboundChannelAdapterSpec spec = Sftp
            .inboundAdapter(dSF())
            .preserveTimestamp(true)
            .remoteDirectory(".")
            .autoCreateLocalDirectory(true)
            .deleteRemoteFiles(false)
            .localDirectory(new File(getDestinationLocation()));

    return IntegrationFlows
            .from(spec, e -> e.id(BEAN_ID)
                    .autoStartup(false)
                    .poller(Pollers
                            .fixedDelay(5000)
                            .advice(
                                    customRotatingServerAdvice(dSF()),
                                    customTriggerAdvice()
                            )
                    )
            )
            .channel(sftpReceiverChannel())
            .handle(sftpInboundMessageHandler())
            .get();
}

private MessageChannel sftpReceiverChannel() {
    return MessageChannels.direct().get();
}

... ... ...

@Bean
public RotatingServerAdvice customRotatingServerAdvice(
      DelegatingSessionFactory<LsEntry> dSF
) {
    List<String> pathList = getSourcePathList();
    for (String path : pathList) {
        keyDirectories.add(new RotationPolicy.KeyDirectory(KEY, path));
    }

    return new RotatingServerAdvice(
            dSF,
            keyDirectories
    );
}

@Bean
public CustomTriggerAdvice customTriggerAdvice() {
    return new CustomTriggerAdvice(customControlChannel(),BEAN_ID);
}


@Bean
public IntegrationFlow customControlBus() {
    return IntegrationFlows.from(customControlChannel())
            .controlBus()
            .get();
}


@Bean
public MessageChannel customControlChannel() {
    return MessageChannels.direct().get();
}

CustomTriggerAdvice

public class CustomTriggerAdvice extends AbstractMessageSourceAdvice {
    private final MessageChannel controlChannel;
    private final String BEAN_ID;

    public CustomTriggerAdvice(MessageChannel controlChannel, String beanID) {
        this.controlChannel = controlChannel;
        this.BEAN_ID = beanID;
    }

    @Override
    public boolean beforeReceive(MessageSource<?> source) {
        return true;
    }

    @Override
    public Message<?> afterReceive(Message<?> result, MessageSource<?> source) {
        if (result == null) {
            controlChannel.send(new GenericMessage<>("@" + BEAN_ID + ".stop()"));
        } 
        return result;
    }
}

使用 MessageChannel 启动 Sftp Inbound

@Qualifier("customControlChannel") MessageChannel controlChannel;


public void startSftpInbound(){
       controlChannel.send(new GenericMessage<>("@" + beanID + ".start()"));
}

Answer 1

我需要系统按需启动并获取文件完成一个周期。如果之后没有停止，它将继续轮询并且不会停止，我的系统将陷入无限循环。当 RotatingServerAdvice 完成从所有服务器至少一个轮询时，有什么方法可以做到这一点？它会引发任何事件或类似的事情吗？

您可能误解了afterReceive(@Nullable Message<?> result, MessageSource<?> source) 合约的逻辑。当其中一台服务器没有返回任何可供轮询的内容时，您无法根据您的要求停止通道适配器。这样您就不会给另一个服务器在下一个轮询周期轮询的机会。

我认为您的想法是只遍历所有服务器一次然后停止。可能与其中任何一个的结果无关。看起来最好的停止方法是使用RotatingServerAdvice 和fair = true 每次移动到下一个服务器。当您看到RotationPolicy.getCurrent() 等于列表中的最后一个时，可能会从自定义afterReceive() 独立于结果执行停止。因此，通过这种方式，您可以遍历所有这些并停止为下一个轮询周期移动第一个。