android - RxJava 将 Single 与 Completable 结合起来

Question

我很困惑如何正确使用 Single 和 Completable。
我有一个返回 Single 的 get 方法，另一个方法只是保存数据并且不返回任何内容( Completable )
代码如下所示:

fun getUserInfo() : Single<UserInfo>

fun save(token: Token) : Completable

fun initialize() {
    getUserInfo()
        .flatMap { 
            // Get token from UserInfo : val token = userInfor.getToken()
            // Call save(token: Token) method to save data
        }
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeBy(
            onSuccess = {
                // HERE:  Using UserInfo to do somtehing
            },
            onError = {
                Log.d(it)
            }
        )
}  

如何在flatMap block 内编写代码？ (或者是否还有其他适合此用例的运算符？)
谢谢您

Answer 1

您可以使用flatMapCompletable:

fun initialize() {
    getUserInfo()
        .flatMapCompletable { userInfo: UserInfo ->
            // save userInfo somewhere. i.e:
            <a href="https://stackoverflow.com/cdn-cgi/l/email-protection" class="__cf_email__" data-cfemail="3145595842717c4870524558475845481f44425443785f575e" rel="noreferrer noopener nofollow">[email protected]</a> = userInfo
            val token = userInfo.getToken()
            return@flatMapCompletable save(token) 
        }
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeBy(
            onComplete = { // <--------- this becomes onComplete instead of onSuccess
                // Here, you can use userInfo that you saved
            },
            onError = {
                Log.d(it)
            }
        )
}  

请注意，在您的 subscribeBy(...) 方法中，不要忘记将回调函数名称从 onSuccess(...) 更改为 onComplete(...) 因为您正在将 single 转换为 completable。

或者，如果您想将 userInfo 保留为局部变量，您可以使用 toSingleDefault(...) 在您的 完成后将其发送到下游>save(...) 方法已完成执行:

fun initialize() {
    getUserInfo()
        .flatMap { userInfo: UserInfo -> // change back to flatmap
            val token = userInfo.getToken()
            return@flatMap save(token).toSingleDefault(userInfo) 
        }
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeBy(
            onSuccess = { userInfo: UserInfo ->
                // do something with userInfo ...
            },
            onError = {
                Log.d(it)
            }
        )
}