swift - RxSwift - 与 Singleton 不同类型的队列可观察值

Question

我有一个带有一些公共(public)可用函数的 Singleton 类(实际上它将进行网络抓取，但这里进行了简化)。这些函数都返回 Single<T>但类型不同。

它可能看起来像这样:

class Singleton {
    static let shared = Singleton()
    private init() { }

    func doSomethingInt() -> Single<Int> {
        return Single.just(1)
            .delay(3, scheduler: MainScheduler.instance)
    }

    func doSomethingString() -> Single<String> {
        return Single.just("Wow")
            .delay(3, scheduler: MainScheduler.instance)
    }
}

当有人打电话Singleton.shared.doSomthingInt()时，该函数应该放置在一个队列中，直到它通过队列才执行。队列中的下一个可观察量不应在所有事情完成之前开始执行。理想的Singleton将有一个函数会延迟传递给它的每个函数的执行。像这样的事情:

private func placeInQueue<T: Any>(operation: Single<T>) -> Single<T> {
    // place in some magic shared queue
    return operation
}

然后我可以将此函数链接到应放置在队列中的函数的开头，如下所示:

func doSomethingString() -> Single<String> {
    let operation = Single.just("Wow")
        .delay(3, scheduler: MainScheduler.instance)
    return placeInQueue(operation)
}

我觉得这应该可以通过 concat 实现不知怎的操作，但我还没有能够解决它。

有任何线索吗？

Answer 1

我已经创建了这个类，它似乎正在工作:)

也许这是运行它的调度程序的问题。至少，如果我向队列添加一些操作，每个操作在 MainScheduler 上有 3 秒的延迟，我可以看到一些排队的操作最终完成了或多或少的 3,5~4上一个操作完成后的秒数。不过，这对我来说并不是什么大问题:)

class ObservableQueue {
    init() { }

    private var queueArray = [(operation: Observable<Void>, id: Double)]()

    /// Adding the `operation` to an internal queue. Starts execution of the `operation` when all previous operations in the queue had sendt an stop event.
    func placeInQueue<T: Any>(_ operation: Single<T>) -> Single<T> {
        let operationId = createId()
        let queuedOperation = currentQueue()
            .flatMap { _ -> Single<T> in
                return operation
            }
            .do(
                onNext: { [weak self] _ in self?.removeFromQueue(id: operationId) },
                onError: { [weak self] _ in self?.removeFromQueue(id: operationId)
            })
        let queueableOperation = operation
            .map { _ in return () }
            .asObservable()
            .catchErrorJustReturn(())
        addToQueue(queueableOperation, id: operationId)
        return queuedOperation
    }

    private func createId() -> Double {
        var operationId: Double = Date().timeIntervalSince1970
        while (queueArray.contains { $0.id == operationId }) {
            operationId = Date().timeIntervalSince1970
        }
        return operationId
    }

    private func currentQueue() -> Single<Void> {
        var queue = queueArray.map{ $0.operation }
        if queue.isEmpty {
            queue = [Observable.just(())]
        }
        return Observable.concat(queue).takeLast(1).asSingle()
    }

    private func addToQueue(_ operation: Observable<Void>, id: Double) {
        queueArray.append((operation: operation, id: id))
    }

    private func removeFromQueue(id: Double) {
        guard let index = (queueArray.index { $0.id == id }) else { return }
        queueArray.remove(at: index)
    }
}

这是我的使用方法:

private let queue = ObservableQueue()

func doSomethingInt() -> Single<Int> {
    let operation = Single.just(1)
        .delay(3, scheduler: MainScheduler.instance)
    return queue.placeInQueue(operation)
}

我希望这对某人有帮助:)请随时提出对此解决方案的疑虑，或者如果您有任何更好的解决方案。