android.os.NetworkOnMainThreadException using rxjava on android

Question

i'm having trouble implementing rxJava in order to check if there is internet connection on android i'm doing it like this:

on my launcher activity i have this in onCreate:

AndroidObservable.bindActivity(this,
            Observable.just(Utils.isActiveInternetConnection(Launcher.this)))
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Action1<Boolean>() {
                @Override
                public void call(Boolean aBoolean) {
                    if (aBoolean) {
                        Toast.makeText(Launcher.this, "There is internet connection", Toast.LENGTH_SHORT).show();
                    } else {
                        Toast.makeText(Launcher.this, "There is no internet connection", Toast.LENGTH_SHORT).show();
                    }
                }
            });

i have an Utils class it is a final class with static methods the methods in that the observable is using are this ones:

    public static boolean isActiveInternetConnection(Context context) {
    if (isNetworkAvailable(context)) {
        try {
            HttpURLConnection urlc = (HttpURLConnection) (new URL("http://www.google.com").openConnection());
            urlc.setRequestProperty("User-Agent", "Test");
            urlc.setRequestProperty("Connection", "close");
            urlc.setConnectTimeout(1500);
            urlc.connect();
            return (urlc.getResponseCode() == 200);
        } catch (IOException e) {
            Log.e("network", "Error checking internet connection", e);
        }
    } else {
        Log.d("network", "No network available!");
    }
    return false;
}

private static boolean isNetworkAvailable(Context context){
    ConnectivityManager cm = (ConnectivityManager) context.getSystemService(Context.CONNECTIVITY_SERVICE);
    NetworkInfo activeNetwork = cm.getActiveNetworkInfo();
    if (null != activeNetwork) {
        return true;
    }
    else {
        return false;
    }
}

i'm receiving android.os.NetworkOnMainThreadException and i can't find why, thanks in advance.

Answer 1

Observable.just(...) is called immediately on the calling thread (the main thread, in this case). Your code is effectively just an inlined version of this:

boolean activeConn = Utils.isActiveInternetConnection(Launcher.this);
AndroidObservable.bindActivity(this, 
        Observable.just(activeConn))
        .subscribeOn(...)
        ...

You've tried to move it off the main thread by calling subscribeOn() - but the call has already happened.

The way we handle this (and I'm not sure that this is the best way, but it works) is to defer the network or blocking call until subscription happens, set up the observable to run on the correct threads, and then subscribe:

AndroidObservable.bindActivity(this,
        Observable.defer(new Func0<Boolean>() {
            @Override
            public Observable<Observable<Boolean>> call() {
                return Observable.just(Utils.isActiveInternetConnection(Launcher.this));
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<Boolean>() {
            @Override
            public void call(Boolean aBoolean) {
                if (aBoolean) {
                    Toast.makeText(Launcher.this, "There is internet connection", Toast.LENGTH_SHORT).show();
                } else {
                    Toast.makeText(Launcher.this, "There is no internet connection", Toast.LENGTH_SHORT).show();
                }
            }
        });

Answer 2

AdamS is correct, however RxJava 2 now offers Observable.fromCallable() to defer an observable operation till subscription. A good reference: https://caster.io/lessons/fromcallable-converting-slow-methods-into-an-observable/

Some example code from my use-case:

Single.fromCallable(new Callable<Response>() {
        @Override
        public Response call() throws Exception {
            return NetworkGateway.networkRequest();
        }
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(observer);

Answer 3

I guess just just calls the method synchronously as it expects the boolean value and it tries to get it.

I am rather bad at RxJava but you may try something like this:

Observable<Boolean> onlineObservable = Observable.create(new Observable.OnSubscribe<Boolean>() {
    @Override
    public void call(Subscriber subscriber) {
        subscriber.onNext(Utils.isActiveInternetConnection(context));
    }
});
onlineObservable.subscribeOn(Schedulers.newThread()).subscribe(result -> {...});

Answer 4

this is my retrieve data from DataBase by RXAndroid code:

    Observable.create(new Observable.OnSubscribe<List<GRO_VO>>() {
        @Override
        public void call(Subscriber<? super List<GRO_VO>> subscriber) {

            String jsonIn;
            jsonIn =retrieveDataFromDB();
            Gson gson = new Gson();
            Type listType = new TypeToken<List<GRO_VO>>() {

            }.getType();   
            eventJoinList = gson.fromJson(jsonIn, listType);
            Log.d("RX",jsonIn);
            subscriber.onNext(eventJoinList);
        }
    })

                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<List<GRO_VO>>() {
        @Override
        public void call(List<GRO_VO> eventJoinList) {
            Log.d("RX", ".subscribe");
       recyclerView.setAdapter(new EventJoinAdapter(eventJoinList));
        }
    });

I think the just operator will emit data immediately, so it's is not useful to retrieve data from a database via network. It is very easy to use, but it can only be used for data that's already in the ram of the device.

I also had that problem, like @Baniares had, but after I use the create operator, all the problem gone...

From the RXJava documentation:

static <T> Observable<T> create(Observable.OnSubscribe<T> f)

Returns an Observable that will execute the specified function when a Subscriber subscribes to it.

Using the create operator can establish the standard process:

1 .subscribe(...) Subscriber(the sub class of class Observer) start the connection to Observable.

2 .subscribeOn(Schedulers.io()) collect a backGround Thread from the RX-ThreadPool

3 .create(...) retrieve data from server...during some netWork..etc

4 .observeOn(AndroidSchedulers.mainThread()) this means data will be set by the UI-Thread

5 Once we get the data , we can set the data in the onNext( ) method in the .subscribe( ) , the data will be set on the UI by UI-Thread since we make UI-thread do the work on the .observerOn(AndroidSchedulers.mainThread())

And note that if you use .create( ) operator,you must finished your observable in the .create( ) , others operator such like map,flatMap will not be executed after the .create( ) operator.

A very important concept we must need to know before start to use RXJava/RXAndroid. RX is a callback-based library, you tell RX what it should to do in what condition, it invoke your pass-in function(or in JAVA I may should to call them anonymous inner class... ) to achieve what you want.