Retrofit RxJava global error handling in Kotlin

ErrorInterceptor.kt

// for any errors that should be handled before being handed off to RxJava. 
// In other words global error logic. 
// An example might be 401 when not logging in

import okhttp3.Interceptor
import okhttp3.Response

class ErrorInterceptor: Interceptor {

    override fun intercept(chain: Interceptor.Chain?): Response {

        val originalResponse = chain!!.proceed(chain.request())

        if (shouldLogout(originalResponse)) {
            // your logout logic here

            // send empty response down the chain
            return Response.Builder().build()

        }
        return originalResponse
    }

    private fun shouldLogout(response: Response) : Boolean {
        if (response.isSuccessful) {
            return false
        }

        // 401 and auth token means that we need to logout
        return (response.code() == 401 &&
                !response.headers().names().contains(AUTH_HEADER_KEY))
    }
}

RetrofitException.kt

import okhttp3.ResponseBody
import your.package.here.ServerError
import retrofit2.Converter
import retrofit2.Response
import retrofit2.Retrofit
import timber.log.Timber
import java.io.IOException

class RetrofitException(private val _message: String?,
                        private val _url: String?,
                        private val _response: Response<*>?,
                        private val _kind: Kind,
                        private val _exception: Throwable?,
                        private val _retrofit: Retrofit?
                        ) : RuntimeException(_message, _exception) {

    private var _errorData : ServerError? = null

    companion object {
        fun httpError(url: String, response: Response<*>, retrofit: Retrofit): RetrofitException {
            val message = response.code().toString() + " " + response.message()
            return RetrofitException(message, url, response, Kind.HTTP, null, retrofit)
        }

        fun httpErrorWithObject(url: String, response: Response<*>, retrofit: Retrofit): RetrofitException {
            val message = response.code().toString() + " " + response.message()
            val error = RetrofitException(message, url, response, Kind.HTTP_422_WITH_DATA, null, retrofit)
            error.deserializeServerError()
            return error
        }

        fun networkError(exception: IOException): RetrofitException {
            return RetrofitException(exception.message, null, null, Kind.NETWORK, exception, null)
        }

        fun unexpectedError(exception: Throwable): RetrofitException {
            return RetrofitException(exception.message, null, null, Kind.UNEXPECTED, exception, null)
        }
    }

    /** The request URL which produced the error. */
    fun getUrl() = _url

    /** Response object containing status code, headers, body, etc. */
    fun getResponse() = _response

    /** The event kind which triggered this error. */
    fun getKind() = _kind

    /** The Retrofit this request was executed on */
    fun getRetrofit() = _retrofit

    /** The data returned from the server in the response body*/
    fun getErrorData() : ServerError?  = _errorData

    private fun deserializeServerError() {
        if (_response != null && _response.errorBody() != null) {
            try {
                _errorData =  getErrorBodyAs(ServerError::class.java)
            } catch (e: IOException) {
                Timber.tag("Retrofit servererror deserialization").e(e)
            }
        }
    }

    /**
     * HTTP response body converted to specified `type`. `null` if there is no
     * response.

     * @throws IOException if unable to convert the body to the specified `type`.
     */
    @Throws(IOException::class)
    fun <T> getErrorBodyAs(type: Class<T>): T? {
        if (_response == null || _response.errorBody() == null || _retrofit == null) {
            return null
        }
        val converter : Converter<ResponseBody, T> =
                _retrofit.responseBodyConverter(type, arrayOfNulls<Annotation>(0))
        return converter.convert(_response.errorBody())
    }

    enum class Kind {
        /** An [IOException] occurred while communicating to the server.  */
        NETWORK,
        /** A non-200 HTTP status code was received from the server.  */
        HTTP,
        HTTP_422_WITH_DATA,
        /**
         * An internal error occurred while attempting to execute a request. It is best practice to
         * re-throw this exception so your application crashes.
         */
        UNEXPECTED
    }
}

RetrofitServiceProvider.kt

import io.reactivex.Observable
import io.reactivex.schedulers.Schedulers
import retrofit2.Call
import retrofit2.CallAdapter
import retrofit2.HttpException
import retrofit2.Retrofit
import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory
import java.io.IOException
import java.lang.reflect.Type

class WebRequest {

    val retrofit: Retrofit

