Весной перезаписываются сообщения с одинаковым идентификатором корреляции и порядковым номером

Question

Исходя из моих собственных экспериментов (нигде не могу найти это задокументированное), если 2 сообщения, имеющие одинаковый идентификатор корреляции и порядковый номер, агрегатор будет принимать только 1-е сообщение и игнорировать / игнорировать другое сообщение.

Есть ли способ заставить агрегатор использовать последнее полученное сообщение?

Агрегация объединит полезную нагрузку в 1 строку.

Простой сценарий: 3 сообщения с одинаковым идентификатором корреляции и размером последовательности 2, упорядоченные по времени получения

• последовательность №: 1; полезная нагрузка: abc
• последовательность №: 1; полезная нагрузка: def
• последовательность №: 2; полезная нагрузка: ги

Текущий выход: abcghi

Ожидаемый результат: defghi

Этот сценарий происходит, когда отсутствует последовательность # 2, которая предназначена для 1-го сообщения. И идентификатор корреляции (полученный из декодированной полезной нагрузки) очень ограничен, поэтому он будет использоваться несколько раз.

Оригинал сообщения

Необработанные сообщения пришли в следующем формате:

• "Размер последовательности","Номер последовательности","Идентификатор","Текст"
• ID варьируется от 0 до 9
• пример сообщения: 2,1,8,abc

Пример исходного сообщения:

• 2,1,8, а
• 2,1,8, четкости
• 2,2,8, ГХИ

Агрегатор в основном объединяет текст

Answer 1

Вам необходимо использовать собственную стратегию выпуска (она может иметь ту же логику, что и по умолчанию). SequenceSizeReleaseStrategy, но это не может быть тот класс). При использовании стратегии по умолчанию повторяющиеся последовательности пропускаются.

Однако вам также понадобится специальное хранилище групп сообщений, чтобы настроить желаемое поведение; в противном случае вывод будет abcdefghi,

Однако в этом случае отброшенное сообщение не будет отправлено на канал сброса.

Как правило, не рекомендуется повторно использовать идентификатор корреляции; если нужно, используйте group-timeout или жнец, чтобы отбросить частичную группу перед любой возможностью повторного использования корреляционного идентификатора.

Кстати, вы могли бы найти проще использовать кастом CorrelationStrategy а также ReleaseStrategy вместо использования по умолчанию и манипулирования заголовками.