Abstract coroutines structure

user_5827565 0
Viewed 237

With RxJava we can do something like this:

BaseViewModel

protected void subscribe(Completable completable, MutableLiveData<Response> response) {
    mDisposable.add(
            completable.observeOn(AndroidSchedulers.mainThread())
                    .subscribeOn(Schedulers.io())
                    .doOnSubscribe(disposable -> response.setValue(Response.loading()))
                    .doFinally(() -> response.setValue(Response.idle()))
                    .subscribe(
                            () -> response.setValue(Response.success(true)),
                            e -> response.setValue(Response.error(e))
                    )
    );
}

protected <T> void subscribe(Single<T> single, MutableLiveData<Response> response) {
    mDisposable.add(
            single.observeOn(AndroidSchedulers.mainThread())
                    .subscribeOn(Schedulers.io())
                    .doOnSubscribe(disposable -> response.setValue(Response.loading()))
                    .doFinally(() -> response.setValue(Response.idle()))
                    .subscribe(
                            result -> response.setValue(Response.success(result)),
                            e -> response.setValue(Response.error(e))
                    )
    );
}

Then, from repository we getting Single/Complete and pass it to our custom subscribe(), then we get generic Result with data(optional), very easy way to work with asynchronous requests.

How we can abstract coroutines with similar structure, instead of write Launch in every method in ViewModel and try/catch error manually?

kotlin-coroutines kotlin android
3 Answers

You are probably looking for something like this

CoroutineWrapper

fun <T> ViewModel.apiCx(context: CoroutineContext = Dispatchers.Default, init: suspend CxWrapper<T>.() -> Unit) {
    val wrap = CxWrapper<T>(context)
    wrap.launch {
        try {
            init.invoke(wrap)
            callCx(wrap)
        } catch (e: Exception) {
            e.printStackTrace()
        }
    }
}

private fun <T> callCx(wrap: CxWrapper<T>) {
    val response: Response<T>? = wrap.request

    response?.let {
        if (it.isSuccessful) {
            wrap.success(it.body())
        } else {
            wrap.fail(Pair(it.code(), it.message()))
        }
    }
}

class CxWrapper<T>(override val coroutineContext: CoroutineContext) : CoroutineScope {

    var request: Response<T>? = null

    internal var success: (T?) -> Unit = {}
    internal var fail: (Pair<Int, String?>) -> Unit = {}

    fun success(onSuccess: (T?) -> Unit) {
        success = onSuccess
    }

    fun error(onError: (Pair<Int, String?>) -> Unit) {
        fail = onError
    }
}

you can have this as a separate helper class and to use this from your ViewModel

apiCx<YourModelClass> {
        request = yourApiCall()

        success { yourModelClass ->
            Log.d(TAG, "success")
        }

        error {
            Log.e(TAG, "error")
        }
    }
user_68451310

You would just do the same, just adapted to coroutines. Just replace the different stream types with the suspension methods you need.

protected inline fun <T> MutableLiveData<Response>.subscribe(single: suspend () -> T) {
    viewModelScope.launch {
        try {
            value = Response.loading()
            value = withContext(Dispatchers.IO) {
                Response.success(single())
            }
        } catch(e: Throwable) {
            value = Response.error(e)
        } finally {
            value = Response.idle()
        }
    }

To use it just call with the livedata as receiver

responseLiveData.subscribe<T> {
    singleFromRepo()
}
responseLiveData.subscribe<Unit> {
    completableFromRepo()
}
user_33852120

Instead of closely following the code you already have with minimal adaptations, I suggest you review your design altogether when migrating to coroutines.

One important principle embedded into coroutines is structured concurrency. This isn't just about the coroutine scopes and cancellation, it is also about the use of futures by any name (be it CompletionStage, Deferred, Task, Single or any other). According to structured concurrency, a future is basically equivalent to a live thread that has no defined scope. You should avoid them.

Instead you should have clearly delineated places in the code that launch new concurrent work contained within a single top-level block of code provided at the launch site.

So far, that implies that you do have a launch block at each entry point into your code from the Android framework, and that's a lot of places due to the nature of the callback-oriented programming model.

However, everything within that block should be coded according to structured concurrency. If you have just one network call to make, your code is entirely sequential: make the call, get the response, process it. The network calls themselves become suspend functions that complete with the result of the call and do not accept callbacks. All the traditional design patterns from the world of blocking calls apply here.

See here for an intro to using coroutines with LiveData, it may help you map your design to the coroutine-oriented one:

https://developer.android.com/topic/libraries/architecture/coroutines#livedata

user_11038720
Related