RxJava: Complete stream based on condition

Question

I have on Observable, which emits items when some data is coming from a BLE connection:

public interface CommunicationController {
     Flowable dataReceived();
}

On top of this I want to build a Observable, which completes when one of the following conditions is true:

a. I receive two messages of a specific type (this is done by using filter operator on the received DataContainer item.

communicationController.dataReceived()
    .filter(data -> isTypeA(data) || isTypeB(data))
    .take(2)
    .toList()
    .map(dataContainers -> doSomeMappingToCommon object)

b. I receive one message of a specific type (again using filter operator).

communicationController.dataReceived()
    .filter(data -> isTypeC(data))
    .firstOrError()
    .map(dataContainers -> doSomeMappingToCommon object);

How can I combine those two Observables into a single one? Additionally only one of the two Observables will emit an item.

user_5923973 · Answer

I think your answer will be Observable.merge: http://reactivex.io/documentation/operators/merge.html

According to the documentation, this does not wait for every observable to emit something.

One option is to use merge and takeUntil: http://reactivex.io/documentation/operators/takeuntil.html

Kotlin code:

val stopStream = Observable.merge(...)

streamThatWillRun = streamThatWillRun.takeUntil(stopStream)

user_10611340 · Answer

To combine 2 observables, you can use operator zipWith. Example:

Observable appMetaDataObservable = this.mAppRepository.getAppMetaData();

Observable productDetailsObservable =
        this.mProductRepository.getProductDetails(this.productId);

return productDetailsObservable.zipWith(appMetaDataObservable,
                                            ((jmProduct, jmAppMetaData) -> {

                                               //TODO: implement business logic

                                                return jmProduct;
                                            }));

RxJava: Complete stream based on condition

2 Answers