RxJava 2 - Fatal Exception thrown on Scheduler

Question

I trying to convert my Asynch task to JavaRx 2. I use google sheets api to download data from spreadsheets. (here is a link how this happens)

Here is a part of my code:

OnCreate:

/**
 * JavaRx
 */

//Observable
Observable<String> observable
        = Observable.create(
        new ObservableOnSubscribe<String>() {
            @Override

            public void subscribe(ObservableEmitter<String> e) throws Exception {
                //Use onNext to emit each item in the stream//
                e.onNext("https://docs.google.com/spreadsheets/d/1W5S5W2QH6WHjUcL1VMwqIqOdFYVleTopJNryQJGw568/gviz/tq?tqx=out:QUERY&tq=select+B,X,Y,Z");

                    //Once the Observable has emitted all items in the sequence, call onComplete//
                e.onComplete();
            }
        }
    ).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());

//Create our subscription
Observer<String> observer = new Observer<String>() {
    @Override
    public void onSubscribe(Disposable d) {
        Log.e(TAG, "onSubscribe " + Thread.currentThread().getName());
    }

    @Override
    public void onNext(String value) {
        try {
            String data = getLeagueData(value);
            mLeagues.add(autoProcessJsonLeague("Argentina Primera Division", returnJSON(data)));
        } catch (IOException e) {
            e.printStackTrace();
        }

        Log.e(TAG, "onNext: " + value + Thread.currentThread().getName());

    }

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

    @Override
    public void onComplete() {
        Log.e(TAG, "onComplete: All Done! " + Thread.currentThread().getName());
    }
};
observable.subscribe(observer);

Other Methods:

private String getLeagueData(String urlString) throws IOException {
//Download JSON file
    InputStream is = null;

    try {
        URL url = new URL(urlString);
        HttpsURLConnection conn = (HttpsURLConnection) url.openConnection();
        conn.setInstanceFollowRedirects(true);  //you still need to handle redirect manually.
        HttpsURLConnection.setFollowRedirects(true);
        conn.setReadTimeout(10000 /* milliseconds */);
        conn.setConnectTimeout(15000 /* milliseconds */);
        conn.setInstanceFollowRedirects(true);
        conn.setRequestMethod("GET");
        conn.setDoInput(true);
        // Starts the query
        conn.connect(); //ERROR HAPPENS HERE!
        int responseCode = conn.getResponseCode();
        is = conn.getInputStream();

        String contentAsString = convertStreamToString(is);
        //Log.d("contentAsString", contentAsString);
        return contentAsString;
    } catch (ProtocolException e) {
        e.printStackTrace();
    } catch (MalformedURLException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    } finally {
        if (is != null) {
            is.close();
        }
    }

    return null;
}