    init {

        val okBuilder = OkHttpClient().newBuilder()
        okBuilder.networkInterceptors().add(ErrorInterceptor())

        retrofit = Retrofit.Builder()
                .addCallAdapterFactory(RxErrorHandlingCallAdapterFactory.create())
                .addConverterFactory(GsonConverterFactory.create())
                .baseUrl(BASE_URL)
                .client(okBuilder.build())
                .build()
    }



}

RxErrorHandlingCallAdapterFactory.kt

// Wraps "regular" Retrofit errors in custom RetrofitException class

import io.reactivex.Observable
import io.reactivex.schedulers.Schedulers
import retrofit2.Call
import retrofit2.CallAdapter
import retrofit2.HttpException
import retrofit2.Retrofit
import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory
import java.io.IOException
import java.lang.reflect.Type

class RxErrorHandlingCallAdapterFactory: CallAdapter.Factory() {

    private val _original by lazy {
        RxJava2CallAdapterFactory.createWithScheduler(Schedulers.io())
    }

    companion object {
        fun create() : CallAdapter.Factory = RxErrorHandlingCallAdapterFactory()
    }

    override fun get(returnType: Type, annotations: Array<Annotation>, retrofit: Retrofit): CallAdapter<*, *> {
        val wrapped = _original.get(returnType, annotations, retrofit) as CallAdapter<out Any, *>
        return RxCallAdapterWrapper(retrofit, wrapped)
    }

    private class RxCallAdapterWrapper<R>(val _retrofit: Retrofit,
                                          val _wrappedCallAdapter: CallAdapter<R, *>
                                          ): CallAdapter<R, Observable<R>> {

        override fun responseType(): Type = _wrappedCallAdapter.responseType()


        @Suppress("UNCHECKED_CAST")
        override fun adapt(call: Call<R>): Observable<R> {
            val adapted = (_wrappedCallAdapter.adapt(call) as Observable<R>)
            adapted.onErrorResumeNext { throwable: Throwable ->
                Observable.error(asRetrofitException(throwable))
            }

            return adapted
        }

        private fun asRetrofitException(throwable: Throwable): RetrofitException {
            // We had non-200 http error
            if (throwable is HttpException) {
                val response = throwable.response()

                if (throwable.code() == 422) {
                    // on out api 422's get metadata in the response. Adjust logic here based on your needs 
                    return RetrofitException.httpErrorWithObject(response.raw().request().url().toString(), response, _retrofit)
                } else {
                    return RetrofitException.httpError(response.raw().request().url().toString(), response, _retrofit)
                }
            }

            // A network error happened
            if (throwable is IOException) {
                return RetrofitException.networkError(throwable)
            }

            // We don't know what happened. We need to simply convert to an unknown error
            return RetrofitException.unexpectedError(throwable)
        }

    }
}

sampleClient.kt

.subscribe({
    // do success action
}, { it ->    
    view.displayError(getLoginErrorMessage(it))
})

private fun getLoginErrorMessage(exception: Throwable) : String {
    if (exception is RetrofitException) {
        when (exception.getKind()) {

            RetrofitException.Kind.HTTP_422_WITH_DATA ->
                return exception.getErrorData()!!.getMessage()

            RetrofitException.Kind.HTTP ->
                return R.string.default_http_error_message

            RetrofitException.Kind.NETWORK ->
                return R.string.default_network_error_message

            RetrofitException.Kind.UNEXPECTED->
                return R.string.default_unexpected_error_message
        }

    }
    return R.string.default_error_message
}

To fix this Crash :
Caused by: java.lang.ClassCastException: io.reactivex.internal.operators.observable.ObservableSingleSingle cannot be cast to io.reactivex.Observable
at ...$RxCallAdapterWrapper.adapt(RxErrorHandlingCallAdapterFactory.kt:37)
at ...$RxCallAdapterWrapper.adapt(RxErrorHandlingCallAdapterFactory.kt:28)

Solution :
override fun adapt(call: Call): Any {
return when (val result = _wrappedCallAdapter.adapt(call)) {
is Single<> -> result.onErrorResumeNext(Function { throwable -> Single.error(asOneAppServiceException(throwable)) })
is Observable<> -> result.onErrorResumeNext(Function { throwable -> Observable.error(asOneAppServiceException(throwable)) })
is Completable -> result.onErrorResumeNext(Function { throwable -> Completable.error(asOneAppServiceException(throwable)) })
is Flowable<*> -> result.onErrorResumeNext(Function { throwable -> Flowable.error(asOneAppServiceException(throwable)) })
else -> result
}
}