将 API 回调绑定到 RxJava Observable

Question

我正在尝试制作一个响应式应用程序，该应用程序在单独的线程上侦听网络套接字的价格，并且对如何构建 Observable 感到有些困惑。我拥有的许多接口都受到我正在使用的 API 的限制，因此无法更改。我提炼了我想要做的测试作为下面的测试，但我看不到如何填写 getPriceReactive() 方法的主体，以便订阅者在控制台上打印价格（请参阅代码中的注释）。

public class PriceObservableTest {

   // This interface is defined externally and used by the API
   private interface ITickHandler {
       void priceReceived(double price);
   }

   // Stores the price (currently just one double for illustration)
   private class Tick {
       double price = Double.NaN;
   }

   // Implementation of handler called by API when it receives a price
   private class TickHandler implements ITickHandler {
       private final Tick tick;

       TickHandler() { this.tick = new Tick(); }

       @Override public void priceReceived(double x) { tick.price = x; }
   }

   // This class emulates the API delivering prices from the socket
   private class PriceSource {
      private final Thread thread;

      PriceSource(final ITickHandler handler) {
          thread = new Thread(new Runnable() {
              final Random r = new Random();
              @Override public void run() {
                  while (!Thread.currentThread().isInterrupted()) {
                      try {
                          Thread.sleep(100);
                          handler.priceReceived(r.nextDouble() * 100);
                      } catch (InterruptedException e) {
                          break;
                      }
                  }
                  System.out.println("Price thread closed");
              }
         });
      }

      void subscribe() { thread.start(); }

      void unsubscribe() { thread.interrupt(); }
  }

  @Test
  public void simpleTest() throws Exception {

      final ITickHandler handler = new TickHandler();

      // Simulate some prices received periodically from a socket
      PriceSource prices = new PriceSource(handler);

      Observable<Tick> reactive = getPriceReactive(handler);

      reactive.subscribe(new Subscriber<Tick>() {
          @Override public void onCompleted() { }
          @Override public void onError(Throwable e) { }
          @Override public void onNext(Tick tick) {
              System.out.println("Received price: " + tick.price);
          }});

      // Observe prices for 1 second. The subscriber should print them to console
      prices.subscribe();
      Thread.sleep(1000); 
      prices.unsubscribe();
   }

   // Returns an observable that reacts to price changes
   private Observable<Tick> getPriceReactive(ITickHandler handler) {
       return Observable.create(new Observable.OnSubscribe<Tick>() {
           @Override public void call(Subscriber<? super Tick> subscriber) {

              // How to call subscriber.onNext() whenever
              // priceReceived() is called with a new price?

           }
       });
   }
}

每当 API 调用 priceReceived() 时，就需要以某种方式调用 subscriber.onNext()，但我不太明白如何实现这一点。当然，我可以在TickHandler 中存储对订阅者的引用，但这违背了拥有Observable 的目的，不是吗？

Answer 1

在ITickHandler 实现中转换为Observable。您控制的不是订阅者，而是发布者

private class TickHandler implements ITickHandler {
   private final Tick tick;
   private final PublishSubject<Tick> priceSubject;

  TickHandler() { 
       this.tick = new Tick(); 
       this.priceSubject = PublishSubject.create();
   }

   @Override public void priceReceived(double x)
   { 
        tick.price = x; 
        priceSubject.onNext(tick);
   }

   public Observable<Tick> priceReceivedObservable()
   {
       return priceSubject.asObservable();   
   }
}

你可以在你的测试中使用它，比如：

final ITickHandler handler = new TickHandler();
PriceSource prices = new PriceSource(handler);

handler.priceReceivedObservable()
       .subscribe(new Subscriber<Tick>() {
          @Override public void onCompleted() { }
          @Override public void onError(Throwable e) { }
          @Override public void onNext(Tick tick) {
              System.out.println("Received price: " + tick.price);
          }});

我警告你，它没有经过测试，因为我不会做很多 Java :)