Rxjava2-如何在可观察序列完成之前立即发出onNext()

发布它并在最后一个值中连接!

observable1(productId)
.publish( { shared ->
     Observable.concatArrayEager(
         shared.flatMap({
             val limit = 8 - it.size
               observable2(productId,  limit)
           }, { p1, p2 ->
               p1.addAll(p2);
           }),
         shared.takeLast(1)
     )
})

我不敢相信这是多么容易。我用了startWith,它发出了第一次观测的结果。一秒钟后,最终结果出来了。看起来是这样的:

override fun getProductObservable(): Observable<List<ProductModel>> {

return observable1(productId)
               .flatMap({
                 val limit = 8 - it.size
                   observable2(productId,  limit)
                    .startWith(Observable.just(emptyList<ProductModel>()))
               }, { p1, p2 ->
                   p1.addAll(p2);  p1 //simply return observables1's results already packed with p2 results
               })