0

我想将聚合的结果作为路由的主要消息，而不是 IN 上的原始消息。我也想在一条路线上做到这一点。

我知道我可以.to("direct:one_result")在聚合之后使用，但是我在一个路由中这样做有很大的限制，因为我正在动态生成路由。

我的.to("mock:out")将被更长的路线定义所取代。

    from("direct:in").routeId("TEST_AGGREGATION_ROUTE")
    .log("<IN> ${body}")
    .aggregate(header("THE_ID"), (oldExchange, newExchange) -> {
        final List<Object> body;
        final Exchange outExchange;

        if (oldExchange == null) {
            outExchange = newExchange;
            body = new ArrayList<>();
            body.add(newExchange.getIn().getBody());
        } else {
            outExchange = oldExchange;
            body = oldExchange.getIn().getBody(List.class);
            body.add(newExchange.getIn().getBody());
        }

        outExchange.getIn().setBody(body);
        return outExchange;
              })
        .completionSize(4)
        .completionTimeout(30000)

        .log("<AGGREGATION> size = ${body.size}") // HERE I GET THE AGGREGATION RESULT
    .end()
    .log("<OUT> ${body}") // HERE I GET THE INPUT MESSAGES
    .to("mock:out")
    ;

测试输出如下所示：

TEST_AGGREGATION_ROUTE - <IN> BODY1
TEST_AGGREGATION_ROUTE - <OUT> BODY1

TEST_AGGREGATION_ROUTE - <IN> BODY2
TEST_AGGREGATION_ROUTE - <OUT> BODY2

TEST_AGGREGATION_ROUTE - <IN> BODY3
TEST_AGGREGATION_ROUTE - <OUT> BODY3

TEST_AGGREGATION_ROUTE - <IN> BODY4
TEST_AGGREGATION_ROUTE - <AGGREGATION> size = 4
TEST_AGGREGATION_ROUTE - <OUT> BODY4
4

1 回答 1

0

你的路由有错误。您不应在“循环外”而是在子路由中处理聚合的最终结果。不要在 end() 之后添加任何语句。

from("direct:in")
    ...
    .aggregate(header("THE_ID"), (oldExchange, newExchange) -> {...})
       .completionSize(4)
       .completionTimeout(30000)
       .to("direct:processAggregation")
     .end();

from("direct:processAggregation")
    .log("<AGGREGATION> size = ${body.size}") 
    .log("<OUT> ${body}");

一旦聚合达到其完成大小，整个聚合将被发送到第一个下一个“to(...)”端点。所以你想对每个聚合做什么应该我在一个单独的路线中建模。

于 2019-02-08T21:33:29.090 回答