如何将多个observable变成一个completable？

Question

我是 RxSwift 新手，刚刚继承了一个旧的代码库，所以如果这是一个愚蠢的问题，请原谅我。

在代码中，数据在 Completables 和 Observables 的帮助下更新。下面是两种方法，大致说明了这是如何完成的（出于隐私目的，有点模糊）：

// note: `getNewData()` returns an Observable
func refreshData() -> Completable {
    dataManager.getNewData()
        .map { DataRepresentation(fromObject: $0) }
        .take(1)
        .asSingle()
        .flatMapCompletable { data in
            self.storageManager.save(data: data)
        }
}

// STORAGE MANAGER

func save(data: DataRepresentation) -> Completable {
    do {
        // PSEUDOCODE: save the data, emit an event about it if necessary.
        return Completable.completed()
    } catch let error {
        return Completable.error(error)
    }
}

所以，我的问题是这样的：让我们假设

getNewData()

允许我传递一些参数来实现它，这样我就不会每次都得到相同的数据。此外，假设我想调用该方法 n 次，等待所有调用返回，然后仍然从

refreshData()

返回一个 Completable （因为不需要更改其签名）。这种事可能吗？我正在研究

.zip

，但我不确定它是否适用于这里。谢谢。

Answer 1

这是最有可能的解决方案。它假设您传入的数据是通过原始数组，并且您将在更新数组后手动调用

refreshData()

...

class Example {
    let dataManager = DataManager()
    let storageManager = StorageManager()

    // example
    var parameters: [Int] = []

    func refreshData() -> Completable {
        Observable.zip(parameters.map(dataManager.getNewData(param:)))
            .map { $0.map(DataRepresentation.init(fromObject:)) }
            .asSingle()
            .flatMapCompletable { Completable.zip($0.map(self.storageManager.save(data:))) }
    }
}

但是，这是一个不合标准的解决方案。它没有充分利用 Rx 所提供的功能。例如，如果您在上述解决方案正在保存数据的过程中调用

refreshData().subscribe()

，您将使程序处于不确定状态。

如果输入数据也包装在 Observable 中，这样你就可以轻松防范此类问题，那就更好了，但这需要更改方法的签名......

class Example {
    let dataManager = DataManager()
    let storageManager = StorageManager()
    let disposeBag = DisposeBag()

    let parameters = PublishSubject<[Int]>()

    init() {
        let newData = parameters
            .flatMapLatest { Observable.zip($0.map(self.dataManager.getNewData(param:))) }

        newData
            .map { $0.map(DataRepresentation.init(fromObject:)) }
            .flatMapLatest { Completable.zip($0.map(self.storageManager.save(data:))) }
            .subscribe(onError: { error in print(error) })
            .disposed(by: disposeBag)
    }
}

通过上述操作，您无需担心是否或何时调用任何函数，也无需担心在先前的写入已经就位时写入数据。您所需要做的就是使用更新的数据调用

parameters.onNext

。 （当然假设幸福的道路。