Отправлять результаты агрегации по основному маршруту, а не по входным сообщениям.

Question

Я хочу получить результат агрегации как основное сообщение маршрута, а не исходные сообщения, которые пришли на 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

Answer 1

В вашем маршруте есть ошибка. Не следует обрабатывать окончательные результаты агрегации "вне цикла", а в под маршруте. Не ставьте никаких заявлений после вашего конца ().

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}");

Как только агрегат достигнет своего размера завершения, весь агрегат будет отправлен в самую первую ближайшую конечную точку "к (...)". То, что вы хотите сделать с каждым агрегатом, должно быть смоделировано по отдельному маршруту.