Spring Integration：MessageSource 不支持 errorChannel 标头

Question

我有以下流程：

@Resource(name = S3_CLIENT_BEAN)
private MessageSource<InputStream> messageSource;

public IntegrationFlow fileStreamingFlow() {
    return IntegrationFlows.from(s3Properties.getFileStreamingInputChannel())
        .enrichHeaders(spec -> spec.header(ERROR_CHANNEL, S3_ERROR_CHANNEL, true))
        .handle(String.class, (fileName, h) -> {
                    if (messageSource instanceof S3StreamingMessageSource) {
                        S3StreamingMessageSource s3StreamingMessageSource = (S3StreamingMessageSource) messageSource;

                        ChainFileListFilter<S3ObjectSummary> chainFileListFilter = new ChainFileListFilter<>();
                        chainFileListFilter.addFilters(...);
                        s3StreamingMessageSource.setFilter(chainFileListFilter);

                        return s3StreamingMessageSource.receive();
                    }                    
                    return messageSource.receive();
                }, spec -> spec
                        .requiresReply(false) // in case all messages got filtered out
        )
        .channel(s3Properties.getFileStreamingOutputChannel())
        .get();
}

我发现如果s3StreamingMessageSource.receive 抛出异常，错误最终会出现在为管道中的前一个流配置的错误通道中，而不是为此流配置的S3_ERROR_CHANNEL。不确定是否与this 问题有关。

Answer 1

s3StreamingMessageSource.receive() 是从SourcePollingChannelAdapter 调用的：

protected Message<?> receiveMessage() {
    return this.source.receive();
}

这个来自AbstractPollingEndpoint：

private boolean doPoll() {

        message = this.receiveMessage();
...

        this.handleMessage(message);
...
}

handleMessage() 这样做：

this.messagingTemplate.send(getOutputChannel(), message);

所以，那肯定离下游提到的.enrichHeaders(spec -> spec.header(ERROR_CHANNEL, S3_ERROR_CHANNEL, true))还很远。

但是，您仍然可以在该 S3_ERROR_CHANNEL 中捕获异常。注意IntegrationFlows.from()的第二个参数：

IntegrationFlows.from(s3Properties.getFileStreamingInputChannel(),
           e -> e.poller(Pollers.fixedDelay(...)
                       .errorChannel(S3_ERROR_CHANNEL)))

或者，根据您当前的情况，您在某个地方有一个全局轮询器，所以在那里配置一个errorChannel。