RXJava - 制作一个可暂停的可观察对象（例如带有缓冲区和窗口）

我想创建执行以下操作的可观察对象：

• 在暂停时缓冲所有项目
• 立即发出项目，同时它们不会暂停
• 暂停/恢复触发器必须来自另一个可观察的
• 必须保存它才能供不在主线程上运行的可观察对象使用，并且必须保存更改主线程的暂停/恢复状态

我想用一个BehaviorSubject<Boolean>作为触发器并将该触发器绑定到活动的onResume and onPause事件。 （附代码示例）

Question

我已经设置了一些东西，但它没有按预期工作。我像下面这样使用它：

Observable o = ...;
// Variant 1
o = o.lift(new RxValve(getPauser(), 1000, getPauser().getValue())
// Variant 2
// o = o.compose(RXPauser.applyPauser(getPauser()));
o
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe();

目前的问题是，变体 1 应该可以正常工作，但有时，事件只是没有发出 - 阀门没有发出，直到阀门一切正常（可能是线程问题......）！解决方案2更简单并且似乎可行，但我不确定它是否真的更好，我不这么认为。我实际上不确定，为什么解决方案一有时会失败，所以我不确定解决方案2是否解决了（目前对我来说未知）问题......

有人可以告诉我可能是什么问题或者简单的解决方案是否应该可靠地工作？或者告诉我一个可靠的解决方案？

Code

RxValue

https://gist.github.com/akarnokd/1c54e5a4f64f9b1e46bdcf62b4222f08 https://gist.github.com/akarnokd/1c54e5a4f64f9b1e46bdcf62b4222f08

RXPauser 函数

public static <T> Observable.Transformer<T, T> applyPauser(Observable<Boolean> pauser)
{
    return observable -> pauser(observable, pauser);
}

private static <T> Observable<T> pauser(Observable<T> source, Observable<Boolean> pauser)
{
    // this observable buffers all items that are emitted while emission is paused
    Observable<T> sharedSource = source.publish().refCount();
    Observable<T> queue = sharedSource
            .buffer(pauser.distinctUntilChanged().filter(isResumed -> !isResumed), aBoolean -> pauser.distinctUntilChanged().filter(isResumed -> isResumed))
            .flatMap(l -> Observable.from(l))
            .doOnNext(t -> L.d(RXPauser.class, "Pauser QUEUED: " + t));

    // this observable emits all items that are emitted while emission is not paused
    Observable<T> window = sharedSource.window(pauser.distinctUntilChanged().filter(isResumed -> isResumed), aBoolean ->  pauser.distinctUntilChanged().filter(isResumed -> !isResumed))
            .switchMap(tObservable -> tObservable)
            .doOnNext(t -> L.d(RXPauser.class, "Pauser NOT QUEUED: " + t));

    // combine both observables
    return queue.mergeWith(window)
            .doOnNext(t -> L.d(RXPauser.class, "Pauser DELIVERED: " + t));
}

Activity

public class BaseActivity extends AppCompatActivity {

    private final BehaviorSubject<Boolean> pauser = BehaviorSubject.create(false);

    public BaseActivity(Bundle savedInstanceState)
    {
        super(args);
        final Class<?> clazz = this.getClass();
        pauser
                .doOnUnsubscribe(() -> {
                    L.d(clazz, "Pauser unsubscribed!");
                })
                .subscribe(aBoolean -> {
                    L.d(clazz, "Pauser - " + (aBoolean ? "RESUMED" : "PAUSED"));
                });
    }

    public PublishSubject<Boolean> getPauser()
    {
        return pauser;
    }

    @Override
    protected void onResume()
    {
        super.onResume();
        pauser.onNext(true);
    }

    @Override
    protected void onPause()
    {
        pauser.onNext(false);
        super.onPause();
    }
}

你实际上可以使用.buffer()运算符将其传递为可观察的，定义何时停止缓冲，来自书中的示例：

Observable.interval(100, TimeUnit.MILLISECONDS).take(10)
    .buffer(Observable.interval(250, TimeUnit.MILLISECONDS))
    .subscribe(System.out::println);

来自第 5 章“驯服序列”：https://github.com/Froussios/Intro-To-RxJava/blob/master/Part%203%20-%20Taming%20the%20 序列/5.%20 时移%20sequences.md https://github.com/Froussios/Intro-To-RxJava/blob/master/Part%203%20-%20Taming%20the%20sequence/5.%20Time-shifted%20sequences.md

您可以使用PublishSubject as Observable在您的自定义运算符中为其提供元素。每次需要开始缓冲时，通过以下方式创建实例Observable.defer(() -> createBufferingValve())