RxJava 在 flatMap 失败后使用 toList 因为 flatMap 不完整

Question

我有三个对象(比如A、B、C(，要获得C，我需要B，要获得A，我需要B.在屏幕上，我需要显示A的属性和C的属性。尽管我可以获得所有必要的数据，但由于我使用没有onComplete的flatMap，toList((不会被执行。这是我的密码。

对于列表中的每个a，我需要得到c，并且我需要返回一个ResultMode类型的列表，其中包括a和c的属性。

override fun methodICall(): LiveData<MutableList<ResultModel>> {
return mySdk
.getAllA() //Returns Flowable<List<A>>
.flatMap { Flowable.fromIterable(it) }
.flatMap { helperMethod(it) }
.toList() // Does not get executed as flatMap isnt completed
.toFlowable()
.onErrorReturn { Collections.emptyList() }
.subscribeOn(Schedulers.io())
.to { LiveDataReactiveStreams.fromPublisher(it) }
}
private fun helperMethod(a:A): Flowable<ResultModel> {
return mySdk
.getB(a.propertyOne!!) // Returns Single<B>
.flatMap { mySdk.getC(it.property!!) } // get C returns Single<C>
.map {
ResultModel(name= a.name,
date = c.date.toString(), 
message = it.messageId!!
)
}.toFlowable()
}

注意：我今天早些时候问了一个类似的问题，但不需要多次使用平面图。你可以在这个链接中查看我的解决方案

RxJava-将列表的结果映射到另一个列表

我的努力(可能是错误的(这是我转换第一种方法的努力(对于第二种方法，我只是将其删除为Flowable并返回single(，但它还有很长的路要走，我认为我走错了路。

override fun methodICall(): LiveData<MutableList<ResultModel>> {
return mySdk
.getAllA()
.concatMapSingle { Flowable.fromIterable(it)
.map { helperMethod(it) }
.toList()
}
.onErrorReturn { Collections.emptyList() }
.subscribeOn(Schedulers.io())
.to { LiveDataReactiveStreams.fromPublisher(it) // here it is single. I think it is because two maps are applied both to helperMethod itself and inside helper method to result model}
}

Answer 1

似乎没有充分的理由不断解构和重构列表。假设没有：

override fun methodICall(): LiveData<MutableList<ResultModel>> {
return mySdk
.getAllA() // Returns Flowable<List<A>>
.flatMapIterable(it)
.concatMapSingle( item => {
mySdk.getB(item.propertyOfA!!)
.flatMap( bItem => mySdk.getC( bItem.propertyOfB!! ) )
.map( ResultModel( name=item.name, message=it.body!! ) )
})
.toList()
.onErrorReturn { Collections.emptyList() }
.subscribeOn(Schedulers.io())
.to { LiveDataReactiveStreams.fromPublisher(it) }
}

因为concatMapSingle()操作符知道每个项，所以在构造ResultModel时可以知道它的名称。现在，你再也不需要经常把事情拆散了。

Answer 2

编辑：我找到了另一个解决方案

override fun methodICall(): LiveData<MutableList<ResultModel>> {
return mySdk
.getAllA()
.concatMapSingle {
Flowable.fromIterable(it)
.flatMap { a ->
mySdk.getB(a.propertyOfA!!)
.flatMap { b -> chatbotSdk.getC(b.propertyOfB!!) }
.map { it ->
ResultModel(name = a.name,
message = it.body!!)
}.toFlowable() 
} .toList()    }
.onErrorReturn { Collections.emptyList() }
.subscribeOn(Schedulers.io())
.to { LiveDataReactiveStreams.fromPublisher(it) }
}

---

原始溶液

这是我的解决方案，但我认为这个解决方案非常混乱，可以改进很多。

data class AtoBDTO(var name: String, var b: Flowable<B>) // I wanted to map one object to more than one so I created this. Probably there is a way to do it with rx functions.
data class BtoCDTO(var name: String, var c: Flowable<C>)

override fun methodICall(): LiveData<MutableList<ResultModel>> {
return mySdk
.getAllA() // Returns Flowable<List<A>>
.concatMapSingle {
Flowable.fromIterable(it)
.map { AtoBDTO(it.name!!,    
mySdk.getB(it.propertyOfA!!).toFlowable()) }  //getB returns Single B
.toList()
}
.concatMapSingle {
Flowable.fromIterable(it)
.map {
BtoCDTO(it.name,
it.b.concatMapSingle { mySdk.getC(it.propertyOfB!!) }) // getC returns Single C
}
.toList()
}
.concatMapSingle {
Flowable.fromIterable(it)
.map {
ResultModel(name = it.name,
message = it.c.blockingFirst().body!!) // I use blocking first because otherwise I can't get rid of flowable
}.toList()
}
.onErrorReturn { Collections.emptyList() }
.subscribeOn(Schedulers.io())
.to { LiveDataReactiveStreams.fromPublisher(it) }
}