Tukukkk

I am developing an Android app using Kotlin, RxJava, Retrofit.
I want to send Http Request to the server.

PUT - update option of job

POST - run the job

After the first request success, then I send a second request. So I used concatMap.

val updateJob = restService.updateJob(token, job.id, options) // PUT

val runJob = restService.runJob(token, job.id) // POST



updateJob.concatMap { runJob }

    .subscribeOn(Schedulers.io())

    .observeOn(AndroidSchedulers.mainThread())

    .subscribe({ job ->

        Log.d(TAG, "runJob - success: $job")

    }, {

        Log.e(TAG, "runJob - failed: ${it.message}")

        it.printStackTrace()

    })

But I don't know in case of multiple jobs like below.

I have a job list.
If one job's "update" request is failed, "run" request should not be sent.
But the next job should continue.
To do this, I make a code like below.

    Observable.fromIterable(jobs.toList())

    .concatMap { job ->

        val updateJob = restService.updateJob(token, job.id, job)   // HTTP PUT Request

        val runJob = restService.runJob(token, job.id)  // HTTP POST Request



        updateJob.concatMap { runJob }

    }

    .window(1)

    .subscribeOn(Schedulers.io())

    .observeOn(AndroidSchedulers.mainThread())

    .subscribe({

        it.subscribeOn(Schedulers.io())

            .observeOn(AndroidSchedulers.mainThread())

            .subscribe({

                Log.e(TAG, "run job - success")

            }, {

                Log.e(TAG, "run job - failed - 1: ${it.message}")

                it.printStackTrace()

            })

    }, {

        Log.e(TAG, "run job - failed - 2: ${it.message}")

        it.printStackTrace()

    })

I thought that "window" operator may be the solution.
But it doesn't ...
If some job is failed, the stream is over with onError().
How should I solve this problem?

I resolve this issue using the following code.

Observable.fromIterable(jobs.toList())

    .concatMap { job ->

        val updateJob = restService.updateJob(token, job.id, job)   // HTTP PUT Request

                .onErrorResumeNext(Observable.empty<Job>()) // Solution Point

        val runJob = restService.runJob(token, job.id)  // HTTP POST Request



        updateJob.concatMap { runJob }

    }

    // .window(2)    I removed this line.

    .subscribeOn(Schedulers.io())

    .observeOn(AndroidSchedulers.mainThread())

    .subscribe({

        Log.d(TAG, "run job - success")

    }, {

        Log.e(TAG, "run job - failed - 2: ${it.message}")

        it.printStackTrace()

    })