Message Splitter / Aggregator

Пришло составное сообщение:
  - Заказ из 100 товаров
  - Обрабатывать 100 товаров последовательно — долго
  - Обрабатывать 100 товаров параллельно — быстро

Splitter:
  - Разбивает заказ на 100 отдельных сообщений (по одному на товар)
  - Каждое сообщение обрабатывается параллельно
  - Время обработки сокращается с суммы до максимума

Пришло 100 сообщений (результаты обработки товаров):
  - Нужно собрать их в один отчёт
  - Нужно дождаться всех
  - Нужно объединить результаты

Aggregator:
  - Собирает 100 сообщений
  - Отслеживает, какие уже пришли
  - Когда все пришли — объединяет в одно
  - Отправляет дальше

1. На вход приходит составное сообщение (корзина, список, заказ)
2. Splitter анализирует структуру
3. Разбивает на отдельные элементы
4. Каждый элемент отправляется как отдельное сообщение
5. Исходное сообщение больше не существует

Код (псевдокод):
  splitter:
    input: order.queue
    output: items.exchange
    logic:
      - получить сообщение (заказ)
      - for each item in order.items:
          отправить item в items.exchange
      - подтвердить исходное сообщение

1. На вход приходят отдельные сообщения
2. Aggregator хранит их до тех пор, пока не соберётся полный набор
3. Определяет, какие сообщения относятся к одной группе (correlation id)
4. Определяет, когда группа завершена (количество, таймаут)
5. Объединяет сообщения в одно
6. Отправляет дальше

Correlation ID:
  - Идентификатор, связывающий сообщения одной группы
  - Пример: order_id, batch_id, transaction_id

Completion criteria:
  - По количеству: когда пришло N сообщений
  - По таймауту: если за 5 минут не пришли все — завершить с тем, что есть
  - По флагу: специальное сообщение "последнее"

State management:
  - Хранить накопленные сообщения
  - Redis / In-memory / База данных

1. Заказ с 10 товарами → Splitter → 10 сообщений (по товару)
2. Каждый товар обрабатывается параллельно (резервирование на складе)
3. Aggregator ждёт 10 результатов
4. Когда все пришли → собирает в один отчёт
5. Отчёт отправляется дальше

RabbitMQ:
  - Нет встроенного splitter/aggregator
  - Реализуется в коде потребителя

Kafka:
  - Нет встроенного splitter/aggregator
  - Реализуется в Kafka Streams (stateful)

Apache Camel:
  - Встроенный splitter/aggregator
  - DSL для конфигурации

Splitter (Python):
  def handle_order(order):
      for item in order.items:
          send_to_queue(item)
      ack()

Aggregator (Python):
  state = {}
  
  def handle_item(item):
      order_id = item.order_id
      if order_id not in state:
          state[order_id] = []
      state[order_id].append(item)
      
      if len(state[order_id]) == expected_count:
          result = merge(state[order_id])
          send_result(result)
          del state[order_id]

Время обработки заказа из 10 товаров:
  - Товар 1: 1 сек
  - Товар 2: 1 сек
  - ...
  - Товар 10: 1 сек
  Итого: 10 секунд

Время обработки заказа из 10 товаров:
  - Товар 1: 1 сек (параллельно)
  - Товар 2: 1 сек (параллельно)
  - ...
  - Товар 10: 1 сек (параллельно)
  Итого: 1 секунда + накладные расходы

1. API получает заказ
2. Отправляет заказ в order.queue
3. Splitter: разбивает на 10 товаров
4. 10 товаров в item.queue
5. 10 воркеров резервируют товары параллельно
6. Результаты в result.queue
7. Aggregator: ждёт 10 результатов
8. Отправляет результат клиенту

Сценарий:
  - Ожидаем 10 результатов
  - Пришло только 9
  - 10-е потерялось или обработчик упал

Варианты:
  - Ждать таймаут (например, 30 секунд)
  - По таймауту завершить с тем, что есть
  - Отправить сообщение об ошибке
  - Переотправить запрос на недостающие

Сценарий:
  - 9 товаров обработаны успешно
  - 10-й товар не может быть обработан (нет на складе)

Варианты:
  - Откатить все 9 (сложно)
  - Обработать 9, отправить отчёт с ошибкой по 10-му
  - Повторить попытку для 10-го
  - Отправить в DLQ, остальные 9 обработать