基于另一个 observable 动态终止 Observable | RxSwift

Question

我有一组 Observable，比如 [Observable <WriteTaskResult>]

我想执行所有写任务保持它们的顺序，如果其中任何一个失败，那么我想执行Observable<ResetTaskResult>

以下函数将返回 BatchTasksResult 类型的 observable 用于跟踪任务进度。

示例代码：

enum BatchTasksResult{
    case elapsedTime(Double)
    case failedFatal
    case rolledback
    case success
}

func writeBlocks(tasks: [WriteTask]) -> Observable<BatchTasksResult>{
    return Observable.create {(observable) -> Disposable in
       let allTasks: [Observable<WriteTaskResult>] = self.writeSomewhere(tasks)
       Observable.concat(allTasks)
         .subscribe { writeTaskResult in
               observable.onNext(.elapsedTime(writeTaskResult.totalTime))
            } 
            onError: { (err) in
               // Perform Observable<ResetTaskResult>
               // if ResetTask was successful then observable.onNext(.rolledback)
               // if ResetTask failed then observable.onNext(.failedFatal)
            }
            onCompleted: {
               observable.onNext(.success)
            }
            .disposed(by: disposeBag)
       return Disposables.create()
    }
}

如何使用来自 allTask​​s 的 observable 的 onError 的 Observable 触发回滚逻辑？

简单的解决方案似乎嵌套了 observable，但我猜这不是一个好习惯？我尝试了 FlatMap，但它并不能真正解决 “如果任何单个任务失败，则回滚并重置”还有其他解决方案吗？

Answer 1

无需使用create 函数添加额外的间接级别。每个 Observable 运算符都已经创建了一个新对象。

当你做使用Observable.create时，不要丢弃在外部处理袋中并返回Disposables.create()。只需退回您刚刚创建的一次性用品。

这是做你想做的事情的适当方法：

func writeBlocks(tasks: [WriteTask], resetTask: Single<ResetTaskResult>) -> Observable<BatchTasksResult> {
    // create the array of write tasks and concat them. You seem to have that down.
    let result = Observable.concat(tasks.map(writeSomewhere(task:)).map { $0.asObservable() })
        .share() // the share is needed because you are using the value twice below.
    return Observable.merge(
        // push out the elapsed time for each task.
        result.map { BatchTasksResult.elapsedTime($0.totalTime) },
        // when the last one is done, push out the success event.
        result.takeLast(1).map { _ in BatchTasksResult.success }
    )
    .catch { _ in
        resetTask // the resetTask will get subscribed to if needed.
            .map { _ in BatchTasksResult.rolledback } // if successful emit a rollback
            .catch { _ in Single.just(BatchTasksResult.failedFatal) } // otherwise emit the failure.
            .asObservable()
    }
}

func writeSomewhere(task: WriteTask) -> Single<WriteTaskResult> {
    // create a Single that performs the write and emits a result.
}