Spring集成文件支持中的条件文件移动

Question

我正在实现如下所示的一个Spring集成工作流程。

IntegrationFlows.from("inputFileProcessorChannel")
                           .split(fileSplitterSpec, spec -> {})
                           .transform(lineItemTransformer)
                           .handle(httpRequestExecutingMessageHandler)
                           .transform(reportDataAggregator)
                           .aggregate(aggregatorSpec -> aggregatorSpec.requiresReply(false))
                           .channel("reportGeneratorChannel")
                           .get();

现在，完成上述流程后，我需要将输入文件移动到一个归档目录。决定目标目录的决定基于消息标头processingfailed，此标头添加到流程中的.transform(reportDataAggregator)步骤中。为了移动这些文件，我创建了另一个流，如下代码所示

 IntegrationFlows.from(MessageChannels.direct("inputFileProcessorChannel"))
                           .routeToRecipients(routerSpec -> {
                               routerSpec.recipient("processedFileMoverChannel", createMessageSelector(Boolean.FALSE))
                                         .recipient("failedFileMoverChannel", createMessageSelector(Boolean.TRUE));
                           })

                           .get();  

选择器方法

 private MessageSelector createMessageSelector(Boolean ruleBoolean) {
    return message -> ruleBoolean.equals(message.getHeaders().get("processingFailed"));
}

IntegrationFlows.from("reportGeneratorChannel")
                           .transform(executionReportTransformer)
                           .handle(reportWritingMessageHandlerSpec)
                           .get();

但是，正如该流所期望的那样，文件移动不会完成，因为所述头没有出现在流执行中。

那么，如何在报表文件创建后执行文件移动器流来实现这一目标呢？

Answer 1

Filesplitter为我们填充每一行的标题，以生成：

@Override
protected boolean willAddHeaders(Message<?> message) {
    Object payload = message.getPayload();
    return payload instanceof File || payload instanceof String;
}

@Override
protected void addHeaders(Message<?> message, Map<String, Object> headers) {
    File file = null;
    if (message.getPayload() instanceof File) {
        file = (File) message.getPayload();
    }
    else if (message.getPayload() instanceof String) {
        file = new File((String) message.getPayload());
    }
    if (file != null) {
        if (!headers.containsKey(FileHeaders.ORIGINAL_FILE)) {
            headers.put(FileHeaders.ORIGINAL_FILE, file);
        }
        if (!headers.containsKey(FileHeaders.FILENAME)) {
            headers.put(FileHeaders.FILENAME, file.getName());
        }
    }
}

因此，即使您完成了聚合并准备将消息发送到.channel（“ReportGeneratorChannel”），您仍然可以访问这些与文件相关的头。

将此ReportGeneratorChannel作为PublissubscribeChannel并将“文件移动器流”移到那里，就可以实现这一目的。

顺便说一下：到目前为止，integrationflows.from（messageChannels.direct（“inputFileProcessorChannel”））和同一通道上的第二个流将导致循环调度。那不是酒吧分销。请参阅docs中的更多信息：https://docs.spring.io/spring-integration/docs/current/reference/html/core.html#channel