kotlin - 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")
    }

错误出现在 switchmap 上。我非常熟悉 RxJS 中的运算符，但我在语法上遇到了困难。

非常感谢任何帮助。

Answer 1

您正在尝试将 Flowable switchMap 转换为 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) }