RxJava flatMapIterable with concatMap

Question

I have 2 retrofit calls I need to make A & B:

(A): returns an ArrayList

(B): gets the result of (A) which is an ArrayList. (B) iterates through the ArrayList and makes a retrofit call using each and combines the resulting data into a final ArrayList that is sent to my subscriber onNext()

I was able to get it working with flatmap, but the solution is not very elegant. As I understand it, a better solution would be to use flatMapIterable with concatMap but I can't seem to adopt what I have working to flatMapIterable with concatMap.

ServiceFactory.createRetrofitService().getUserFollowing(sessionMgr.getAuthToken())
                .flatMap(new Func1<UserSelfFollows, Observable<? extends ArrayList<Media.MediaData>>>() {
                    @Override
                    public Observable<? extends ArrayList<Media.MediaData>> call(UserSelfFollows userSelfFollows) {

                        //make second requests based on response from First request to get all Users
                        ArrayList<Media.MediaData> arAllMedia = new ArrayList<>();
                        for(UserSelfFollows.UserDataFollows user : userSelfFollows.userdata){

                            Response <ResponseBody> response ;
                            Call <ResponseBody> call;
                            try {
                                call = ServiceFactory.createRetrofitService().getMediaOfUser(user.id,sessionMgr.getAuthToken());
                                response =  call.execute();
                            }catch(IOException ex){
                                return Observable.error(ex);
                            }

                            if (response.isSuccessful()) {

                                try {
                                    String str = responseHelper.streamToString( response.body().byteStream());
                                    Gson gson = new GsonBuilder().create();
                                    Media media = gson.fromJson(str, Media.class);

                                    arAllMedia.addAll(media.mediaData);

                                } catch (IOException e) {
                                    return Observable.error(e);
                                }
                            } else {
                                return Observable.error( new Exception(  responseHelper.getErrorString( response.errorBody().byteStream())) );
                            }
                        }
                        return Observable.just(arAllMedia);

                    }
                })
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<ArrayList<Media.MediaData>>() {
                    @Override
                    public final void onCompleted() {

                    }

                    @Override
                    public final void onError(Throwable e) {

                    }

                    @Override
                    public final void onNext(ArrayList<Media.MediaData> arMedia) {

                    }
                })

Here is what I have so far, but it won't compile:

ServiceFactory.createRetrofitService().getUserFollowing(sessionMgr.getAuthToken())
            .flatMapIterable(new Func1<UserSelfFollows, Iterable<?>>() {
                @Override
                public Iterable<?> call(UserSelfFollows userSelfFollows) {
                    return userSelfFollows.userdata;
                }
            })
            .<Media.MediaData>flatMap(new Func1<UserSelfFollows.UserDataFollows, Observable<Media.MediaData>>() {
                @Override
                public Observable<Media.MediaData> call(UserSelfFollows.UserDataFollows user) {
                    return ServiceFactory.createRetrofitService().getMediaOfUser(user.id, sessionMgr.getAuthToken());
                }
            })
            .toList();

The error is:

yosriz · Answer 1 · 2017-02-16T15:56:43.117

If I understand correctly your scenario you can use flatMapIterable, and then flatMap operators, at the end collect all the retrofit calls result with toList.
In the first flatMapIterable you're flatting the UserDataFollows List you get from the first call (getUserFollowing()) to an Obsevrable that emits multiple items from this List, then flatMap create an Observable that makes the retrofit call for each UserSelfFollows.UserDataFollows data object, (it will happens in parallel which seems more suits here, but you can also use concatMap if you interested in sequential execution), then to collect all final data together, as a list of MediaData objects, you can use the toList

ServiceFactory.createRetrofitService().getUserFollowing(sessionMgr.getAuthToken())
            .flatMapIterable(new Func1<UserSelfFollows, Iterable<?>>() {
                @Override
                public Iterable<?> call(UserSelfFollows userSelfFollows) {
                    return userSelfFollows.userdata;
                }
            })
            .flatMap(new Func1<UserSelfFollows, Observable<Media>>() {
                @Override
                public Observable<Media> call(UserSelfFollows user) {
                    return ServiceFactory.createRetrofitService().getMediaOfUser(user.id, sessionMgr.getAuthToken());
                }
            })
            .toList()
            .subscribe(new Action1<List<Media>>() {
                @Override
                public void call(List<Media> rs) {
                    //do something with the list of media data
                }
            });

