spring cloud函数webflux+spring cloud stream是否与http源代码集成

Question

我正在尝试将spring cloud stream与spring cloud函数webflux集成

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

从cloud stream中，我可以看到源代码需要定义为供应商https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/2.1.2.release/single/spring-cloud-stream.html#_spring_cloud_function

但是我的用例是从Active httpendpoint获取POST数据并摄取到kafka，有没有办法从spring cloud function web和spring cloud stream实现它？

@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);
    }
}

如果我运行它，我可以看到date每1秒钟就会被插入到kafka中，如果我调用供应商的getendpoint，比如localhost://8080/date结果生成date响应，那么有没有什么方法可以用spring cloud函数将paylaod从post注入到kafka？

Answer 1

您的问题帮助发现了一个问题，它与由函数和流提供的自动配置之间的生命周期不一致有关。这个问题的表现方式是，由Spring Cloud函数创建的rest点不能看到绑定，因为它是在更早的时候创建的

所以我们很快就会解决这个问题。同时，有一个变通方法，需要您从ApplicationContext访问Output通道（见下文）：

@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());
     };
  }
}