Функция Spring Cloud Stream: функция вызова через вызов REST и вывести его в тему KAFKA

Question

У меня есть простые @Bean (Java 8 функций), которые отображаются на адрес назначения topic (-out и -in).

@Bean
public Function<String, String> transform() {
    return payload -> payload.toUpperCase();
}

@Bean
public Consumer<String> receive() {
    return payload -> logger.info("Data received: " + payload);
}

.yml config :

spring:
  cloud:
    stream:
      function:
        definition: transform;receive
      bindings:
        transform-out-0:
          destination: myTopic
        receive-in-0:
          destination: myTopic

Теперь я хочу вызвать функцию transform с помощью вызова REST, чтобы ее вывод шел в destination topic (т.е. transform-out-0 сопоставлен с myTopic ) и забирается consumer из этого пункта назначения (receive-in-0 сопоставлен с myTopic). По сути, каждый вызов REST должен порождать новый экземпляр KAFKA Producer и закрывать его.

Как я могу этого добиться, используя spring-cloud-stream?

Спасибо

Ангшуман

Answer 1

Вы должны использовать StreamBridge вместо этой transform функции. Это новый рекомендуемый подход для динамических c пунктов назначения в Spring Cloud Stream. Вот основная идея c:

@Autowired
private StreamBridge streamBridge;

@RequestMapping
public void delegateToSupplier(@RequestBody String body) {
    streamBridge.send("transform-out-0", body);
}

, а затем предоставьте это свойство через конфигурацию - spring.cloud.stream.source: transform

Spring Cloud Stream создаст для вас выходную привязку с именем transform-out-0. Каждый раз, когда вызывается конечная точка REST через StreamBridge, вы будете отправлять данные в пункт назначения topi c.

Для получения дополнительной информации см. this .