java - 如何从另一种方法动态地向 react 器热通量添加元素？

Question

我有一个数据源服务，它以观察者为参数。

void subscribe(Consumer onEventConsumer);

我想使用通量作为 RSocket 的响应流。
我怎样才能做到这一点？
正如我现在看到的，它应该是这样的

Flux<T> controllerMethod(RequestMessage mgs) {
   var flux = Flux.empty();
   dataSource.subscribe(event -> flux.push(event));
   return flux;
}

但是我非常怀疑这是一个合适的解决方案，而且我是响应式(Reactive)方法的新手，我不知道我应该在这里使用什么方法？

Answer 1

正如西蒙已经指出的那样，这就是您使用的 Flux.create为了。
看看Getting Started Guide在 projectreactor.io .
在镜头中，您在 create 的 lambda 内注册了一个自定义监听器。方法:

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

您要做的是将传入的元素传递给 FluxSink ，然后将在 Flux 上发布这些元素。