RxJava: Come applicare condizionatamente operatori a un osservabile senza rompere la catena

Question

Ho una catena di operatori su un osservabile RxJava. Mi piacerebbe essere in grado di applicare uno dei due operatori in base a un valore booleano senza "spezzare la catena".

Sono relativamente nuovo a Rx (Java) e mi sembra che ci sia probabilmente un modo più idiomatico e leggibile di farlo rispetto al mio attuale approccio di introduzione di una variabile temporanea.

Ecco un esempio concreto, il buffering articoli di un osservabile se un campo dimensione del lotto non è nullo, altrimenti emette un singolo lotto di dimensioni illimitate con il toList():

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

è qualcosa di simile possibile? (Pseudo-lambda perché Java):

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

Answer 1

È possibile utilizzare compose(Func1) di rimanere in sequenza, ma fare un comportamento personalizzato

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