Скачать список файлов с модификацией и rxJava

Question

Я пытаюсь загрузить список файлов из API с Retrofit и RXJAVA, я много читаю, но я не могу понять, как Retrofit и RXJAVA работают вместе, и я не уверен, какой поток должен использовать на наблюдателе

Я пытаюсь изменить процесс загрузки файла приложения, изначально созданного с помощью doAsync (Anko). Приложение работает на очень ограниченном планшете с api 19 и медленным интернет-соединением.Первоначальная проблема заключалась в том, что планшет испытывал утечку памяти при первой загрузке файлов (около 200 файлов, используя 100 МБ памяти (все они)).Поэтому после долгих исследований я решил попробовать переписать код для процесса загрузки с использованием Rx и модернизации (но у меня нет большого опыта в kotlin или movil).

моя модель

data class File(
         val id: String
)      //staging

data class Movimientos(val borrar: List<File>,
                       val descargar: List<File>,
                       val estructura: List<Nodo>)

data class Nodo( val id: String,
                 val root: String,
                 val name: String,
                 val file: Boolean,
                 val descargado: Boolean,
                 val version: Int)

Мой сервис

//return a json with a list of files to download (and to more list -not important for now-)
 @GET("index.php")
    fun getMovimientos(@Query("r") r: String = "movimientos",
                       @Query("imei") id: String,
                       @Query("access-token") accesstoken: String): Observable<Movimientos>


    @Streaming
    @GET("index.php")
    fun getArchivo(@Query("r") r: String = "descarga",
                    @Query("id") fileID: String,
                    @Query("access-token") accesstoken: String): Observable<ResponseBody>


    companion object {
        fun create(): WikiApiService {

            val retrofit = Retrofit.Builder()
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                    .addConverterFactory(GsonConverterFactory.create())
                    .baseUrl("https://url/api.biblioteca/")
                    .build()

            return retrofit.create(WikiApiService::class.java)
        }
    }

вызов для получения списка для загрузки и вызов процесса загрузки

private fun beginSearch(searchString: String) {
        disposable = wikiApiServe.getMovimientos("movimientos", "d55a374eebc242a5", "XXXXXXXXXXXX")
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())

                .subscribe(
                        {
                            result -> txt_search_result.text = "${result.estructura.size} tamaño estructura"
                            Log.d("Estructura: ", "Es: ${result.estructura}")
                            //var aux =10

                            result.estructura.forEach {
                                nodo ->
                                if (nodo.file/) {
                                    //aux--
                                    descargarArchivo(nodo.id)

                                }
                            }
                        },
                        { error -> Toast.makeText(this, error.message, Toast.LENGTH_SHORT).show() }
                )




    }

вызовдля функции загрузки

private fun descargarArchivo(id: String){
        disposable = wikiApiServe.getArchivo("descarga", id, "XXXXXXXXXXX")
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.newThread())
                .subscribe(
                        {
                            //result -> txt_search_result.text = "${result.estructura.size} tamaño estructura"
                            totalDescargas = totalDescargas +1
                            Log.d("Descargado: ", "Id archivo: $id bytes: ${it.bytes().size} Total descargas: $totalDescargas")

                            //total.text = totalDescargas
                        },
                        { error -> Toast.makeText(this, error.message, Toast.LENGTH_SHORT).show() }
                )

    }

Когда я запускаю код, приложение загружает некоторые файлы (около пяти), а затем больше ничего не делает (logcat посылает мне много предупреждений GC_FOR_ALLOC и пропуска кадров).Я попытался ограничить вызовы метода загрузки (программно с переменной count (ограничение до 10), чтобы избежать попытки загрузить 200 файлов вместе) и работает, но все еще отправляет GC_FOR_ALLOC, и мне нужно добавить некоторый код для загрузки других 190 файловпартиями по десять (но я думаю, что это не лучший подход).я вижу, что мне нужно использовать список наблюдателей и вызвать функцию downlad с flatMap в качестве другого наблюдателя.Но я не понимаю, как работать с этим подходом.

Я пытаюсь использовать другой поток (новый для каждой загрузки), но у меня есть ошибка с пулом потоков

Answer 1

Я бы посмотрел на преобразование вашего кода для использования Observable.fromIterable и flatMap, flatMap имеет перегрузку с параметром maxConcurrency, который вы можете использовать, чтобы избежать создания неограниченного количества потоков, которые Schedulers.io () будет делать по умолчанию.

Перед подпиской, содержащей цикл foreach, добавьте вызов flatMap, как показано ниже. После этого ваша подписка будет вызываться с результатом каждого вызова getArchivo, поэтому замените ее содержимое кодом, который у вас есть в вашей подписке getArchivo

private fun beginSearch(searchString: String) {
    disposable = wikiApiServe.getMovimientos("movimientos", "d55a374eebc242a5", "XXXXXXXXXXXX")
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .flatMap { result ->
            Observable.fromIterable(result.estructura)
                .flatMap({ nodo ->
                    wikiApiServe.getArchivo("descarga", nodo.id, "XXXXXXXXXXX")
                    .subscribeOn(Schedulers.io())
            }, 4)
        }
        .subscribe(
            {
            totalDescargas = totalDescargas +1
            Log.d("Descargado: ", "Id archivo: $id bytes: ${it.bytes().size} Total descargas: $totalDescargas")
            },
            { error -> Toast.makeText(this, error.message, Toast.LENGTH_SHORT).show() }