How i can handle network disconnect using RxJava 2

Question

I have some problems, which bound with lose of network connection. How i can handle it in RxJava 2? Thank you very much.

i have this method:

Disposable disposable = api.setStatus(params)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .doOnSubscribe(listener::onPreExecute)
            .doFinally(listener::onPostExecute)
            .subscribe(serviceRequest -> handleResponse(listener, serviceRequest), listener::onError);

//////////////

@POST("set_status")
Single<OrderResponse> setStatus(@FieldMap Map<String, String> params);

Answer 1

New answer after question was update:

You can handle it in doOnError or your listener::onError:

Disposable disposable = api.setStatus(params)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io())
                .doOnError(error->{
                    handleError(error)
                })
                .subscribe(serviceRequest -> handleResponse(listener, serviceRequest), listener::onError);

void handleError(Throwable error){
    if (error instanceof IOException){
        // handle network error
    } else {
        if(error instanceof SocketTimeoutException){
            // handle timeout error 
        }
    }


}

Old answer about subscribing to connection change:

For this purposes you need to catch the network connection change and dispatch it via BehaviorSubject.

Like this:

class NetworkManager(
        private val context: Context
) {
    private val state: BehaviorSubject<Boolean> = BehaviorSubject.create()

    private val receiver = object : BroadcastReceiver() {
        override fun onReceive(c: Context?, intent: Intent?) {
            state.onNext(isConnected())
        }
    }

    init {
        val intentFilter = IntentFilter()
        intentFilter.addAction(ConnectivityManager.CONNECTIVITY_ACTION)
        context.registerReceiver(receiver, intentFilter)
        state.onNext(isConnected())
    }

    fun subscribe(): Observable<Boolean> {
        return state
    }

    fun isConnected(): Boolean {
        val cm = context.applicationContext.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
        val netInfo = cm.activeNetworkInfo
        return netInfo != null && netInfo.isConnectedOrConnecting
    }
}

Create this classs in your Application scope and it will be ok