Rxjava2 - как сгенерировать onNext () сразу перед завершением наблюдаемой последовательности

Question

У меня есть две наблюдаемые, которые объединены в плоскую карту следующим образом:

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

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

обратите внимание, как observable2 зависит от результата из observable1 (предельный аргумент). ОБА НАБЛЮДАЮЩИХ ВОЗВРАЩАЙТЕСЬ, ОДИН И ТО ЖЕ ТИП

Теперь на мой вопрос, я хотел бы, чтобы observable1 после завершения немедленно вызывал onNext абонента, но все еще продолжал переходить к плоской карте. По сути, подписчики onNext будут вызываться дважды, это то, что я хочу; один раз , когда observable1 завершается, и t hen снова , когда завершается весь вызов. Как это можно организовать?

Answer 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
               })

Answer 2

Опубликуйте его и объедините в последнем значении!

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