小编alo*_*rey的帖子

我有一个使用Boot 2.0和webflux的应用程序,并且有一个端点返回Flux of ServerSentEvent.通过利用spring-amqp从RabbitMQ队列中消耗消息来创建事件.我的问题是:如何最好地将已MessageListener配置的侦听器方法桥接到可以传递给控制器​​的Flux？

Project Reactor的create部分提到"将现有的API与反应世界联系起来非常有用 - 例如基于侦听器的异步API",但我不确定如何直接挂钩消息监听器,因为它包含在DirectMessageListenerContainer和MessageListenerAdapter.他们的例子来自创建部分:

Flux<String> bridge = Flux.create(sink -> {
    myEventProcessor.register( 
      new MyEventListener<String>() { 

        public void onDataChunk(List<String> chunk) {
          for(String s : chunk) {
            sink.next(s); 
          }
        }

        public void processComplete() {
            sink.complete(); 
        }
    });
});

到目前为止,我最好的选择是创建一个Processor并且onNext()每次在RabbitMQ监听器方法中调用以手动生成事件.