RxJava использует toList после сбоя flatMap, так как flatMap не завершен

Question

У меня есть три объекта (скажем, A, B, C), и для получения CI нужен B, а для получения A мне нужен B. На экране мне нужно отобразить свойство A вместе со свойством C. Даже еслиЯ могу получить все необходимые данные, так как я использую flatMap, у которого нет onComplete, toList () не выполняется.Вот мой код.

Для каждого a в List мне нужно получить 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()
    }

Примечание. Я задавал похожий вопрос ранее сегодня, но для этого не требовалось использовать flatmap более одного раза.Вы можете просмотреть мое решение по этой ссылке

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