Как передать значение размера в секвенсор динамически?

Question

Как вы можете передать значение прямо из заголовка как целое число в метод размера?

from(...).resequence(body()).batch().size(${pass_some_value_from_header_as_integer}).timeout(4000L).to(....)

Спасибо за любую помощь.

Answer 1

Спасибо большое.На данный момент я решил проблему с помощью агрегатора, который также сортирует порядок сообщений.Я знаю, что это, вероятно, не идеальное решение, а обходной путь, но пока он работает.Как у меня будет время, пытаясь применить ваше решение.

    @RequiredArgsConstructor
      class SampleEventRoute extends RouteBuilder {
        @Override
        public void configure() throws Exception {
          from(DIRECT_START)
              .process(exchange -> System.out.println("Input event : " + exchange.getIn().getBody()))
              .aggregate(header(CORRELATION_ID), new ResequencerStrategy())
              .completionSize(header(SIZE)) //here i can pass messages size in dynamic way
              .completionTimeout(2000)
              .log("${body}")
              .log("Completed by ${property.CamelAggregatedCompletedBy}")
              .split().body()
              .setBody(simple("${body.payload}"))
              .process(exchange -> System.out.println("Output event  : " + exchange.getIn().getBody()))
              .end();
        }
      }

@Value
class PayloadWrapper {

    private final String payload;
    private final int order;
    private final String correlationId;

}

Answer 2

Я думаю, вы не можете использовать поле заголовка сообщения для настройки Resequencer , потому что маршрут является своего рода «статическим», в то время как заголовок сообщения является значением времени выполнения, которое может изменяться для каждого сообщения.

Что должно произойти, если партия собрала 50 сообщений, а следующая устанавливает размер партии равным 10?

Однако, несмотря на мои сомнения, BatchProcessor Camel (Resequencer расширяет BatchProcessor) знает Свойство Exchange с именем Exchange.BATCH_SIZE.

Не знаю, работает ли это, но вы можете попробовать и

• Установить размер пакета по умолчанию в маршруте (начальное значение)
• Установить ExchangeСвойство для текущего сообщения всякий раз, когда вы хотите изменить размер пакета.Очевидно, перед отправкой сообщения в Resequencer.

Установите свойство Exchange

.setProperty(Exchange.BATCH_SIZE, "yourValue")

или если вы хотите взять значение из заголовка вашего сообщения

.setProperty(Exchange.BATCH_SIZE, header("yourHeaderName"))