我可以跟蹤RxJava訂戶中事件的消耗情況嗎？

Question

我想跟蹤訂閱者開始消費事件和完成時的情況。 是否有任何通用的方式，將適用於所有Observable/Subscribers？

Answer 1

是1.x還是2.x？對於2.x而言，它會變得相當複雜，因爲所有內部協議都必須考慮到不會意外地降低流量。

否則，它可以像寫那纔是真正Observer和運營商之間勻一個Observer簡單：

import io.reactivex.Observer; 

RxJavaPlugins.setOnObservableSubscribe((observable, observer) -> { 
    if (!observable.getClass().getName().toLowerCase().contains("map")) { 
     return observer; 
    } 

    System.out.println("Started"); 

    class SignalTracker implements Observer<Object>, Disposable { 
     Disposable upstream; 
     @Override public void onSubscribe(Disposable d) { 
      upstream = d; 
      // write the code here that has to react to establishing the subscription 
      observer.onSubscribe(this); 
     } 
     @Override public void onNext(Object o) { 
      // handle onNext before or aftern notifying the downstream 
      observer.onNext(o); 
     } 
     @Override public void onError(Throwable t) { 
      // handle onError 
      observer.onError(t); 
     } 
     @Override public void onComplete() { 
      // handle onComplete 
      System.out.println("Completed"); 
      observer.onComplete(); 
     } 
     @Override public void dispose() { 
      // handle dispose 
      upstream.dispose(); 
     } 
     @Override public boolean isDisposed() { 
      return upstream.isDisposed(); 
     } 
    } 
    return new SignalTracker(); 
    }); 

    Observable<Integer> observable = Observable.range(1, 5) 
     .subscribeOn(Schedulers.io()) 
     .observeOn(Schedulers.computation()) 
     .map(integer -> { 
     try { 
      TimeUnit.SECONDS.sleep(1); 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
     } 
     return integer * 3; 
     }); 

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

    Thread.sleep(6000L); 

打印：

Started 
3 
6 
9 
12 
15 
Completed 

編輯：的RxJava 1版需要更多lambdas但可行：

RxJavaHooks.setOnObservableStart((observable, onSubscribe) -> { 
    if (!onSubscribe.getClass().getName().toLowerCase().contains("map")) { 
     return onSubscribe; 
    } 

    System.out.println("Started"); 

    return (Observable.OnSubscribe<Object>)observer -> { 
     class SignalTracker extends Subscriber<Object> { 
      @Override public void onNext(Object o) { 
       // handle onNext before or aftern notifying the downstream 
       observer.onNext(o); 
      } 
      @Override public void onError(Throwable t) { 
       // handle onError 
       observer.onError(t); 
      } 
      @Override public void onCompleted() { 
       // handle onComplete 
       System.out.println("Completed"); 
       observer.onCompleted(); 
      } 
      @Override public void setProducer(Producer p) { 
       observer.setProducer(p); 
      } 
     } 
     SignalTracker t = new SignalTracker() 
     observer.add(t); 
     onSubscribe.call(t); 
    }; 
    }); 

    Observable<Integer> observable = Observable.range(1, 5) 
     .subscribeOn(Schedulers.io()) 
     .observeOn(Schedulers.computation()) 
     .map(integer -> { 
     try { 
      TimeUnit.SECONDS.sleep(1); 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
     } 
     return integer * 3; 
     }); 

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

    Thread.sleep(6000L);