Cannot use RX Android Observable.zip operator

Question

I am using rx-android zip operator to merge two retrofit calls.

Previously the code was like this:

  affinityService.rewardsStatusChanges()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .compose(this.<RewardsStatus>bindToLifecycle())
            .subscribe(new Action1<RewardsStatus>() {
                @Override
                public void call(RewardsStatus rewardsStatus) {
                    onRewardStatus(rewardsStatus);
                }
            });

    affinityService.affinityStatusChanges()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .compose(this.<AffinityStatus>bindToLifecycle())
            .subscribe(new Action1<AffinityStatus>() {
                @Override
                public void call(AffinityStatus affinityStatus) {
                    onAffinityStatus(affinityStatus);
                }
            });

rewardsStatusChanges() and affinityStatusChanges() are two retrofit calls.

Now I need to merge them.

What I have tried:

affinityService.rewardsStatusChanges()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .compose(this.<RewardsStatus>bindToLifecycle())
                .flatMap(new Func1<RewardsStatus, Observable<RewardsStatus>>() {
                    @Override
                    public Observable<RewardsStatus> call(RewardsStatus rewardsStatus) {
                        return Observable.just(rewardsStatus);
                    }
                })
                .flatMap(new Func1<RewardsStatus, Observable<RewardsStatus>>() {
                    @Override
                    public Observable<RewardsStatus> call(RewardsStatus rewardsStatus) {
                        return Observable.zip(Observable.just(rewardsStatus),
                        affinityService.affinityStatusChanges(),new Func2<RewardsStatus, AffinityStatus, RewardsStatus>() {
                                    @Override
                                    public RewardsStatus call(RewardsStatus rewardsStatus, AffinityStatus affinityStatus) {
                                        onAffinityAndRewardsMerged(rewardsStatus,affinityStatus);
                                        return null;
                                    }

                        });
                    }
                });

But unfortunately the above codebase is not working. Any idea how to do this.

I am using:

RX_ANDROID_VERSION=1.0.1
RX_JAVA_VERSION=1.0.14

Answer 1

posting as you wished, however with that return null in the anonymous function you will get null in your consumer, so I think returning like Pair<RewardsStatus,AffinityStatus> would be nicer, and do that result processing in the consumer.

 Observable.zip(affinityService.rewardsStatusChanges(), affinityService.affinityStatusChanges(),
            object : Func2<RewardsStatus, AffinityStatus, RewardsStatus>() {
                fun call(rewardsStatus: RewardsStatus, affinityStatus: AffinityStatus): RewardsStatus? {
                    onAffinityAndRewardsMerged(rewardsStatus, affinityStatus)
                    return null
                }
            })