RXJava - 缓冲 observable 1 直到 observable 2 发出一项答案

【问题标题】：RXJava - buffer observable 1 until observable 2 emits one itemRXJava - 缓冲 observable 1 直到 observable 2 发出一项
【发布时间】：2016-04-25 07:56:02
【问题描述】：

我想要以下行为：

observableMain 应该缓冲所有项目，直到observableResumed 发出一个值。然后observableMain 应该发出所有缓冲和所有特征值...

我在我的activity's onCreate 中做什么：

 PublishSubject<T> subject = ...; // I create a subject to emit items to and to subscribe to

 // 1) I create a main observable from my subject
 final Observable<T> observableMain = subject
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());

 // 2) I use a custom base class in which I can register listeners 
 // for the onResume event and which I can query the isResumed state! 
 // I call the object the pauseResumeProvider!

 // 2.1) I create an observable, it emits a value ONLY if my activity is resumed
 final Observable<Boolean> obsIsResumed = Observable
            .defer(() -> Observable.just(pauseResumeProvider.isActivityResumed()))
            .skipWhile(aBoolean -> aBoolean != true);

 // 2.2) I create a second observable, it emits a value as soon as my activity is resumed
 final Observable<Boolean> obsOnResumed = Observable.create(new Observable.OnSubscribe<Boolean>()
    {
        @Override
        public void call(final Subscriber<? super Boolean> subscriber)
        {
            pauseResumeProvider.addPauseResumeListener(new IPauseResumeListener() {
                @Override
                public void onResume() {
                    pauseResumeProvider.removePauseResumeListener(this);
                    subscriber.onNext(true);
                    subscriber.onCompleted();
                }

                @Override
                public void onPause() {

                }
            });
        }
    });

// 2.3) I combine the resumed observables and only emit the FIRST value I can get
final Observable<Boolean> observableResumed = Observable
        .concat(obsIsResumed, obsOnResumed)
        .first();

// 3) here I'm stuck
// 3 - Variant 1:
Observable<T> observable = observableMain
            .buffer(observableResumed)
            .concatMap(values -> Observable.from(values));
// 3 - Variant 2:
// Observable<T> observable = observableMain.delay(t -> observableResumed);

// 4) I emit events to my my subject...
// this event is LOST!
subject.onNext("Test in onCreate");

问题

activity 恢复后发送到subject 的所有项目都在工作，之前的所有项目都丢失了（至少使用delay 解决方案）。我无法让我想要的行为发挥作用。我该如何正确解决这个问题？

【问题讨论】：

这似乎是不必要的复杂。为什么要求只有在您的活动恢复时才发出值？
我想在我的活动中重用一个可观察对象，并且我想从后台任务更新 UI，所以我想控制何时发出项目（我不希望更新 UI，如果活动未恢复，我也不想丢失更新事件）。我想用 RXJava 解决方案替换我使用了相当长一段时间的现有事件总线 + 队列系统...

标签： android rx-java

【解决方案1】：

重播源并使用 delaySubscription 触发真正的订阅。

PublishSubject<Integer> emitNow = PublishSubject.create();

ConnectableObservable<T> co = source.replay();

co.subscribe();

co.connect();

co.delaySubscription(emitNow).subscribe(...);

emitNow.onNext(1);

编辑：

Here is a gist 使用运算符，您可以将lift 放入可以暂停和恢复上游排放的序列。

【讨论】：

这看起来很有希望。将对其进行更多测试。记住ConnectableObservable 并取消订阅并重新订阅它应该可以暂停/恢复我的主订阅并且永远不会丢失已发出的项目（这就是我最终想要的）
您无法暂停此设置。它将在后台运行，一旦您发出 emitNow 信号，您就会获得所有数据。
我认为，从co 取消订阅并以同样的方式重新订阅它就像暂停一样？
我明白了，这不起作用。任何想法如何实现这一目标？我希望能够重新订阅一个 observable 并且只重新发送尚未发出的项目。我想在活动暂停时暂停我的订阅，并在活动恢复时恢复它并获取同时发出的所有项目
断开连接会完全删除 co.delaySubscription.subscribe 链，而且您将从头开始获得整个序列。 distinct() 可能会有所帮助，具体取决于您的数据类型

【解决方案2】：

另一种方法是延迟可观察源（热）的排放，直到“阀门”打开

source.delay(new Func1<Integer, Observable<Boolean>>() {
    @Override
    public Observable<Boolean> call(Integer integer) {
        return valve.filter(new Func1<Boolean, Boolean>() {
            @Override
            public Boolean call(Boolean isOpen) {
                //emits only when isOpen is true 
                return isOpen;
            }
        });
    }
})
.subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("out: " + integer);
    }
});

要点：Rx Valve

【讨论】：