Android RxJava asynchronous call in map function

Question

On the change "SortBy", my program will do a NetworkIO to retrieve the top movies and display them.

However, it seems that though I have done subscribeOn(Schedulers.io()), the NetworkIO MovieDB.getPopular() and MovieDB.getTopRated() in the function call in map are excuted on the main thread and I get a android.os.NetworkOnMainThreadException.

I was wondering how to make the public Movie[] call(SortBy sortBy) asynchronous.

sortObservable.map(new Func1<SortBy, Movie[]>() {
    @Override
    public Movie[] call(SortBy sortBy) {
        try {
            switch (sortBy) {
                case POPULAR:
                    return MovieDB.getPopular(); // NETWORK IO
                case TOP_RATED:
                    return MovieDB.getTopRated(); // NETWORK IO
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (JSONException e) {
            e.printStackTrace();
        }
        return new Movie[0];
    }
})
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<Movie[]>() {
            @Override
            public void call(Movie[] movies) {
                imageAdapter.loadData(movies);
            }
        });

you can use `flatMap` instead, and wrap your io calls in an observable. — njzk2, Jan 17 '17 at 03:52
`subscribeOn` is where `subscribe` is called. Use `observeOn` to control the thread of the previous observable. — njzk2, Jan 17 '17 at 03:56
@njzk2 I still get `android.os.NetworkOnMainThreadException`, source code here: https://github.com/zizhengwu/Popular-Movies-Stage-1/blob/load-image/app/src/main/java/com/zizhengwu/popular_movies_stage_1/MainActivity.java#L75 — knx54693, Jan 17 '17 at 06:33
@kmx no, that's not how you would wrap this in an observable. This does not change the fact that the io call is made in the same thread as the flatmap method is called. Use a callable and `fromCallable` instead, so you can give it a thread to use when subscribed to — njzk2, Jan 17 '17 at 14:53
hi @njzk2, I still don't know how to chain `callable` and `fromCallable` with my map function. Could you provide some code for me? — knx54693, Jan 18 '17 at 13:54
If you have java 8 or retrolambda in your project, you can simply write `return Observable.fromCallable(MovieDB::getPopular);` — njzk2, Jan 18 '17 at 16:19
@njzk2 The code still gets `android.os.NetworkOnMainThreadException` https://github.com/zizhengwu/Popular-Movies-Stage-1/blob/fromCallable/app/src/main/java/com/zizhengwu/popular_movies_stage_1/MainActivity.java#L75 — knx54693, Jan 18 '17 at 17:18

Pavan Kumar · Answer 1 · 2017-01-17T08:06:51.950

Please check if the below works for you. It uses flatMap instead of map.

sortObservable.flatMap(new Func1<SortBy, Observable<Movie[]>>() {

        @Override
        public Observable<Movie[]> call(SortBy sortBy) {
            try {
                switch (sortBy) {
                    case POPULAR:
                        return Observable.just(MovieDB.getPopular()); // NETWORK IO
                    case TOP_RATED:
                        return Observable.just(MovieDB.getTopRated()); // NETWORK IO
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (JSONException e) {
                e.printStackTrace();
            }
            return Observable.just(new Movie[0]);
        }
    }).subscribe(new Action1<Movie[]>() {
        @Override
        public void call(Movie[] movies) {
            imageAdapter.loadData(movies);
        }
    });

From your source code on Github, it seems like you are using synchronous mode of executing requests using OkHttp. OkHttp also supports asynchronous requests and that can be preferred. Below would be the changes required in few of the methods.

run method should consume enqueue instead of execute.

Observable<String> runAsync(String url){
return Observable.create(subscriber -> {
    Request request = new Request.Builder().url(url).build();

    client.newCall(request).enqueue(new Callback() {

        @Override
        public void onResponse(Call call, Response response) throws IOException {
            subscriber.onNext(response.body().string());
        }

        @Override
        public void onFailure(Call call, IOException e) {
            subscriber.onError(e);
        }
    });
});
}

getApi can return an Observable<Movie[]> instead of Movie[]

public Observable<Movie[]> getApiAsync(String type){
return runAsync("http://api.themoviedb.org/3/movie/" + type
        + "?api_key=412e9780d02673b7599233b1636a0f0e").flatMap(response -> {
            Gson gson = new Gson();
            Map<String, Object> map = gson.fromJson(response,
                    new TypeToken<Map<String, Object>>() {
                    }.getType());
            Movie[] movies = gson.fromJson(gson.toJson(map.get("results")),
                    Movie[].class);
            return Observable.just(movies);
        });
}

Still get `android.os.NetworkOnMainThreadException`. Source code here: https://github.com/zizhengwu/Popular-Movies-Stage-1/blob/load-image/app/src/main/java/com/zizhengwu/popular_movies_stage_1/MainActivity.java#L75 — knx54693, Jan 17 '17 at 06:25
That issue does not relate with rx, you can use [AsyncTask](https://developer.android.com/reference/android/os/AsyncTask.html) to overcome that. You may please refer: http://stackoverflow.com/a/6343299/3940047 — Pavan Kumar, Jan 17 '17 at 06:34
I am just curious about why my code doesn't work. From my understanding, the network IO should be in a different thread. I followed this earlier https://stackoverflow.com/questions/32687921/using-rxjava-and-okhttp and it worked well for me. — knx54693, Jan 17 '17 at 06:40
Though not sure about that part, have made changes in answer on using OkHttp in async mode. Please take a look. — Pavan Kumar, Jan 17 '17 at 08:07

knx54693 · Accepted Answer · 2017-01-19T13:19:47.743

Finally I sort it out by myself:

sortObservable.flatMap(new Func1<SortBy, Observable<Movie[]>>() {

    @Override
    public Observable<Movie[]> call(SortBy sortBy) {
        switch (sortBy) {
            case POPULAR:
                return Observable.fromCallable(() -> MovieDB.getPopular()).subscribeOn(Schedulers.io());
            case TOP_RATED:
                return Observable.fromCallable(() -> MovieDB.getTopRated()).subscribeOn(Schedulers.io());
            default:
                return Observable.fromCallable(() -> new Movie[0]).subscribeOn(Schedulers.io());
        }
    }
})
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<Movie[]>() {
            @Override
            public void call(Movie[] movies) {
                imageAdapter.loadData(movies);
            }
        });

Android RxJava asynchronous call in map function

2 Answers2