Zatrzymaj i wznów obserwowalne na podstawie bramki logicznej w RxJava 2.X?

Question

Powiedzmy, że mam procesor, który emituje wartość boolowską, gdy naciśnięty jest dowolny przycisk, potraktuj to jako przełączenie.

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

Co chciałbym zrobić, to użyć wartości bramy, aby wstrzymać i wznowić zaobserwowania sekwencję, buforowanie wartości emitowanych podczas pauzy.

Wczytałem się w to dużo i chociaż wydaje się to możliwe w reaktywnych rozszerzeniach dla innych języków, RxJava wydaje się nie obsługiwać tego.

Oto przykład tego, co chciałbym osiągnąć, po prostu generuje przyrostową wartość w każdej sekundzie. Po naciśnięciu przycisku chcę wyjście zatrzymać aż naciśnij go ponownie, który powinien wypisać każdy element emitowane pomiędzy dwoma przyciskami pras:

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

Czy ktoś zna sposób, aby osiągnąć coś takiego?

Edit pisałem na blogu o tym, jak osiągnąć ten https://medium.com/@scottalancooper/pausing-and-resuming-a-stream-in-rxjava-988a0977b771#.gj7fsi1xk

Answer 1

Istnieje teraz operator valve() w bibliotece RxJava2Extensions, który wykonuje żądane zachowanie.

Answer 2

Wystarczy użyć gotowych Observable.window operatora, który przyjmuje jedną Observable<T> parametr.