基於RxJava 2.X中的布爾門暫停和恢復觀察值？

Question

假設我有一個處理器，當按下按鈕時會發出一個布爾值，可以認爲這是一個切換。

boolean gateValue = true; 
    PublishProcessor<Boolean> gate = PublishProcessor.create(); 
    view.onButtonClicked() 
      .subscribe(new Action1<Void>() { 
       @Override 
       public void call(final Void aVoid) { 
        gate.onNext(gateValue = !gateValue); 
       } 
      })); 

我想這樣做是使用門的價值，暫停和恢復可觀察序列，緩衝發射值，而暫停。

我已經閱讀了很多，雖然看起來可能在其他語言的反應式擴展中，但RxJava似乎不支持它。

下面是我想達到的一個例子，它只是每秒輸出一個增量值。當我按下按鈕我所要的輸出停止，直到我再次按下它應該輸出兩次按鍵之間發出的每一個項目：

Flowable.interval(1, TimeUnit.SECONDS) 
       .bufferWhile(gate) 
       .flatMapIterable(longs -> longs) 
       .subscribe(new Consumer<Long>() { 
        @Override 
        public void accept(final Long aLong) throws Exception { 
         view.displayTime(aLong); 
        } 
       }); 

有誰知道的方式來實現這樣的事情？

編輯我已經寫了關於如何實現這一目標https://medium.com/@scottalancooper/pausing-and-resuming-a-stream-in-rxjava-988a0977b771#.gj7fsi1xk

Answer 1

RxJava2Extensions庫中現在有一個運算符valve()來執行請求的行爲。

Answer 2

只需使用現成的Observable.window運營商，它有一個Observable<T>參數博客文章。