CODEWITHSUNDEEP
javaobservablerx-java2
Rohit
Rohit

Reputation: 2681

Cant Merge RXJava2

Hi I am new to RXJava.

I am trying to get Users from multiple Observable which contains an arraylist. I want to merge both the data and publish it to Observer.

Here is my snippet.

    Observable<Users> baseDataObservable = mService.getUsers(Constants.SITE);

    Items items = new Items();
    items.setName("User Defined...");
    items.setImageURL("http://www.proto.gr/sites/www.proto.gr/files/styles/colorbox/public/images/fruits/cherry.png");
    ArrayList<Items> items1 = new ArrayList<>();
    items1.add(items);

    Users users = new Users();
    users.setItems(items1);
    Observable<Users> usersObservable = Observable.just(users);

    Observable.merge(baseDataObservable, usersObservable)
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(observer);

In above code, 1st observable brings data from API and 2nd one puts user defined data.

But when I use merge on observer I have only single data in ArrayList<Items> which is set defined data, the data which comes initially from api shows data provided by 2nd observable.

Can someone please help me

Upvotes: 1

Views: 84

Answers (1)

Sarath Kn
Sarath Kn

Reputation: 2725

If you are trying to "merge" the array lists you are getting from both the sources, and you need to combine those into a single list, you need to use zip operator instead of merge. The merge operator will emit items immediately when any of the observables triggers, where zip will collect result from all the input observables and proceed after that.

In your approach, the observer will trigger twice with 2 results. first the user-defined items, then the result from API (with a delay since its an API call)

Try like this to get a combined result

        Observable.zip(baseDataObservable, usersObservable, new BiFunction<Users, Users, ArrayList<Items>>() {
        @Override
        public ArrayList<Items> apply(Users usersFromApi, Users usersUserDefined) throws Exception {
            ArrayList<Items> itemsArrayList = new ArrayList<>();
            itemsArrayList.addAll(usersFromApi.getItems());
            itemsArrayList.addAll(usersUserDefined.getItems());
            return itemsArrayList;
        }
    }).subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new DisposableObserver<ArrayList<Items>>() {
                @Override
                public void onNext(ArrayList<Items> items) {
                    // here is the combined Arraylist
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onComplete() {

                }
            });

Upvotes: 1

Related Questions