android - 使用 collectInto 从 RxKotlin/RxJava 中的两个可观察源构建列表

Question

我有一个 Category数据类和一个 Plan数据类。每个Category有一个计划 ID 列表。通过 Room 存储了类别和计划。我正在尝试构建本地 List<Any>我将每个类别添加到列表中，然后添加每个类别的计划。

因此对于每个类别，将类别添加到列表中，然后添加属于该类别的每个计划。

最终结果看起来像这样......

0 -> a Category
1 -> a Plan
2 -> a Plan
3 -> a Plan
4 -> a Category
5 -> a Plan

等等

以下调用成功返回 Observable<List<Category>>和一个 Observable<Plan>

AppDatabase
   .getDatabase(context)
   .categoryDao()
   .getAll()

AppDatabase.getDatabase(context).planDao().getPlan(planId)

我在这里尝试构建我的列表，但实际上当我订阅它时它从未发出。没有完成，或错误。流中的所有其他内容都会受到打击。为什么我得不到最终结果？

    fun fetchCategoriesAndPlans() {
    val items = mutableListOf<Any>()
    AppDatabase
        .getDatabase(context)
        .categoryDao()
        .getAll()
        .concatMap { listOfCategories ->
            listOfCategories.toObservable()
        }
        .doOnNext { category ->
            items.add(category)
        }
        .concatMap { category ->
            category.getPlanIds()!!.toObservable()
        }
        .flatMap { planId ->
            AppDatabase.getDatabase(context).planDao().getPlan(planId)
        }.collectInto(items, BiConsumer{ list, i ->
            Log.d(TAG, "Collect into")
            list.add(i)
        })
        .subscribeBy(
            onSuccess = {
                Log.d(TAG, "Got the list")
            },
            onError = {
                Log.e(TAG, "Couldn't build list ${it.message}", it)
            })
}

Answer 1

我根据您的情况制作了一个演示，它有助于发出 Category 和 Plan

override fun onCreate(savedInstanceState: Bundle?) {
    ...

    getCategories()
        .flattenAsObservable { it }
        .flatMap { getPlanWithCategory(it) }
        .toList()
        .subscribe({
            for (item in it) {
                Log.i("TAG", " " + item.javaClass.canonicalName)
            }
        }, {

        })
}

fun getPlanWithCategory(category: Category): Observable<Any> {
    val getPlansObservable = Observable.fromArray(category.planIds).flatMapIterable {
        it
    }.flatMap {
        getPlan(it).toObservable()
    }
    return Observable.concat(Observable.just(category), getPlansObservable)
}


fun getPlan(planId: String): Single<Plan> {
    return Single.just(Plan())
}

fun getCategories(): Single<List<Category>> {
    val categories = arrayListOf<Category>()
    categories.add(Category(arrayListOf("1", "2", "3")))
    categories.add(Category(arrayListOf("1", "2")))
    return Single.just(categories)
}

class Category(val planIds: List<String>)

class Plan

输出

 I/TAG:  Category
 I/TAG:  Plan
 I/TAG:  Plan
 I/TAG:  Category
 I/TAG:  Plan
 I/TAG:  Plan

希望对你有帮助