эквивалент RxJS.pounable (наблюдаемый) для RxJava

Question

Я ищу эквивалент RxJS.pausable(Observable) для RxJava.

В настоящее время у меня есть следующее, но это выглядит немного странно. Есть ли способ лучше?

PublishSubject<Boolean> allValidSubject = PublishSubject.create(); 
PublishSubject<Void> clickSubject = PublishSubject.create(); 

allValidSubject 
     .switchMap(valid -> valid ? clickSubject : Observable.empty()) 
     .skipUntil(allValidSubject) 
// ... 

В конце у меня есть два наблюдаемых. Один для проверки ввода и один для события нажатия кнопки отправки.

Answer 1

Я думаю, что что-то подобное сделало бы это.

Observable<String> event = Observable.just("My Input"); 
// false - paused by default 
final BehaviorSubject<Boolean> pauser = BehaviorSubject.create(false); 

event 
    .flatMap(new Func1<String, Observable<String>>() { 
     @Override 
     public Observable<String> call(final String input) { 
      return pauser 
        .distinctUntilChanged() 
        .filter(new Func1<Boolean, Boolean>() { 
         @Override 
         public Boolean call(Boolean resumed) { 
          return resumed; 
         } 
        }) 
        .map(new Func1<Boolean, String>() { 
         @Override 
         public String call(Boolean resumed) { 
          return input; 
         } 
        }); 
     } 
    }) 
    .subscribe(new Action1<String>() { 
     @Override 
     public void call(String input) { 
      // handle input 
     } 
    }); 

Чтобы возобновить/Pase:

pauser.onNext(true); 
pauser.onNext(false); 

событие наблюдаемым может быть заменен combineLatest, как указано выше, но в случае использования, и по-прежнему может быть flatMapped. Тот же самый fasle/true может быть нажат на паузу, вот где-то недопустимое значениеUntilChanged, чтобы избежать возобновления или паузы, если они уже возобновлены/приостановлены.