RXJava - 缓冲区可观察1直到可观察2发出一个项目

Question

我想要以下行为:

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

问题

恢复subject之后发送到的所有项目都在activity工作,之前的所有项目都将丢失(至少在delay解决方案中).我无法达到理想的行为.我该如何正确解决这个问题？

Answer 1

重播源并使用delaySubscription触发真实订阅.

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

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

co.subscribe();

co.connect();

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

emitNow.onNext(1);

编辑:

这里有一个操作员的要点,你可以lift进入一个可以暂停和恢复上游排放的序列.