private String convertStreamToString(InputStream is) {
    BufferedReader reader = new BufferedReader(new InputStreamReader(is));
    StringBuilder sb = new StringBuilder();

    String line = null;
    try {
        while ((line = reader.readLine()) != null) {
            sb.append(line + "\n");
        }
    } catch (IOException e) {
        e.printStackTrace();
    } finally {
        try {
            is.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    return sb.toString();
}


private League autoProcessJsonLeague(String leagueName, JSONObject object) {
    //Get the data from the JSON string
    ArrayList<Team> teams = new ArrayList<>();
    try {
        JSONArray rows = object.getJSONArray("rows");
        for (int r = 0; r < rows.length(); ++r) {
            JSONObject row = rows.getJSONObject(r);
            JSONArray columns = row.getJSONArray("c");
            String name = columns.getJSONObject(0).getString("v");
            int points = columns.getJSONObject(1).getInt("v");
            double hGoalAv = columns.getJSONObject(2).getDouble("v");
            double aGoalAv = columns.getJSONObject(3).getDouble("v");
            hGoalAv = Utilities.round(hGoalAv, 2);
            aGoalAv = Utilities.round(aGoalAv, 2);
            teams.add(new Team(name, points, hGoalAv, aGoalAv));
            //Log.d("Team", name + " " + hGoalAv + " " + aGoalAv);
        }
    } catch (JSONException e) {
        e.printStackTrace();
        e.printStackTrace();
    }
    return new League(leagueName, teams);
}

So I create an observable, I subscribe on IO thread and observeOn the main thread. With onNext I send the url link to the observer and then I try to connect to the server to download the json string file.

Error happens on method getLeagueData() on the line conn.connect(); It says java.lang.IllegalStateException: Fatal Exception thrown on Scheduler.

FULL STACK TRACE ERROR:

08-16 08:53:09.934 29841-29841/com.aresproductions.bettingtools E/AndroidRuntime: FATAL EXCEPTION: main
Process: com.aresproductions.bettingtools, PID: 29841
java.lang.IllegalStateException: Fatal Exception thrown on Scheduler.
  at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:111)
  at android.os.Handler.handleCallback(Handler.java:751)
  at android.os.Handler.dispatchMessage(Handler.java:95)
  at android.os.Looper.loop(Looper.java:154)
  at android.app.ActivityThread.main(ActivityThread.java:6195)
  at java.lang.reflect.Method.invoke(Native Method)
  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:874)
  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:764)
 Caused by: android.os.NetworkOnMainThreadException
  at android.os.StrictMode$AndroidBlockGuardPolicy.onNetwork(StrictMode.java:1303)
  at java.net.Inet6AddressImpl.lookupHostByName(Inet6AddressImpl.java:86)
  at java.net.Inet6AddressImpl.lookupAllHostAddr(Inet6AddressImpl.java:74)
  at java.net.InetAddress.getAllByName(InetAddress.java:752)
  at com.android.okhttp.internal.Network$1.resolveInetAddresses(Network.java:29)
  at com.android.okhttp.internal.http.RouteSelector.resetNextInetSocketAddress(RouteSelector.java:187)
  at com.android.okhttp.internal.http.RouteSelector.nextProxy(RouteSelector.java:156)
  at com.android.okhttp.internal.http.RouteSelector.next(RouteSelector.java:98)
  at com.android.okhttp.internal.http.HttpEngine.createNextConnection(HttpEngine.java:346)
  at com.android.okhttp.internal.http.HttpEngine.connect(HttpEngine.java:329)
  at com.android.okhttp.internal.http.HttpEngine.sendRequest(HttpEngine.java:247)
  at com.android.okhttp.internal.huc.HttpURLConnectionImpl.execute(HttpURLConnectionImpl.java:457)
  at com.android.okhttp.internal.huc.HttpURLConnectionImpl.connect(HttpURLConnectionImpl.java:126)
  at com.android.okhttp.internal.huc.DelegatingHttpsURLConnection.connect(DelegatingHttpsURLConnection.java:89)
  at com.android.okhttp.internal.huc.HttpsURLConnectionImpl.connect(HttpsURLConnectionImpl.java)
  at com.aresproductions.bettingtools.MainActivity.getLeagueData(MainActivity.java:307)
  at com.aresproductions.bettingtools.MainActivity.access$000(MainActivity.java:80)
  at com.aresproductions.bettingtools.MainActivity$2.onNext(MainActivity.java:180)
  at com.aresproductions.bettingtools.MainActivity$2.onNext(MainActivity.java:171)
  at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:198)
  at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:250)
  at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:109)
  at android.os.Handler.handleCallback(Handler.java:751) 
  at android.os.Handler.dispatchMessage(Handler.java:95) 
  at android.os.Looper.loop(Looper.java:154) 
  at android.app.ActivityThread.main(ActivityThread.java:6195) 
  at java.lang.reflect.Method.invoke(Native Method) 
  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:874) 
      at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:764) 
08-16 08:53:09.936 29841-29841/com.aresproductions.bettingtools E/MQSEventManagerDelegate: failed to get MQSService.

Thanks in advance!

Answer 1

The problem is that you are doing the network call on the main thread. Though you have subscribed on Schedulers.io(), the onNext() method where you are doing the network call will be called on main thread since you are observing on main thread observeOn(AndroidSchedulers.mainThread()).

The solution will be to call the getLeagueData(String urlString) inside subscribe() method of observable and call e.onNext(result) with the result of the network call.

Answer 2

I assume you get a network on main thread exception because the network call is at the wrong place. You should have moved it into the create thus it gets executed on the IO scheduler instead as part of observing the trivial string:

Observable<String> observable = Observable.create(
    new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {

            String value = "https://docs.google.com/spreadsheets/d/1W5S5W2QH6WHjUcL1VM" +
                "wqIqOdFYVleTopJNryQJGw568/gviz/tq?tqx=out:QUERY&tq=select+B,X,Y,Z";

            String data = getLeagueData(value);

            //Use onNext to emit the item in the stream//
            e.onNext(data);

            /* Once the Observable has emitted all items 
            in the sequence, call onComplete */
            e.onComplete();
        }
    }
)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());

//Create our subscription
Observer<String> observer = new Observer<String>() {
    @Override
    public void onSubscribe(Disposable d) {
        Log.e(TAG, "onSubscribe " + Thread.currentThread().getName());
    }

    @Override
    public void onNext(String data) {
        try {
            mLeagues.add(autoProcessJsonLeague(
                "Argentina Primera Division", returnJSON(data)));
        } catch (IOException e) {
            e.printStackTrace();
        }


        Log.e(TAG, "onNext: " + value + Thread.currentThread().getName());

    }

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

    @Override
    public void onComplete() {
        Log.e(TAG, "onComplete: All Done! " + Thread.currentThread().getName());
    }
};
observable.subscribe(observer);

You may also want to move

autoProcessJsonLeague(
                "Argentina Primera Division", returnJSON(data))

into the ObservableOnSubscribe in case the processing is expensive.