RxJava фильтра «еще»

Question

Я хочу разделить мое наблюдаемые как if/else заявления

Что-то вроде:

A[] array = new A[10]; 
for (int i = 0; i < array.length; i++) { 
    array[i] = new A(getRandomString(), getRandomInt()); 
} 

Observable.from(array) 
    .filter(a -> condition(a)) 
     .// <--- do stuff if condition returns true 
     .// <- back to parent 
    .filter(a -> complexCondition(a)) // filter all elements(!) 
     .// <--- do stuff if complex condition returns true 
     .// <- back to iterate all elements 

Возможно ли это?

Answer 1

В Observable не работает таким образом, вы должны думать, что наблюдаемое подобно потоку, в котором вы можете получить некоторые предметы, другие. Самое близкое, если/else, которое вы найдете в наблюдаемых, будет использовать GroupsBy.

Здесь есть Розыгрыши пример, который я объясняющую, как работает

https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/transforming/ObservableGroupBy.java

Answer 2

Одним из способов достижения такого поведения является, чтобы подписаться на ваши данные дважды и отфильтровать его по-разному:

Subscription subscriptionA = Observable.from(array) 
    .filter(a -> condition(a)) 
    .subscribe(...) // <-- do stuff for condition A 

Subscription subscriptionB = Observable.from(array) 
    .filter(a -> complexCondition(a)) 
    .subscribe(...) // <-- do stuff for condition B 

Answer 3

Для расширения на Ken's answer другой вариант заключается в «разветвлении» наблюдаемого в двух ветвях, используя replay operator. Это гарантирует, что исходный наблюдаемый вызывается только один раз. Это полезно, если есть некоторые дорогостоящие эффекты обработки или где-то в стороне цепи:

ConnectableObservable<A> connectable = Observable.fromArray() 
    .replay(); 

connectable 
    .filter(a -> condition(a)) 
    .// <--- do stuff if condition returns true 

connectable.connect(); // do this after the first branch 

connectable 
    .filter(a -> complexCondition(a)) // filter all elements(!) 
    .// <--- do stuff if complex condition returns true 

connectable 
    .// <- iterate all elements 

Помните, что все ветви должны обращаться как onNext и OnError события.