Camel - 使用 Camel Bindy 使用本地磁盘中的 CSV 丰富 FTP 中的 CSV

Question

目标是通过比较两个 CSV 文件与 使用骆驼 3.0.0。一个位于 FTP 服务器上，另一个位于磁盘上。如何结合使用轮询丰富模式和使用 Bindy 数据格式解组磁盘上的 CSV？

示例代码（为简单起见，将 FTP 端点替换为文件端点）：

@Component
public class EnricherRoute extends RouteBuilder {
    @Override
    public void configure() {
        from("file://data?fileName=part_1.csv&scheduler=quartz2&scheduler.cron=0+0+0/1+*+*+?")
            .unmarshal().bindy(BindyType.Csv, Record.class)
            .pollEnrich("file://data?fileName=part_2.csv", new ReportAggregationStrategy())
            .marshal().bindy(BindyType.Csv, Record.class)
            .to("file://reports?fileName=report_${date:now:yyyyMMdd}.csv");

    }
}

这个例子中的问题是ReportAggregationStrategy 中的resource（来自data/part_2.csv，见下文）没有被解组。如何也解组data/part_2.csv？

public class ReportAggregationStrategy implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange original, Exchange resource) {
        final List<Record> originalRecords = original.getIn().getBody(List.class);
        final List<Record> resourceRecords = resource.getIn().getBody(List.class); // Results in errors!

        ...
    }
}

Answer 1

您可以使用 direct endpoint 包装扩充内容并在那里进行解组。

from("file://data?fileName=part_1.csv&scheduler=quartz2&scheduler.cron=0+0+0/1+*+*+?")
    .unmarshal().bindy(BindyType.Csv, Record.class)
    .enrich("direct:enrich_record", new ReportAggregationStrategy())
    .marshal().bindy(BindyType.Csv, Record.class)
    .to("file://reports?fileName=report_${date:now:yyyyMMdd}.csv");

from("direct:enrich_record")
    .pollEnrich("file://data?fileName=part_2.csv")
    .unmarshal().bindy(BindyType.Csv, Record.class);