亚洲视频二区_亚洲欧洲日本天天堂在线观看_日韩一区二区在线观看_中文字幕不卡一区

本文介紹了將回調(diào)轉(zhuǎn)換為被動(dòng)發(fā)布者(Flux)的處理方法，對大家解決問題具有一定的參考價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)吧！

問題描述

我正在使用第三方庫注冊MessageListener，當(dāng)某些事件發(fā)生時(shí)，它們調(diào)用已注冊的監(jiān)聽器onMessage方法

public interface MessageListener {  
   // third party code, it auto-scans for all MessageListeners and registers them
    void onMessage(Message message);
}


public class SimpleMessageListener implements MessageListener {
   public void onMessage(Message message) {
      //do something non blocking
      //is it possible to 'transmit' to messagePublisher
}
   public Flux<Message> messagePublisher() {
       // a method to which to subscribeOn    
   }
}

所以我的問題是，將其轉(zhuǎn)化為流量的最佳方法是什么

最后，我希望能夠做這樣的事情

messagePublisher().subscribe(System.out::println);

編輯*
我的第一次嘗試是這樣的

private List<FluxSink<Message>> handlers = new ArrayList<>();
public void onMessage(Message message) {
   handlers.forEach(han -> han.next(message));
}
public Flux<Message> messagePublisher() {
        return Flux.create(sink -> {
            handlers.add(sink);
            sink.onDispose(() -> handlers.remove(sink));
        });
    }

有效-但我覺得這不是一個(gè)很好的解決方案，讓類實(shí)現(xiàn)FlosSink并手動(dòng)處理是不是更好–目前我預(yù)計(jì)不會(huì)有很多訂閱者。
但是有許多MessageListeners(每種類型一個(gè))

推薦答案

您可以創(chuàng)建單個(gè)Flux實(shí)例來橋接MessageListener觀察到的消息，例如

public class SimpleMessageListener implements MessageListener {
   private FluxSink<Message> handler;
   private Flux<Message> flux;

   public SimpleMessageListener() {
      flux = Flux.create(emitter -> {
          handler = emitter;
      }, OverflowStrategy.DROP); // or some other overflow strategy
   }

   public void onMessage(Message message) {
       if (handler != null) {
           /* 
            * null check is required to avoid NPE if a message is received 
            * before any subscription occurs since handler is instantiated
            * lazily when the first subscription is requested
            */
           handler.next(message);
       }
   }

   public Flux<Message> messagePublisher() {
       return flux;
   }
}

現(xiàn)在所有監(jiān)聽器都可以使用Flux‘publish()方法及其返回的ConnectableFlux訂閱相同的messsagePublisher()Flux實(shí)例：

// fetch message publisher
Flux<Message> messagePublisher = messageListener.messagePublisher();

// prepare ConenctableFlux
ConnectableFlux<Message> connectableFlux = messagePublisher().publish();

// register subscribers
connectableFlux.subscribe(/* aConsumer */);
connectableFlux.subscribe(/* aCoreSubscriber */);
connectableFlux.subscribe(/* aSubscriber */);

// connect the ConnectableFlux to messagePublisher
connectableFlux.connect();

這篇關(guān)于將回調(diào)轉(zhuǎn)換為被動(dòng)發(fā)布者(Flux)的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，