Эквивалент enrichHeaders в Spring-интеграции 5.1.1

Question

Следующий код был переписан в версии 1.2.1 spring -gration-java-dsl

IntegrationFlows.from(inMessageChannel())//
                    .enrichHeaders(
                            c -> c.<InMessage>headerFunction(HEADER_IN_MESSAGE_ID,
                                    (message) -> message.getPayload().getInMessage().getId())//
                                    .<InMessage>headerFunction(HEADER_ORIGINAL_MESSAGE_ID,
                                            (message) -> message.getPayload().getInMessage().getMessageId()),
                            c -> c.poller(Pollers.fixedDelay(100)//
                                    .transactional(platformTransactionManager)//
                                    .errorChannel(errorChannel())))//
                    .handle(...)

Подробности: inMessageChannel() - это канал Pollable

errorChannel()является SubscribeableChannel

Функция enrichHeaders() изменилась с тех пор, и я не могу найти эквивалентный код для записи в версии 5.1.1 (spring-интеграция-ядро).Пытаясь использовать разные подходы (помимо прочего, вызывать enrichHeaders один раз для каждого потребителя или использовать метод моста), я получаю следующее исключение:

java.lang.IllegalArgumentException: A poller should not be specified for endpoint 'myFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#1', since 'myFlow.channel#0' is a SubscribableChannel (not pollable).

Если я правильно понимаю код 1.2.1, средство добавления добавляется вinMessageChannel(), в новой версии он пытается добавить опрашивающее устройство к errorChannel(), и это не разрешено.

Есть идеи?Спасибо!

PS: очевидно, я новичок в весенней интеграции, которая должна адаптировать некоторый код:)

Answer 1

Начиная с версии 5.0, это enrichHeaders() действительно изменилось.То, что мы сделали, это просто слияние двух Consumer с.Итак, в настоящее время вам необходимо настроить его следующим образом:

        .enrichHeaders(
                        c -> c.<InMessage>headerFunction(HEADER_IN_MESSAGE_ID,
                                (message) -> message.getPayload().getInMessage().getId())//
                                .<InMessage>headerFunction(HEADER_ORIGINAL_MESSAGE_ID,
                                        (message) -> message.getPayload().getInMessage().getMessageId())
                                .poller(Pollers.fixedDelay(100)//
                                        .transactional(platformTransactionManager)//
                                        .errorChannel(errorChannel())))//

См. Дополнительную информацию в Руководстве по миграции: https://github.com/spring-projects/spring-integration/wiki/Spring-Integration-4.3-to-5.0-Migration-Guide#java-dsl-breaking-changes