Есть ли интеграция функции весеннего облака webflux + поток весеннего облака с http источником

Question

Я пытаюсь интегрировать весенний облачный поток с функцией весеннего облака. Webflux

поскольку в будущих выпусках они устарели в потоках реагирования на весенние облака, я пытаюсь использовать функции весенних облаков https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/2.1.2.RELEASE/single/spring-cloud-stream.html#spring-cloud-stream-preface-notable-deprecations

Веб-функция Spring Cloud может предоставить конечную точку своей функции путями, как в документе

https://cloud.spring.io/spring-cloud-static/spring-cloud-function/1.0.0.RELEASE/single/spring-cloud-function.html

из облачного потока. Я вижу, что источник должен быть определен как поставщик https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/2.1.2.RELEASE/single/spring-cloud-stream.html#_spring_cloud_function

но мой вариант использования - получить данные POST из реактивной конечной точки http и ввести их в kafka, есть ли способ добиться этого с помощью веб-функции весеннего облака и весеннего потока облака?

из документа для функции весеннего облака с потоком весеннего облака

@SpringBootApplication
@EnableBinding(Source.class)
public static class SourceFromSupplier {
    public static void main(String[] args) {
        SpringApplication.run(SourceFromSupplier.class, "--spring.cloud.stream.function.definition=date");
    }
    @Bean
    public Supplier<Date> date() {
        return () -> new Date(12345L);
    }
}

и если я запускаю это, я вижу, что дата вставляется в kafka каждую 1 секунду, и если я вызываю конечную точку get для поставщика, например localhost: / 8080 / date приводит к ответу на дату, есть ли способ ввести paylaod? от поста до кафки с функцией весеннего облака?

Answer 1

Существует проблема, которую ваш вопрос помог обнаружить, и он связан с несоответствием жизненного цикла между автоконфигурациями, предоставляемыми функцией и потоком. Эта проблема проявляется в том, что точка отдыха, созданная Spring Cloud Functions, не может видеть привязки, так как она создана намного раньше

Итак, мы скоро решим проблему. Между тем существует обходной путь, который потребует от вас доступа к каналу output из ApplicationContext (см. Ниже):

@SpringBootApplication
@EnableBinding(Source.class)
public class SimpleFunctionRabbitDemoApplication {

  public static void main(String[] args) throws Exception {      
    SpringApplication.run(SimpleFunctionRabbitDemoApplication.class);
  }

  @Bean
  public Consumer<String> storeSync(ApplicationContext context) {
     return v -> {
        MessageChannel channel = context.getBean(Source.OUTPUT, MessageChannel.class);
        channel.send(MessageBuilder.withPayload(v).build());
     };
  }
}