RXKotlin Break Inside doOnNext and Call Another Function

Question

i am using rx kotlin newly and didn't understand all of it yet. I am trying to loop over a list of queries, and execute them one by one. in this list i have a special string that once reached, i want to break the loop and perform another function

how can i do this in the below example?

fun runQueries() {

     Observable.fromIterable(queriesTemp)

                    .subscribeOn(Schedulers.computation())
                    .doOnNext { query ->
                        if (query.contains("COMPLETION OF SDF QUERIES")) {
                            if (loginStatus == StaticVariables.FT_CASE_NEW_LOGIN) {
                                tasksQueriesTemp = arrayOfNulls(queries.size - queries.indexOf(query))
                                System.arraycopy(queries, queries.indexOf(query), tasksQueriesTemp, 0, tasksQueriesTemp!!.size)
                            }
                          // break the loop here 
                            runOtherQueries()
                             break
                        }
                        if (!TextUtils.isEmpty(query)) {
                            mDatabase.execSQL(query, false, "")
                        }
                        action(tasksQueriesTemp!!.indexOf(query))
                    }
                    .doOnComplete { executeOtherUpdates(tasksQueriesTemp) }
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe()
}

fun runOtherQueries() {
}

Answer 1

Factor out the part you want to break on from the doOnNext use takeWhile:

val broken = AtomicBoolean();
Observable.fromIterable(queriesTemp)
    .subscribeOn(Schedulers.computation())
    .takeWhile { query ->
        if (query.contains("COMPLETION OF SDF QUERIES")) {
            if (loginStatus == StaticVariables.FT_CASE_NEW_LOGIN) {
                tasksQueriesTemp = arrayOfNulls(queries.size -
                    queries.indexOf(query))
                System.arraycopy(queries, queries.indexOf(query), 
                    tasksQueriesTemp, 0, tasksQueriesTemp!!.size)
            }
            // break the loop here 
            runOtherQueries()
            broken.set(true)
            return@takeWhile false  // whatever the Kotlin syntax is for local returns
        }
        return@takeWhile true
    }
    .doOnNext { query ->

        if (!TextUtils.isEmpty(query)) {
            mDatabase.execSQL(query, false, "")
        }
        action(tasksQueriesTemp!!.indexOf(query))
    }
    .doOnComplete { 
         // if you don't want to execute the other updates if the code
         // in takeWhile has "broken out of the loop"
         if (!broken.get())
             executeOtherUpdates(tasksQueriesTemp) 
    }
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe()