RestApi.kt

interface RestApi<Request, Response> {

    /**
     * Function that will receive [Response] from server and return [Response]
     */
    fun obtainResponse(request: Request): Response
}

RestApiRepository.kt

/**
 * [Request] - input data
 * [Response] - response dto
 * [T] - final result
 */
abstract class RestApiRepository<Request, Response, T> @Inject constructor(
    private val restApi: RestApi<Request, Response>
) : NonNullFunction<RestCallback<T>, RestCallback<T>> {

    /**
     * Function that takes in [Request], converts it to [Response]
     * and passes it to [getResult] to return [T].
     *
     * @param request Input data that will be used to [obtainResponse].
     * @return Observable that will pass [RestCallback] to observer
     */
    fun getData(
        request: Request,
        applyScheduler: Scheduler = AndroidSchedulers.mainThread()
    ): Observable<RestCallback<T>> {
        val observable = Observable.create<RestCallback<T>> { emitter ->
            emitter.onNext(getLoadingMessage())
            try {
                val parsedResponse = restApi.obtainResponse(request)
                val result = getResult(parsedResponse)

                if (emitter.isDisposed.not()) {
                    emitter.onNext(getSuccessMessage(result))
                }
            } catch (e: Exception) {
                val parsedError = parseExceptionToResult(e) ?: return@create
                if (emitter.isDisposed.not()) {
                    emitter.onNext(parsedError)
                }
            }
            if (emitter.isDisposed.not()) {
                emitter.onComplete()
            }
        }
        return observable
            .observeOn(applyScheduler)
            .concatMap {
                Observable.just(apply(it))
            }
    }

    abstract fun getResult(response: Response) : T

    abstract fun parseExceptionToResult(exception: Exception): RestCallback.Error<T>?

    // methods for testsing observer results
    fun getSuccessMessage(result: T): RestCallback.Success<T> {
        return RestCallback.Success("Success", result)
    }

    fun getLoadingMessage(): RestCallback.Loading<T> {
        return RestCallback.Loading()
    }
}

RestCallback.kt

sealed class RestCallback<T> {
    class Success<T>(val message: String,val data: T): RestCallback<T>()
    class Error<T>(val errorType: ErrorType): RestCallback<T>()
    class Loading<T>: RestCallback<T>()
}

sealed class ErrorType {
    object InternetConnectionError: ErrorType()
    object TimeOutError: ErrorType()
    data class ServerErrorCodeException(val code: Int, val message: String): ErrorType()
    data class ParsingException(val exception: Throwable): ErrorType()
}