在房间中将 Flowable<List<Obj1>> 更改为 Flowable<List<Obj2>>

Question

如何从房间中读取可流动的值列表并将其转换为另一个对象，该对象是房间中更多值的组合

database.leadsDao().getLeads(leadState.name)
    .flatMap {
        val len = it.size.toLong()
        Flowable.fromIterable(it)
            .flatMap {
                Flowable.zip(
                        database.orderDao().getById(it.orderId),
                        database.orderMedicineDao().getByOrderId(it.orderId),
                        database.patientDao().getById(it.patientId),
                        Function3<Order, List<OrderMedicine>, Patient, LeadDetail>
                        { order, orderMedicines, patient -> LeadDetail.from(it, patient, order, orderMedicines) })
            }
            .take(len)
            .toList()
            .toFlowable()
    }

上面的代码有效，但我不喜欢take(len) 部分。没有它，流永远不会调用订阅者的 onNext。流一直在等待更多的项目，这不应该发生，因为 Flowable.fromIterable 给出了有限的数量或项目然后结束。即，下面的代码不起作用

database.leadsDao().getLeads(leadState.name)
    .flatMap {
        Flowable.fromIterable(it)
            .flatMap {
                Flowable.zip(
                        database.orderDao().getById(it.orderId),
                        database.orderMedicineDao().getByOrderId(it.orderId),
                        database.patientDao().getById(it.patientId),
                        Function3<Order, List<OrderMedicine>, Patient, LeadDetail>
                        { order, orderMedicines, patient -> LeadDetail.from(it, patient, order, orderMedicines) })
            }
            .toList()
            .toFlowable()
    }

Answer 1

Flowable.fromIterable 给出有限数量或项目然后结束。

但是 flatmap 内部的 Flowable.zip 不会结束，因为 Room 的 DAO 对象会发出当前值和所有未来的更新，所以压缩在一起的 database.*() 调用不是有限的。如果您将.first() 调用添加到内部Flowable.zip，则第二个版本也应该可以工作。