Download list of files with retrofit and rxJava

Question

I´m trying to download a list of files from an api with retrofit and rxjava i read a lot but i can't figurated how retrofit and rxjava work together, and im not sure what kind of thread must use on the observer

I´m trying changing the file download process of an application originally was made with doAsync (Anko), The application run in a very limited tablet with api 19 and slow internet connection. The inicial problema was thath the tablet expermient leak of memory in the first files download (about 200 files, using 100 mb of memory (all them)). So after a long research i decide to try rewrite the code for the download process using Rx and retrofit (but i nos have much experince in kotlin or movil).

my Model

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)

My service

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

call for get the list to download and call the download process

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




    }

call for the download function

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

    }

When i run the code the app download some files (about five) and then do not do anything more (The logcat send me a lot of GC_FOR_ALLOC warinings and frame skip). I tried limit the calls to the download method (programmatically with a count variable (limit to 10), for avoid to try download the 200 files togheter) and works but still sending GC_FOR_ALLOC and i have to add some code to download the other 190 files in batches of ten (but i think this is not the best approach). im see thath i has to use I list of observers and call the downlad function with flatMap as another obsever. But i not understand how to work with this approach.

I try to use another thread (a new one for each download) but i have an error with the thread pool

Answer 1

I'd look at converting your code to use Observable.fromIterable and flatMap, flatMap has an overload with a maxConcurrency parameter you can use to avoid creating the unlimited number of threads that Schedulers.io() will do by default.

Before your subscribe containing the foreach loop add a call to flatMap as shown below. Your subscribe will then get called with the result of each call to getArchivo, so replace it's contents with the code you currently have in your getArchivo subscribe

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