RxJava Merge Flowable and Completable

user_3168190 0
Viewed 1866

I have a Flowable that continuously emits items and never calls onError or onComplete. Now I have a Completable that I would like to merge with this Flowable so that when the Completable completes the Flowable calls onComplete. I cannot directly change the Flowable object that is given to me.

One issue I am having is that I would have used takeUntil on the Flowable however the Flowable may stop emitting items at any point and I would still like the Completable to be able to call onComplete.

Update: Since we can do Completable.toFlowable() we are able to merge two Flowables. The issue is that I still cannot find a way to complete both when one completes.

rx-java2 java
2 Answers

As you pointed out, you cannot change the original Flowable itself, so it will not emit onComplete. You can have however make the resulting Flowable emit it by doing the following (pseudocode):

val f: Flowable = ...
val c: Completable = ...
val r: Flowable = f.materialize().mergeWith(c.materialize()).dematerialize()
user_8565880

Here is a generalized solution. As noted in @Maxim's answer merge/mergeWith does not accomplish my functionality because it requires both the Flowable and the Completable to finish. This solution also handles disposing of both properly.

    Flowable<Integer> f = Flowable.fromArray(1, 2, 3);
    Completable c = Completable.complete();

    final PublishSubject<Integer> subject = PublishSubject.create();
    final CompositeDisposable cd = new CompositeDisposable();
    Flowable result = subject
            .doOnSubscribe(__ -> {
                cd.add(f.subscribe(subject::onNext));
                cd.add(c.subscribe(subject::onComplete));
            })
            .doOnDispose(cd::dispose)
            .toFlowable(BackpressureStrategy.LATEST);
user_31681900
Related