Spring Integration, глобально перенаправляет недопустимые сообщения в другое место назначения - PullRequest
Новая
       31

Spring Integration, глобально перенаправляет недопустимые сообщения в другое место назначения

0 голосов
/ 12 октября 2018

Настройка

У меня есть приложение обмена сообщениями JMS, в которое поступают сообщения от нескольких адресатов JMS.Полезная нагрузка сообщения - это различные представления JSON с некоторыми общими заголовками.Я полагаюсь на динамическое преобразование типов Джексона в Spring на ServiceActivators для преобразования в настоящие POJO.В настоящее время маршрутизация тривиальна, потому что каналы по сути являются каналами типа данных, разделенными по типу полезной нагрузки JSON (все они являются полезными нагрузками JSON String, но JSON представляет очень разные типы объектов).

Проблема

Я хотел бы применить глобальную логику проверки ко всем входящим сообщениям по нескольким сопоставленным с шаблоном каналам, например, "*input*", и перенаправить недействительные сообщения в канал ошибок проверки для просмотра.Независимо от того, является ли сообщение действительным или недействительным, локальная транзакция JMS должна быть зафиксирована;если сообщение недействительно, я не хочу, чтобы неверное сообщение было повторно отправлено позже.

Рассмотрены возможные варианты

Канальный перехватчик

Сначала я задумался о реализации ChannelInterceptor, который соответствует всем каналам, где должна применяться эта логика, но она не появляетсячто возможность перенаправить сообщение может быть реализована в ChannelInterceptor.Похоже, что у меня есть два варианта с ChannelInterceptor:

  1. Откат транзакции JMS, когда перехватчик возвращает ноль на preSend, ИЛИ
  2. недействительное сообщение все ещеотправлено вместе с первоначальным адресатом.

Ни одно из этих действий не является желаемым.Локальная транзакция JMS должна всегда фиксироваться (при условии отсутствия другой ошибки), и сообщение либо отправляется в исходное место назначения, либо перенаправляется в неверный канал сообщений.

Маршрутизатор

A Router может быть хорошим выбором, но не похоже, что есть способ применить маршрутизатор к набору каналов, сопоставленных с шаблоном, поэтому я считаю, чтопридется применить его к каждому каналу в отдельности.Такого рода дублирование я надеюсь избежать.

AspectJ Pointcut

Другой вариант, о котором я подумал, - это запустить AspectJ и реализовать @Around рекомендацию по AbstractMessageSendingTemplate.convertAndSend(destination, payload, postProcessor)метод.Это кажется навязчивым, но кажется, что это может работать.Если есть опция, лучше поддерживаемая непосредственно фреймворком, я буду рад ее услышать.

Общий входной канал с маршрутизацией полезной нагрузки

Если я не могу найти способ применить это глобальнотип логики маршрутизации, тогда другой вариант может заключаться в маршрутизации всех входящих сообщений JMS через один канал.Пользовательский Router может быть применен к тому входящему каналу, который использует заголовки типа полезной нагрузки, чтобы направлять сообщения в их надлежащие каналы типа данных и направлять недопустимые сообщения в канал ошибок проверки.

Вопрос (ы)

  • Есть ли способ применить этот тип переадресации сообщений к набору каналов, соответствующих шаблону?
  • Не упустил ли я важную возможность среды Spring Integration, которая поможет выполнить одно из моих соображений?
  • Если нет, есть ли лучшие варианты EIP, чем я упомянул?

Большое спасибо!

Ответы [ 2 ]

0 голосов
/ 13 октября 2018

Моя позиция не делает этого в глобальном ChannelInterceptor, потому что легко попасть в шаблон с каналом, на который не должна влиять такая логика фильтрации.

Вы всегда можете отправить всесообщения на тот же канал для общей логики.Поведение маршрутизации, которым вы можете управлять через заголовок replyChannel, заполненный перед отправкой в ​​канал проверки.Итак, для меня логика выглядит так:

  1. Каждый поток выполняет HeaderEnricher для заполнения заголовка replyChannel требуемым следующим шагом в потоке.

  2. После этого все потоки обогащаются, отправляя сообщения компоненту Filter с логикой проверки.

  3. Там, при неудаче, вы отправляете сообщение на discardChannelFilter, как вы объяснили в своем вопросе.

  4. В случае успеха вы просто никуда не отправляете, если только заголовок replyChannel.Итак, ваши действительные сообщения возвращаются в свои исходные потоки.

Имеет ли это смысл для вас?

0 голосов
/ 12 октября 2018

Первоначально я думал о реализации ChannelInterceptor, который соответствует всем каналам, где должна применяться эта логика, но не похоже, что возможность переадресации сообщения может быть реализована в ChannelInterceptor

Что заставляет вас в это верить?preSend() может вернуть null, что эффективно завершает операцию;просто отправьте неудачную проверку в общий канал и верните null. 

/**
 * Invoked before the Message is actually sent to the channel.
 * This allows for modification of the Message if necessary.
 * If this method returns {@code null} then the actual
 * send invocation will not occur.
 */
@Nullable
default Message<?> preSend(Message<?> message, MessageChannel channel) {
    return message;
}

Это вызовет MessageDeliveryException на входящем адаптере, но вы можете просто принять это в потоке канала ошибок.

...