Getting NetworkOnMainThreadException with RxKotlin

Question

I'm trying to make a network request using RxKotlin, but keep getting a NetworkOnMainThreadException I'm subscribing on the main thread, so I'm not sure why it's not taking it off of the UI thread.

Here is where I subscribe to the Observable

weatherInteractor.getWeather(lat, lng)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                        {response ->
                            try {
                                val jsonData = response.body().string()
                                val currentWeather = getCurrentWeatherData(jsonData)
                                view!!.displayCurrentWeather(currentWeather)
                            } catch (e: JSONException) {
                                Log.d("Present JSON Exception", e.message)
                            } catch (e: IOException) {
                                Log.d("Present IO Exception", e.message)
                            }
                        },
                        {
                            error ->
                            error.printStackTrace()
                        }
                )
    }

Here is where I create my Observable

fun getWeather(lat: Double, lng: Double): Observable<Response> {
        val URL = ""
        val client = OkHttpClient()
        val request = Request.Builder()
                .url(URL)
                .build()

        return Observable.create { em ->
            try {
                val response = client.newCall(request).execute()
                em.onNext(response)
                em.onComplete()
            } catch (err: IOException) {
                err.printStackTrace()
                em.onError(err)
            }
        }
    }

Answer 1

It seems that you confuse subscribeOn and observeOn methods.

subscribeOn specifies the scheduler observable will be created on and will operate on. (You specify it once, position doesn't matter).

observeOn changes the scheduler for every action you type after it. You can use it multiple times and each set of actions will be executed on specified scheduler.

Here is an example:

Observable
            .just("test")
            .subscribeOn(Schedulers.io())
            .map(s -> { //this and all Observable code is executed on Schedulers.io()

                return s;
            })
            .observeOn(Schedulers.computation())
            .map(s -> { // executed on Schedulers.computation()

                return s;
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(s -> { // executed on Android main thread

            }, throwable -> {

            });