RxJava：如何有条件地将运算符应用于Observable而不中断链

Question

我在RxJava上有一连串的运算符。我希望能够根据布尔值应用两个运算符之一，而不会“破坏链”。

我是Rx（Java）的新手，我觉得这样做可能比我目前引入临时变量的方法更具惯用性和可读性。

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

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

这样的事情可能吗？（伪lambdas，因为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(...)