SFTP 适配器正在跳过备用文件
SFTP Adapter is skipping alternate files
我有一个 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
:
channel(INBOUND_CHANNEL).handle(new MessageHandler())
.
from(INBOUND_CHANNEL).transform(unZipTransformer)
我不确定您的目标是什么,但该代码片段中的逻辑比仅从 SFTP 轮询文件并处理它们要复杂得多。
您应该修改到目前为止的内容,但这不是 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
:
channel(INBOUND_CHANNEL).handle(new MessageHandler())
.from(INBOUND_CHANNEL).transform(unZipTransformer)
我不确定您的目标是什么,但该代码片段中的逻辑比仅从 SFTP 轮询文件并处理它们要复杂得多。
您应该修改到目前为止的内容,但这不是 SFTP 入站通道适配器问题,而更像是同一个直接通道上的两个竞争消费者。