ios - 如何在 RxSwift 中等待可变数组中的所有可观察对象完成

Question

我的期望是动态添加可观察对象(例如:图像上传)，让它们启动，然后，当我完成动态地将所有内容排队时，等待所有可观察对象完成。

这是我的类(class):

open class InstantObservables<T> {
    lazy var disposeBag = DisposeBag()

    public init() { }

    lazy var observables: [Observable<T>] = []
    lazy var disposables: [Disposable] = []

    open func enqueue(observable: Observable<T>) {
        observables.append(observable)

        let disposable = observable
            .subscribe()

        disposables.append(disposable)

        disposable
            .addDisposableTo(disposeBag)
    }

    open func removeAndStop(atIndex index: Int) {
        guard observables.indices.contains(index)
            && disposables.indices.contains(index) else {
            return
        }
        let disposable = disposables.remove(at: index)
        disposable.dispose()

        _ = observables.remove(at: index)
    }

    open func waitForAllObservablesToBeFinished() -> Observable<[T]> {
        let multipleObservable = Observable.zip(observables)
        observables.removeAll()
        disposables.removeAll()
        return multipleObservable
    }

    open func cancelObservables() {
        disposeBag = DisposeBag()
    }
}

但是当我订阅 waitForAllObservablesToBeFinished() 发送的可观察量时，所有这些都被重新执行(这是关于 Rx 如何工作的逻辑)。

无论订阅数量是多少，我如何保证每个都执行一次？

Answer 1

在写问题的时候，我得到了答案!通过通过 shareReplay(1) 更改 observable，然后排队并订阅这个更改后的 observable。它有效!

这是更新后的代码:

open class InstantObservables<T> {
    lazy var disposeBag = DisposeBag()

    public init() { }

    lazy var observables: [Observable<T>] = []
    lazy var disposables: [Disposable] = []

    open func enqueue(observable: Observable<T>) {
        let shared = observable.shareReplay(1)
        observables.append(shared)

        let disposable = shared
            .subscribe()

        disposables.append(disposable)

        disposable
            .addDisposableTo(disposeBag)
    }

    open func removeAndStop(atIndex index: Int) {
        guard observables.indices.contains(index)
            && disposables.indices.contains(index) else {
            return
        }
        let disposable = disposables.remove(at: index)
        disposable.dispose()

        _ = observables.remove(at: index)
    }

    open func waitForAllObservablesToBeFinished() -> Observable<[T]> {
        let multipleObservable = Observable.zip(observables)
        observables.removeAll()
        disposables.removeAll()
        return multipleObservable
    }

    open func cancelObservables() {
        disposeBag = DisposeBag()
    }
}