android - RXKotlin 在 doOnNext 内部中断并调用另一个函数

Question

我刚开始使用 rx kotlin，还没有完全理解。我正在尝试遍历查询列表，并一个一个地执行它们。在这个列表中，我有一个特殊的字符串，一旦到达，我想打破循环并执行另一个功能

我怎样才能在下面的例子中做到这一点？

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

使用 takeWhile 从 doOnNext 中分解出你想要中断的部分:

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