IntegrationFlowDefinition.aggregate не работает: может быть, CorrelationStrategy не работает?

Question

Сообщение об ошибке: Причина: java .lang.IllegalStateException: Корреляция с нулевым значением недопустима. Может быть, CorrelationStrategy не работает?

Моя реализация,

    @Bean
    public IntegrationFlow start() {
        return IntegrationFlows
                .from("getOrders")
                .split()
                .publishSubscribeChannel(c -> c.subscribe(s -> s.channel(q -> q.queue(1))
                        .<Order, Message<?>>transform(p -> MessageBuilder.withPayload(new Item(p.getItems())).setHeader(ORDERID, p.getOrderId()).build())
                        .split(Item.class, Item::getItems)
                        .transform() // let's assume, an object created for each item, let's say ItemProperty to the object.
                                     //  Transform returns message; MessageBuilder.withPayload(createItemProperty(getItemName, getItemId)).build();
                        .aggregate() // so, here aggregate method needs to aggregate ItemProperties.
                        .handle() // handler gets List<ItemProperty> as an input.
      ))
      .get();
    }

Оба сплиттера работают нормально. Еще тестировал трансформатор после второго разветвителя, работает нормально. Но когда дело доходит до агрегирования, это не удается. Что мне здесь не хватает?

Answer 1

Вам не хватает того факта, что transformer - это тот тип конечной точки, который обрабатывает все сообщение как есть. И если вы создаете сообщение самостоятельно, оно не меняет его. Итак, с вашим MessageBuilder.withPayload(createItemProperty(getItemName, getItemId)).build(); вы просто пропустите важные заголовки деталей последовательности после разделителя. Следовательно, агрегатор после этого не знает, что делать с вашим сообщением, поскольку вы настраиваете его для стратегий корреляции по умолчанию, но вы не предоставляете соответствующие заголовки в сообщении.

Технически я не вижу причин для создайте там сообщение вручную: простого return createItemProperty(getItemName, getItemId); должно хватить. И фреймворк возьмется за создание сообщения от вашего имени с соответствующим копированием заголовков сообщения запроса.

Если вы действительно все еще думаете, что вам нужно создать сообщение самостоятельно в этом преобразовании, тогда вам нужно подумать о copyHeaders() на этом MessageBuilder из сообщения запроса для переноса заголовков требуемых деталей последовательности.