相当于 RxJava 的 RxJS.pausable(Observable)

Question

我为 RxJava 搜索 RxJS.pausable(Observable) 的等价物。

目前我有以下但看起来有点奇怪。有没有更好的方法？

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

至resume/pase：

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

Event observable 可以按照上面的建议用 combineLatest 替换，但它是针对每个用例的，并且仍然可以进行平面映射。可以将相同的 fasle/true 推到暂停器上，这就是 distrinctUntilChanged 启动的地方，以避免在已经 resumed/paused.

时恢复或暂停

相当于 RxJava 的 RxJS.pausable(Observable)

equivalent of RxJS.pausable(Observable) for RxJava

java

reactive-programming

rxjs

rx-java

rx-android