如何使用协程Flow定义链式调用DSL由于Flow属于实验性质的，暂时用Rx代替，这需要使用coroutines的Rx扩展包

74 阅读 0 评论 49 点赞

我是靠谱客的博主激动茉莉，最近开发中收集的这篇文章主要介绍如何使用协程Flow定义链式调用DSL由于Flow属于实验性质的，暂时用Rx代替，这需要使用coroutines的Rx扩展包，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

定义DSL

定义启动flow的函数：


    protected val _spinner: MutableLiveData<Boolean> by lazy { MutableLiveData<Boolean>() }
    val spinner: LiveData<Boolean>
        get() = _spinner
        
    fun <T> loadAsync(showSpinner: Boolean = true, block: suspend () -> T): Flow<T> {
        if (showSpinner) {
            _spinner.value = true
        }
        return block.asFlow()
    }

定义变换函数：

    fun <T, R> Flow<T>.then(block: suspend (T) -> R): Flow<R> {
        return this.flatMapConcat { flow { block(it) } }
    }

定义订阅函数：


 protected val uiScope = CoroutineScope(Dispatchers.main
                    + viewModelJob
                    + CoroutineExceptionHandler { _, exception ->
                		Logger.e("caught original $exception")})
                		
 fun <T> Flow<T>.result(success: (T) -> Unit,
                        error: (Throwable) -> Unit = { _errorToast.value = it.message }) {
        uiScope.launch {
            this@result.flowOn(Dispatchers.io)
                    .catch { cause ->
                        val err = cause as? AccessThrowable
                        when (err?.code) {
                            ERROR.ERROR_NETWORK -> {
                                _networkError.postValue(err)
                            }
                            ERROR.ERROR_TOKEN_EXPIRED -> {
                                _reLogin.postValue(true)
                            }
                        }
                        if (err == null) {
                            error(AccessThrowable(ERROR.UNKNOWN, cause.message ?: ""))
                        } else {
                            error(err)
                        }
                    }.onCompletion {
                        if (_spinner.value == true) {
                            _spinner.value = false
                        }
                    }.collect {
                        success.invoke(it)
                    }
        }
    }

如何使用？

fun function(...parameters...) {
        loadAsync {
        	//suspend function，which returns bool
            repo.function1(...parameters...)
        }.then {
            if (it) {
                //suspend function
                repo.function2(...parameters..)
            } else throw AccessThrowable(ERROR.UNKNOWN, "失败")
        }.result({
            _value.postSuccess(it)
        }, {
            _value.postError(it)
        })
    }

由于Flow属于实验性质的，暂时用Rx代替，这需要使用coroutines的Rx扩展包

定义启动flow的函数：

                		
    protected val _spinner: MutableLiveData<Boolean> by lazy { MutableLiveData<Boolean>() }
    val spinner: LiveData<Boolean>
        get() = _spinner
        
    fun <T : Any> loadAsync(showSpinner: Boolean = true,
                            block: suspend CoroutineScope.() -> T): Single<T> {
        if (showSpinner) {
            _spinner.value = true
        }
        return rxSingle(block = block)
    }

定义变换函数：

    fun <T : Any, R : Any> Single<T>.then(block: suspend CoroutineScope.(T) -> R): Single<R> {
        return flatMap {
            rxSingle { block(it) }
        }
    }

定义订阅函数：

    fun <T> Single<T>.result(success: (T) -> Unit,
                           error: (AccessThrowable) -> Unit = { _errorToast.value = it.message }) {
        disposable = subscribeOn(schedulers.io)
                .observeOn(schedulers.main)
                .subscribe({
                    if (_spinner.value == true) {
                        _spinner.value = false
                    }
                    success.invoke(it)
                },{
                    if (_spinner.value == true) {
                        _spinner.value = false
                    }
                    (it as? AccessThrowable)?.run { error.invoke(this) }
                })
    }