This line does not complile: return ServiceFactory.createRetrofitService().getMediaOfUser(user.id, sessionMgr.getAuthToken()); there is no attribute 'id' of 'user'.....I have updated my question with your code and the error being thrown. — Mike6679, Feb 16 '17 at 13:47
Your code is basically throwing the same error I was getting with my code — Mike6679, Feb 16 '17 at 13:53
well, it is rather difficult to solve compile issue like this from the snippet you provided, I'm not sure what are the exact types that your methods are returned, I did simulation based on your flow of the code, and the types that each method seem to return, anyhow, if you have difficulties you can let the autocomplete do it's job with the flatMap, and it will suggest you the right types it is expected. — yosriz, Feb 16 '17 at 14:22
I've updated the answer to reflect the correct object of Media instead of MediaData, according to your code — yosriz, Feb 16 '17 at 16:02
The main reason it did not compile is because of missing return type: Iterable> but I did upvote your answer because it did help. — Mike6679, Feb 16 '17 at 18:12

score 1 · Answer 2 · answered Feb 16 '17 at 08:56

I never used both flatMapIterable or concatMap, but as alternative solution you could use the flatMap and compose operator:

ServiceFactory.createRetrofitService().getUserFollowing(sessionMgr.getAuthToken())
   .flatMap(list -> Observable.from(list))
   .compose(getComposer())
   .subscribe(mediaData -> mMediaDataList.add(mediaData), throwable -> {}, () -> { // on complete do something with mMediaList});

Where getComposer() returns a Transformer UserSelfFollows -> Media.MediaData

protected Observable.Transformer<UserSelfFollows, Media.MediaData> getComposer() {
    return  ;
}

without lamba is the same. Use new inside the operator. AndroidStudio will suggest what to implement — Blackbelt, Feb 16 '17 at 13:44

score 0 · Answer 3 · answered Feb 16 '17 at 15:06

Yosriz's answer put me in the right direction. So to avoid others pulling there hair out with this, here is the FULL code solution:

ServiceFactory.createRetrofitService().getUserFollowing(sessionMgr.getAuthToken())
                .flatMapIterable(new Func1<UserSelfFollows, Iterable<UserSelfFollows.UserDataFollows>>() {
                    @Override
                    public Iterable< UserSelfFollows.UserDataFollows > call(UserSelfFollows userSelfFollows) {
                        return userSelfFollows.userdata;
                    }
                })
                .flatMap(new Func1<UserSelfFollows.UserDataFollows, Observable<Media>>() {
                    @Override
                    public Observable<Media> call(UserSelfFollows.UserDataFollows user) {
                        return ServiceFactory.createRetrofitService().getMediaOfUser(user.id, sessionMgr.getAuthToken());
                    }
                })
                .toList()
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<List<Media>>() {
                    @Override
                    public final void onCompleted() {

                    }

                    @Override
                    public final void onError(Throwable e) {

                        userMessageHandler.showDialog(mParentActivity, mParentActivity.getString(R.string.error_retrieving_data_title),
                                mParentActivity.getString(R.string.error_self_following_media) + e.getMessage(), 0);
                    }

                    @Override
                    public final void onNext(List<Media> arMedia) {

                        if (arMedia.size() == 0)
                            userMessageHandler.showToast(mParentActivity, mParentActivity.getString(R.string.warn_no_following));
                        else {

                            ArrayList<Media.MediaData> allMedia = new ArrayList<>();
                            for(Media media : arMedia){
                                allMedia.addAll(media.mediaData);
                            }
                            mBinding.gridview.setAdapter(new MediaGridViewAdapter(mParentActivity,FollowingViewModel.this, allMedia));
                        }
                    }
                });

great! please accept my answer if that's was helpful :) – yosriz Feb 16 '17 at 15:47 — yosriz, Feb 16 '17 at 15:47

RxJava flatMapIterable with concatMap

3 Answers3