SFTP 适配器正在跳过备用文件

问题描述

我有一个 SFTP 适配器，它可以从远程位置下载文件并进行转换。但是它跳过了备用文件。即如果在 SFTP HOST 文件是 1.zip,2.zip,3.zip ，那么它只处理 1.zip 和 3.zip

@Bean
@Primary
public IntegrationFlow sftpInboundFlow(){
...
..
 SftpInboundChannelAdapterSpec messageSourceBuilder = 
...
..

IntegrationFlowBuilder flowBuilder = IntegrationFlows
                .from(messageSourceBuilder,consumerSpec())
                .log(Level.INFO,m -> "INBOUND: " + m.getPayload() + " HEADERS: " + m.getHeaders()
                );


return flowBuilder.channel(INBOUND_CHANNEL).handle(new MessageHandler());

// Works fine if changed to 
// flowBuilder.channel(INBOUND_CHANNEL).get();
// 
}




    @Bean
    public IntegrationFlow uncompressionfileFlow() {

        UnZipTransformer unZipTransformer = new UnZipTransformer();

        IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(INBOUND_CHANNEL).transform(unZipTransformer)
                .split(new UnZipAbstractMessageSplitter(prop1,prop2))
                .log(Level.INFO,m -> "OUTBOUND: " + m.getPayload() + " HEADERS: " + m.getHeaders())
                .enrichHeaders(h -> h.headerExpression(FileHeaders.ORIGINAL_FILE,"payload.headers['" + FileHeaders.FILENAME + "']"));
        return flowBuilder.channel(OUTBOUND_CHANNEL).get();
    }

解决方法

您所描述的内容与 DirectChannel 上的循环调度策略完全相关：https://docs.spring.io/spring-integration/docs/current/reference/html/core.html#channel-implementations-directchannel。根据您的配置，我们确实有两个订阅者订阅了同一个 INBOUND_CHANNEL：

1. channel(INBOUND_CHANNEL).handle(new MessageHandler())。
2. from(INBOUND_CHANNEL).transform(unZipTransformer)

我不确定您的目标是什么，但该代码段中的逻辑比仅从 SFTP 轮询文件并处理它们复杂得多。

您应该修改目前已有的内容，但这不是 SFTP 入站通道适配器问题，而更像是同一直接通道上的两个相互竞争的消费者。