NetworkBoundResource with Rx

NetworkBoundResource.kt

abstract class NetworkBoundResource<ResultType, RequestType> {

    private lateinit var result: Flowable<Resource<ResultType>>

    init {
        // Lazy disk observable.
        val diskObservable = Flowable.defer {
            loadFromDb()
                    // Read from disk on Computation Scheduler
                    .subscribeOn(Schedulers.computation())
        }

        // Lazy network observable.
        val networkObservable = Flowable.defer {
            createCall()
                    // Request API on IO Scheduler
                    .subscribeOn(Schedulers.io())
                    // Read/Write to disk on Computation Scheduler
                    .observeOn(Schedulers.computation())
                    .doOnNext { request: Response<RequestType> ->
                        try {
                            saveCallResult(processResponse(request))
                        } catch (e: Throwable) { //TODO throw error from api object
                            val jObjError = JSONObject(request.errorBody()?.string())
                            throw Exception(jObjError.getString("data"))
                        }
                    }
                    .flatMap { loadFromDb() }
        }

        loadFromDb().map {
            result = if (shouldFetch(it)) {
                networkObservable
                        .map<Resource<ResultType>> { Resource.Success(it) }
                        .onErrorReturn { Resource.Failure(it) }
                        // Read results in Android Main Thread (UI)
                        .observeOn(AndroidSchedulers.mainThread())
                        .startWith(Resource.Loading())
            } else {
                diskObservable
                        .map<Resource<ResultType>> { Resource.Success(it) }
                        .onErrorReturn { Resource.Failure(it) }
                        // Read results in Android Main Thread (UI)
                        .observeOn(AndroidSchedulers.mainThread())
                        .startWith(Resource.Loading())
            }
        }
    }

    fun asFlowable(): Flowable<Resource<ResultType>> {
        return result
    }

    private fun processResponse(response: Response<RequestType>): RequestType {
        return response.body()!!
    }

    protected abstract fun saveCallResult(request: RequestType)

    protected abstract fun loadFromDb(): Flowable<ResultType>

    protected abstract fun shouldFetch(data: ResultType): Boolean

    protected abstract fun createCall(): Flowable<Response<RequestType>>
}