RxJava：如何在不破坏链条件的情况下有条件地将运算符应用于Observable

Question

我在RxJava observable上有一个运算符链。我希望能够根据布尔值应用两个运算符中的一个，而不会“断链”。

我对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(...)