RxJava: Jak warunkowo zastosować operatorów do obserwowalnego bez zerwania łańcucha

Question

Mam łańcuch operatorów na RxJava obserwowalne. Chciałbym móc zastosować jednego z dwóch operatorów w zależności od wartości boolowskiej bez "zerwania łańcucha".

Jestem stosunkowo nowy w Rx (Java) i wydaje mi się, że istnieje prawdopodobnie bardziej idiomatyczny i czytelny sposób na to, niż moje obecne podejście do wprowadzania zmiennej tymczasowej.

Oto konkretny przykład, buforowanie obiektów z obserwowalne jeśli pole wielkość partii jest niezerowe, inaczej emitujące pojedynczej partii nieograniczonej wielkości z toList():

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

Czy coś takiego jest możliwe? (Pseudo-lambdas ponieważ Java):

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

Answer 1

Można użyć compose(Func1) pozostać w kolejności, ale zrobić zwyczaj zachowanie

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