RxJava for each item in a list make api request with Retrofit in Android

Question

I have the following issue in my Android project:

I have a list of elements (ArrayList) obtained from a previous network call using Retrofit. CategoryItem looks like this:

public class CategoryItem {
   String id;
   String name;
   public CategoryItem(String id, String name) {
     this.id = id;
     this.name = name;
   }
   public String getId() {
     return id;
   }
   public String getName() {
     return name;
   }
}

Practically the categories is made of 20 elements. Now I need to make a network API and obtain the list of products for each of the categories. In order to do so, the products API has the id from the category as query parameter. After I obtain the list of products of a certain categoy I am adding it into internal SQLite DB.

I would like to do this with RxJava (either 1 and 2).

What I have done up to now is not multi-threaded and is on the MainThread which is not correct. I will add the code snippet here:

for (int i = 0; i < categoryItems.size(); i++) {
    CategoryItem categoryItem = categoryItems.get(i);
    final String iCat = categoryItem.getId();
    Observable<ProductResponse> call = networkManager.getApiServiceProductsRx().getProductsRx(iCat);

    Subscription subscription = call
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<ProductResponse>() {
                @Override
                public void onCompleted() {
                }

                @Override
                public void onError(Throwable e) {
                    if (e instanceof HttpException) {
                        HttpException response = (HttpException) e;
                        int code = response.code();
                        Log.d("RetrofitTest", "Error code: " + code);
                    }
                }

                @Override
                public void onNext(ProductResponse response) {
                    mProductItemsBelongingToACategory = new ArrayList<>();
                    mProductItemsBelongingToACategory = response.getProductChannel().getProductItems();
                    mDatabaseHandler.addProducts(iCat, mProductItemsBelongingToACategory);
                }
            });

    subscriptions2.add(subscription);

}

I would like to do this with flatMap or flatMapIterable and eventually using multi-threading.

Answer 1

RxJava has an equivalent of for loop:

Observable.from(...) //RxJava1

or

Observable.fromIterable(...) //RxJava2

Applying that would allow you for iterating over CategoryItem list. Then each categoryItem.getId() must be paired with the result from getProductsRx() API call. And then merge all results into single list with toList() This would look like that:

RxJava1:

Observable.from(categoryItems)
    .flatMap(new Func1<CategoryItem, Observable<Pair<String, ProductResponse>>>() {
        @Override
        public Observable<Pair<String, ProductResponse>> call(@NonNull CategoryItem categoryItem) throws Exception {
            return Observable.zip(
                    Observable.just(categoryItem. getId()),
                    networkManager.getApiServiceProductsRx().getProductsRx(categoryItem. getId()),
                    new Func2<String, ProductResponse, Pair<String, ProductResponse>>() {
                        @Override
                        public Pair<String, ProductResponse> call(@NonNull String id, ProductResponse productResponse) throws Exception {
                            return new Pair<String, ProductResponse>(id, productResponse);
                        }
                    });
        }
    })
   .toList()
   .subscribeOn(Schedulers.io())
   .observeOn(AndroidSchedulers.mainThread())

RxJava2:

Observable.fromIterable(categoryItems)
    .flatMap(new Function<CategoryItem, ObservableSource<Pair<String, ProductResponse>>>() {
        @Override
        public ObservableSource<Pair<String, ProductResponse>> apply(@NonNull CategoryItem categoryItem) throws Exception {
            return Observable.zip(
                    Observable.just(categoryItem. getId()),
                    networkManager.getApiServiceProductsRx().getProductsRx(categoryItem. getId()),
                    new BiFunction<String, ProductResponse, Pair<String, ProductResponse>>() {
                        @Override
                        public Pair<String, ProductResponse> apply(@NonNull String id, @NonNull ProductResponse productResponse) throws Exception {
                            return new Pair<String, ProductResponse>(id, productResponse);
                        }
                    });
        }
    })
   .toList()
   .subscribeOn(Schedulers.io())
   .observeOn(AndroidSchedulers.mainThread())

Written in RxJava2.

Above are only Observables. You can still use you Subscriber and save data to database in onNext(). However, you might also use doOnNext() operator after toList() to save data in database.