房间内将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给出有限的数目或项，然后结束。

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