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
– njzk2, Commented Jan 17, 2017 at 3:52
subscribeOn is where subscribe is called. Use observeOn to control the thread of the previous observable. — njzk2
– njzk2, Commented Jan 17, 2017 at 3:56
@njzk2 I still get android.os.NetworkOnMainThreadException, source code here: github.com/zizhengwu/Popular-Movies-Stage-1/blob/load-image/app/… — knx54693
– knx54693, Commented Jan 17, 2017 at 6: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
– njzk2, Commented Jan 17, 2017 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
– knx54693, Commented Jan 18, 2017 at 13:54

Pavan Kumar · Accepted Answer · 2017-01-17 08:06:51Z

1

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);
        });
}

edited Jan 17, 2017 at 8:06

answered Jan 17, 2017 at 5:02

Pavan Kumar

4,9104 gold badges35 silver badges50 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

knx54693 Over a year ago

Still get android.os.NetworkOnMainThreadException. Source code here: github.com/zizhengwu/Popular-Movies-Stage-1/blob/load-image/app/…

Pavan Kumar Over a year ago

That issue does not relate with rx, you can use AsyncTask to overcome that. You may please refer: stackoverflow.com/a/6343299/3940047

knx54693 Over a year ago

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 stackoverflow.com/questions/32687921/using-rxjava-and-okhttp and it worked well for me.

Pavan Kumar Over a year ago

Though not sure about that part, have made changes in answer on using OkHttp in async mode. Please take a look.

knx54693 · Accepted Answer · 2017-01-19 13:19:47Z

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);
            }
        });

Collectives™ on Stack Overflow

Android RxJava asynchronous call in map function

2 Answers 2

4 Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

4 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related