rx-java - 如何使用RxJava协调可完成执行的列表？

Question

我有一个想要顺序运行的BlockingTasks列表:

class BlockingTaskManager {
    val tasks = mutableListOf<BlockingTask>()

    fun run () : Completable {
        /* What can I put here in order to run
           all tasks and return a completable according with the
           requirements described below? */
    }
}

interface BlockingTask {

    fun run () : Completable

    fun getDescription(): String

    fun blockPipe(): Boolean
}

要求

如果管道中的任何Completable以错误结束，则整个执行应停止。

当Completable返回的run()完成(错误或成功)时，应发出包含时间戳记和getDescription()返回的字符串的日志；

我考虑过使用 Completable.concat { tasks }，但是我看不到如何调用 currentIterationTask.run()，以及如果 run()导致错误以及如何记录 currentInterationTask.getDescription()，如何使管道失败。

另一个想法是使用 andThen()，但是我不确定它是否可以根据需要工作:

    tasks.forEachIndexed {
        idx, task ->
        var isBroken = false
        task.run()
                .doOnComplete {
                    logTaskCompletedSuccessfully(task)
                }
                .doOnError {
                    error ->
                    logTaskFailed(task, error)
                    isBroken = task.blockPipe()
                }
                .andThen {
                    if (!isBroken && idx < tasks.size) {
                        tasks[idx+1]
                    }
                }
    }

Answer 1

您可以通过多种方式进行操作:

 fun run(): Completable {
    return Completable.concat(tasks.map { task ->
        task.run()
                .doOnComplete { logTaskCompletedSuccessfully(task) }
                .doOnError { error ->
                    logTaskFailed(task, error)
                }
    })
}

顺序执行是由 concat运算符实现的，某些 Completable中的错误将使 onError()停止流，并且您将获得成功或错误的日志记录功能。

至于 blockPipe()标志，如果我理解正确，它用于标志任务已失败，应该中断流，如果是这样，对我来说似乎不必要，对于任务中的任何失败，请抛出 Exception而不是引发标记，则异常将使用 onError中断流。

另一种选择是使用更具响应性的方法，而不是迭代列表中的任务，而使用 Observable对其进行迭代。
以 Observable开头，该任务会迭代任务列表，然后将每个任务 flatMap迭代为 Completable。
由于您没有对每个 Completable应用任何Scheduler，因此可以实现顺序执行，因此可以保持执行顺序。

 fun run(): Completable {
    return Observable.fromIterable(tasks)
            .flatMapCompletable { task ->
                task.run()
                        .doOnComplete { logTaskCompletedSuccessfully(task) }
                        .doOnError { error ->
                            logTaskFailed(task, error)
                        }
            }
            .subscribeOn(Schedulers.newThread() // Schedulers.io())
}