Используйте преобразователь JSON в Spring Integration

Question

У меня есть проблема, которая кажется нерешенной ни в одном из примеров, которые я могу найти.

Мое приложение читает тему ActiveMQ о сообщениях JSON. Он построит совершенно новый исходящий вызов REST на основе этих данных. Обратите внимание, что это не «трансформация». Дана буква "X", произведена буква "Y", т.е. ServiceActivator.

Мои потоки до сих пор

public IntegrationFlow splitInputFlow() {
    return IntegrationFlows.from("inboundJmsChannel")
            .split()
            .log(LoggingHandler.Level.DEBUG)
            .route(Message.class, m -> m.getHeaders().get("x-bn-class").equals("Healthcheck.class") ? "healthcheckChannel" : "metricChannel")
            .get();
}

public IntegrationFlow healthcheckFlow() {
    return IntegrationFlows.from("healthcheckChannel")
            .log(LoggingHandler.Level.DEBUG)
        .transform(Transformers.fromJson(Healthcheck.class))
        .handle("healthcheckActivator", "process")
        .get();
}

Есть десятки примеров того, как использовать пружинные трансформаторы. Я даже подумывал попробовать MessageConverter. Но я не понимаю, почему это помогло бы, и это не кажется нормальным подходом.

Основная проблема заключается в том, что интеграция вызывает healthcheckActivator.process (String payload). Сама полезная нагрузка является ожидаемой допустимой строкой JSON.

Я немного удивлен, что он не вызывает healtcheckActivator.process (полезная нагрузка сообщения), но это не поможет, так что это не имеет большого значения.

Реальный вопрос - почему он не вызывает healtcheckActivator.process (Healthcheck healthcheck)?

Ну, на самом деле я понимаю, "почему". Это потому, что DSL генерирует внутренний канал, чтобы связать шаги вместе, и, насколько я понимаю, что-то на канале - это spring.messaging.Message.

Я могу легко создать экземпляр моего объекта Healthcheck, как только попаду внутрь SA. Но это оставляет насущный вопрос: что может быть хорошего во всем шаге преобразования? Если он всегда «сериализует» объект обратно в Сообщение - какой смысл.

Как я уже сказал, я думаю, что мне здесь не хватает чего-то фундаментального.

EDIT Моя новая (и, вероятно, последняя) идея заключается в том, что, возможно, я публикую это неправильно.

Для публикации я использую

jmsTemplate.convertAndSend(topicName, healthcheck, messagePostProcessor -> {
            messagePostProcessor.setJMSType("TextMessage");
            messagePostProcessor.setStringProperty("x-bn-class", "Healthcheck.class");
            messagePostProcessor.setStringProperty("x-bn-service-name", restEndpoint.getServiceName());
            messagePostProcessor.setStringProperty("x-bn-service-endpoint-name", restEndpoint.getEndpointName());
            messagePostProcessor.setLongProperty("x-bn-heathcheck-timestamp", queryDate);
            messagePostProcessor.setStringProperty("x-bn-healthcheck-status", subsystemStatus.getStatus(subsystemStatus));
            messagePostProcessor.setIntProperty("httpStatus", httpStatus.value());
            return messagePostProcessor;
        });

В методе SI-процесса (String payload) поступает следующее:

LoggingHandler - GenericMessage [payload={"healthcheckType":"LOCAL","outcome":"PASS","dependencyType":"DB","endpoint":"NODE TABLE","description":"Read from DB","durationSecs":0.025}, headers={x-bn-service-name=TG10-CS2, x-bn-service-endpoint-name=TG Q10-CS2 Ready Check, jms_destination=topic://HEALTH_MONITOR, _type=com.healthcheck.response.Healthcheck, x-bn-heathcheck-timestamp=1558356538000, priority=4, jms_timestamp=1558356544244, x-bn-healthcheck-status=SEV0, jms_redelivered=false, x-bn-class=Healthcheck.class, httpStatus=200, jms_type=TextMessage, id=b29ffea7-7128-c543-9a14-8bab450f0ac6, jms_messageId=ID:39479-1558356520091-1:2:1:1:1, timestamp=1558356544409}]

Я раньше не замечал параметр _type в заголовке jms_destination. Но прежде чем я начал разбираться с этим (потому что это не сработало), это правильное имя класса для того, что предоставила другая команда.

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

Answer 1

Ваше понимание верно; у меня отлично работает, так что происходит что-то еще

@SpringBootApplication
public class So56169938Application {

    public static void main(String[] args) {
        SpringApplication.run(So56169938Application.class, args);
    }

    @Bean
    public IntegrationFlow flow() {
        return IntegrationFlows.from(() -> "{\"foo\":\"bar\"}", e -> e.poller(Pollers.fixedDelay(5000)))
            .transform(Transformers.fromJson(Foo.class))
            .handle("myBean", "method")
            .get();
    }

    @Bean
    public MyBean myBean() {
        return new MyBean();
    }

    public static class MyBean {

        public void method(Foo foo) {
            System.out.println(foo);
        }

    }

    public static class Foo {

        private String foo;

        String getFoo() {
            return this.foo;
        }

        void setFoo(String foo) {
            this.foo = foo;
        }

        @Override
        public String toString() {
            return "Foo [foo=" + this.foo + "]";
        }

    }

}

и

Foo [foo=bar]
Foo [foo=bar]
Foo [foo=bar]
Foo [foo=bar]
Foo [foo=bar]
Foo [foo=bar]

Answer 2

Ну, Spring Integration - это фреймворк Messaging .Он передает сообщения от конечной точки к конечной точке по каналам между ними.Это уже целевая ответственность конечной точки, чтобы правильно обрабатывать использованное сообщение.Фреймворк не заботится о payload.Это действительно бизнес-часть целевого приложения.Таким образом, мы можем сделать компоненты инфраструктуры как можно более общими, оставляя место для целевых бизнес-типов для конечных пользователей.

В любом случае Framework предоставляет некоторые механизмы для взаимодействия с полезными нагрузками.Мы называем это вызов метода POJO .Таким образом, вы предоставляете бизнесу произвольный контракт, однако придерживаясь некоторых правил Spring Integration: https://docs.spring.io/spring-integration/docs/current/reference/html/#service-activator.

Итак, согласно вашему описанию, действительно удивительно, что он не работает для healtcheckActivator.process(Healthcheck healthcheck).Ваш transform(Transformers.fromJson(Healthcheck.class)) должен действительно создать Message с объектом Healthcheck в качестве полезной нагрузки.Платформа проверяет сигнатуру метода и пытается сопоставить полезную нагрузку и / или заголовки с аргументами вызова метода, имея целое сообщение в качестве контейнера для данных для делегирования вызову метода.

Отсюда было бы здоровочтобы увидеть ваш healtcheckActivator.process() метод, чтобы определить, почему результат transform(Transformers.fromJson(Healthcheck.class)) можно сопоставить с аргументами этого метода.