Пауза и возобновление наблюдаемого на основе логических ворот в 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); 
        } 
       }); 

Кто-нибудь знает, как достичь чего-то вроде этого?

Edit я написал в блоге о том, как достичь этого https://medium.com/@scottalancooper/pausing-and-resuming-a-stream-in-rxjava-988a0977b771#.gj7fsi1xk

Answer 1

В библиотеке RxJava2Extensions теперь есть оператор valve(), который выполняет запрошенное поведение.

Answer 2

Просто используйте готовый Observable.window оператор, который принимает один параметр Observable<T>.