java - RxJava : How to conditionally apply Operators to an Observable without breaking the chain

Question

我在 RxJava 可观察对象上有一个运算符链。我希望能够根据 boolean 值应用两个运算符之一，而不会“打断链条”。

我对 Rx(Java) 比较陌生，我觉得可能有一种比我目前引入临时变量的方法更惯用和可读的方法。

这是一个具体的例子，如果批处理大小字段为非空，则从可观察对象中缓冲项目，否则使用 toList() 发出单个无限制大小的批处理:

Observable<Item> source = Observable.from(newItems);
Observable<List<Item>> batchedSource = batchSize == null ?
                source.toList() :
                source.buffer(batchSize);
return batchedSource.flatMap(...).map(...)

这样的事情可能吗？ (伪 lambda，因为 Java):

Observable.from(newItems)
    .applyIf(batchSize == null,
             { o.toList() },
             { o.buffer(batchSize) })
    .flatMap(...).map(...)

Answer 1

您可以使用 compose(Func1) 保持顺序但执行自定义行为

source
.compose(o -> condition ? o.map(v -> v + 1) : o.map(v -> v * v))
.filter(...)
.subscribe(...)