Agfdhyk

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

public interface CommunicationController {

     Flowable<DataContainer> 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.

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)

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

Observable<AppMetaDataBiz> appMetaDataObservable = this.mAppRepository.getAppMetaData();



Observable<ProductBiz> productDetailsObservable =

        this.mProductRepository.getProductDetails(this.productId);



return productDetailsObservable.zipWith(appMetaDataObservable,

                                            ((jmProduct, jmAppMetaData) -> {



                                               //TODO: implement business logic



                                                return jmProduct;

                                            }));