Spring Cloud Stream - несколько функций

Можно ли использовать 2 функции, где выход первой функции является входом второй?

Мои функции:

@Configuration
public class StringStream {

    @Bean
    public Supplier<Flux<String>> stringSupplier() {
        return () -> Flux.interval(Duration.of(1L, ChronoUnit.SECONDS))
                .map(v -> RandomStringUtils.randomAlphabetic(10));
    }

    @Bean
    public Function<KStream<Bytes, String>, KStream<Bytes, String>> reverse() {
        return strings -> strings.mapValues(StringUtils::reverse);
    }

    @Bean
    public Function<KStream<Bytes, String>, KStream<Bytes, String>> uppercase() {
        return strings -> strings.mapValues((ValueMapper<String, String>) String::toUpperCase);
    }

    @Bean
    public Consumer<String> print() {
        return message -> LOGGER.info("==> {}", message);
    }
}

Моя конфигурация:

...
spring.cloud.stream:
  function:
    definition: stringSupplier;uppercase;reverse;print

spring.cloud.stream.bindings.stringSupplier-out-0:
  content-type: text/plain
  destination: strings-random

spring.cloud.stream.bindings.reverse-in-0:
  destination: strings-random
spring.cloud.stream.bindings.reverse-out-0:
  destination: strings-reversed

spring.cloud.stream.bindings.uppercase-in-0:
  destination: strings-reversed
spring.cloud.stream.bindings.uppercase-out-0:
  destination: strings-uppercase

spring.cloud.stream.bindings.print-in-0:
  destination: strings-uppercase

Он отправляет только строки в strings-reversed. Это ограничение или я что-то не так делаю?


spring-cloud-stream spring-cloud-stream-binder-kafka
person Ben Keil    schedule 06.01.2021    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Да, решение состоит в том, чтобы использовать оператор | для объединения функций.

Например, эта конфигурация должна объединять все ваши операции.

spring.cloud.stream:
  function:
    definition: stringSupplier|uppercase|reverse|print

Под капотом spring-cloud-stream используется функция spring-cloud-function для обеспечения функциональной композиции фреймворка.

Дополнительную информацию можно найти в документации ..
https://docs.spring.io/spring-cloud-function/docs/3.1.0/reference/html/spring-cloud-function.html#_declarative_function_composition

person 50qbits    schedule 07.01.2021
comment
Как тогда выглядит переплет? - person Ben Keil; 08.01.2021
comment
Для меня это выглядит примерно так: github.com/spring-cloud/spring-cloud-stream-samples/blob/master/ невозможно с Kafka - person Ben Keil; 08.01.2021
comment
это зависит от того, чего вы хотите достичь, я имею в виду, что ваш вариант использования похож на этот конвейер ... поставщик - ›функция (составная) -› kafka. Имея это в виду, вам нужна только конечная точка outboud kafka (привязка вывода), потому что внутренняя маршрутизация обмена сообщениями зависит от каналов весенней интеграции. - person 50qbits; 11.01.2021