Retrofit - android.os.NetworkOnMainThreadException with RxKotlin

Question

I've created rx function to call a network call from view-model in android, it parses network on main thread function.

I just change few line of code it worked. but i need to know the reason for this because its use same builder pattern to create a rx-call. once I tried with changing .doOnSubscribe() ,doOnComplete () , .applySchedulers() after the flatmap call it worked? how is this happened?

fun loadjobs(var countryID:String){
subscription.add(
repository.getMainJobsFromLocal(countryID)
          .doOnSubscribe { postProgress(StatusModel(Status.IN_PROGRESS))}
          .doOnComplete { postProgress(StatusModel(Status.COMPLETED)) }
          .applySchedulers()
          .flatMap {
           if (it.isNullOrEmpty()) {
              repository.getMainJobsFromServer(countryID)
           } else {
              Flowable.just(Response.success(it))
           }
          }
          .subscribe({
            if (it.isResponseOk()) {
             postProgress(StatusModel(Status.SUCCESS))
             mainJobResponse.postValue(it.body())
           } else {
             postProgress(StatusModel(Status.FAILED))
             mainJobResponse.postValue(null)
           }
          }, {
           postProgress(StatusModel(Status.FAILED))
           mainJobResponse.postValue(null)
        }))
}


fun loadjobs(var countryID){
subscription.add(
repository.getMainJobsFromLocal(countryID)
          .flatMap {
           if (it.isNullOrEmpty()) {
             repository.getMainJobsFromServer(countryID).flatMap {
               Flowable.just(it)
             }
           } else {
             Flowable.just(Response.success(it))
           }
          }.doOnSubscribe { postProgress(StatusModel(Status.IN_PROGRESS)) }
            .doOnComplete { postProgress(StatusModel(Status.COMPLETED)) }
            .applySchedulers()
            .subscribe({
              if (it.isResponseOk()) {
                postProgress(StatusModel(Status.SUCCESS))
                mainJobResponse.postValue(it.body())
               } else {
               postProgress(StatusModel(Status.FAILED))
               mainJobResponse.postValue(null)
              }
           }, {
            postProgress(StatusModel(Status.FAILED))
            mainJobResponse.postValue(null)
    }))
}

Answer 1

applySchedulers() after the flatmap call it worked? how is this happened?

observeOn() affects everything downstream. If you have a flatMap() after observeOn(), it gets executed on that scheduler.

Similarly subscribeOn() affects the upstream chain.

For these reasons, for most use cases you'd want to have the schedulers applied at the end of your rx chain and not in the middle.