Retrofit 2 + RxJava chaining similar requests together cleanly

Question

Goal

Within my application, I wish to use Retrofit 2 to upload some files to my web server (multiple at a time, in this case pictures.) and upon completion of the upload of each file I want to update a SQL table with the path to that particular file.

Tried

I'm new to using functional programming paradigms, so my understand might be mislead here. I have two different objects, a FileResponse (DTO representing the response from my web server after a file is uploaded) and a Photo object. This Photo object has two fields, and acts as the Entity that is persisted within my backend server. It's field correspond to columns, and uploading each of these objects individually works fine. What I've tried to do is do this dual operation in one sweep.

Here is what I have tried:

List<Observable<FileResponse>> mObservables = new ArrayList<>(3);

...
...

Observable.merge(mObservables).
            map(new Func1<FileResponse, Observable<Photo>>() {
                @Override
                public Observable<Photo> call(FileResponse fileResponse) {
                    Photo photo = new Photo();
                    photo.setPath(fileResponse.getUrl());
                    photo.setId("demo123");
                    return mPhotoService.createPhoto(photo);
                }
            }).forEach(
                new Action1<Observable<Photo>>() {
                    @Override
                    public void call(Observable<Photo> photoObservable) {

                    }
                },
                new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        Log.e(TAG, "Error: " + throwable.getMessage());
                    }
                }
            );

Here is the logic I had behind it:

merge( ) — combine multiple Observables into one.

I want my List of Observables I created to be merged together so I perform the same action on every single one of them. In this case, the Observables are of type FileResponse. This is the upload to the file web service.

map ( ) - transform the items emitted by an Observable by applying a function to each item.

In this case, I want to apply a function to the FileResponse Observables that returns a new Observable, Observable.

forEach( ) — invoke a function on each item emitted by the Observable; block until the Observable completes.

This would then take each emitted Observable and upload it. In this case though, this part fails. The files successfully upload, but the forEach section just returns "Error: null".

Question

Is this understanding close to what I am trying to achieve? Again, I just wish to chain multiple requests together, performing new requests when the other ones succeed. Is there a better / correct way to do this?

Edit Note:

I forgot to mention, I am able to achieve what I want like so:

Observable.merge(mObservables)
            .map(new Func1<FileResponse, Observable<Photo>>() {
                @Override
                public Observable<Photo> call(FileResponse fileResponse) {
                    Photo photo = new Photo();
                    photo.setPath(fileResponse.getUrl());
                    photo.setDogId("123");
                    return mPhotoService.createPhoto(photo);
                }
            })
           .subscribe(new Subscriber<Observable<Photo>>() {
                    @Override
                    public void onCompleted() {
                        Log.d(TAG, "Save complete!");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "Error: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Observable<Photo> photoObservable) {
                        photoObservable.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread())
                                .subscribe(new Subscriber<Photo>() {
                                    @Override
                                    public void onCompleted() {
                                        Log.d(TAG, "Here!");
                                    }

                                    @Override
                                    public void onError(Throwable e) {
                                        Log.d(TAG, "Error! " + e.getMessage());
                                    }

                                    @Override
                                    public void onNext(Photo photo) {
                                        Log.d(TAG, "Woo!!");
                                    }
                                });
                    }
                }
            );

But I was trying to find a possibly cleaner way to do this. If this is the only way, fair enough. Just wanted to get more opinions on what I was thinking to see what I can do to make it looked better (or best practices).

AndroidEx AndroidEx · Accepted Answer · 2016-03-19T01:49:15

To make it cleaner (and avoid additional subscription to the items in onNext) you can switch map() for flatMap():

Observable.merge(mObservables)
                .flatMap(new Func1<FileResponse, Observable<Photo>>() {
                    @Override
                    public Observable<Photo> call(FileResponse fileResponse) {
                        Photo photo = new Photo();
                        return mPhotoService.createPhoto(photo);
                    }
                })
                .forEach(new Action1<Photo>() {
                    @Override
                    public void call(Photo photo) {

                    }
                });

As you can see in the snippet, the next operator after the flatMap is going to see Photo objects (compare to Observable<Photo> after just map()).

Retrofit 2 + RxJava chaining similar requests together cleanly

1 Answers