RxJava: Как условно применять операторы к наблюдаемому, не нарушая цепочку

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

ли что-то подобное возможно? (Псевдо-лямбды, так как 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(...)