RxJava2 Kotlin SwitchMap 从 RealmResults 到 Observable

Question

好的，这是我在 RxJava 中的第一个问题，所以请温柔一点。

我正在查询 Realm 中的现有用户，将 RealmResults 列表作为可流返回，然后我想创建一个新用户或返回现有用户，然后转换为 JSON。

这是我目前所拥有的。我有点卡住了。

 fun getUsers(realm: Realm): Flowable<RealmResults<User>> {
     return when (realm.isAutoRefresh) {
         true -> realm.where<User>().findAllAsync().asFlowable().filter(RealmResults<User>::isLoaded)
         false -> Flowable.just(realm.where<User>().findAll())
     }
 }

fun checkNewUserRequired(realm: Realm, results: RealmResults<User>): Observable<String> {
    if (results.isEmpty()) {
       //not complete, I will create a new user here
       return Observable.just("Dummy")
    } else {
       val user = realm.where<User>().findFirst()!!
       val detachedUser = realm.copyFromRealm(user)
       return Observable.just(userToJsonString(realm, detachedUser))
    }
}

val getNewUser= getUsers(realm)
    .take(1)
    .switchMap{ results -> checkNewUserRequired(realm, results) }
    .subscribe{
        //log result
        result : String -> Log.d(TAG, "JSON OUTPUT: $result")
    }

错误在开关图上。我对 RxJS 中的运算符非常熟悉，但我在语法上遇到了困难。

非常感谢任何帮助。

Answer 1

您正在尝试将switchMap 和Flowable 转换为Observable，它们实际上是不同的类型。您需要从一种类型转换为另一种类型。

在您的情况下，最简单的解决方案是，将checkNewUserRequired 转换为Flowable

，因为看起来您不会遇到与背压 相关的任何问题

示例

fun checkNewUserRequired(realm: Realm, results: RealmResults<User>): Flowable<String> = Flowable.just(
    if (results.isEmpty()) "Dummy"
    else {
        val user = realm.where<User>().findFirst()!!
        val detachedUser = realm.copyFromRealm(user)
        userToJsonString(realm, detachedUser)
    }
)

您也可以使用函数toFlowable 将现有的Observable 转换为Flowable，但是您需要指定BackpressureStrategy。

示例

.switchMap{ results -> checkNewUserRequired(realm, results).toFlowable(BackpressureStrategy.DROP) }