发布者发布运营进度和最终价值

Question

鉴于我有一个提供以下功能的 SDK

class SDK {
    static func upload(completion: @escaping (Result<String, Error>) -> Void) {
        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
            completion(.success("my_value"))
        }
    }
}

我能够创建一个包装器以使其使用更实用

class CombineSDK {
    func upload() -> AnyPublisher<String, Error> {
        Future { promise in
            SDK.upload { result in
                switch result {
                case .success(let key):
                    promise(.success(key))
                case .failure(let error):
                    promise(.failure(error))
                }
            }
        }.eraseToAnyPublisher()
    }
}

现在我试图了解如果 SDK 上传方法还提供如下所示的进度块，我的 JointSDK.upload 方法应该是什么样子：

class SDK {
    static func upload(progress: @escaping (Double) -> Void, completion: @escaping (Result<String, Error>) -> Void) {
        DispatchQueue.main.asyncAfter(deadline: .now() + 0.5) {
            progress(0.5)
        }

        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
            progress(1)
        }

        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
            completion(.success("s3Key"))
        }
    }
}

Answer 1

我们需要Output为您的发布商提供一个代表进度或最终值的类型。所以我们应该使用一个enum. 由于 Foundation 框架已经定义了一个名为 的类型Progress，因此我们将为我们的类型命名Progressable以避免名称冲突。我们不妨将其通用化：

enum Progressable<Value> {
    case progress(Double)
    case value(Value)
}

现在我们需要考虑发布者应该如何行事。典型的发布商（例如）在获得订阅之前URLSession.DataTaskPublisher不会执行任何操作，并且它会为每个订阅重新开始工作。retry仅当上游发布者的行为如此时，该运算符才起作用。

所以我们的发布商也应该这样做：

extension SDK {
    static func uploadPublisher() -> UploadPublisher {
        return UploadPublisher()
    }

    struct UploadPublisher: Publisher {
        typealias Output = Progressable<String>
        typealias Failure = Error

        func receive<S>(subscriber: S) where S : Subscriber, Self.Failure == S.Failure, Self.Output == S.Input {
            <#code#>
        }
    }
}

创建发布者（通过调用SDK.uploadPublisher()）不会开始任何工作。我们将替换<#code#>为开始上传的代码：

extension SDK {
    static func uploadPublisher() -> UploadPublisher {
        return UploadPublisher()
    }

    struct UploadPublisher: Publisher {
        typealias Output = Progressable<String>
        typealias Failure = Error

        func receive<S>(subscriber: S) where S : Subscriber, Self.Failure == S.Failure, Self.Output == S.Input {
            let subject = PassthroughSubject<Output, Failure>()
            subject.receive(subscriber: subscriber)

           upload(
                progress: { subject.send(.progress($0)) },
                completion: {
                    switch $0 {
                    case .success(let value):
                        subject.send(.value(value))
                        subject.send(completion: .finished)
                    case .failure(let error):
                        subject.send(completion: .failure(error))
                    }
                }
            )
        }
    }
}

请注意，我们在开始上传subject.receive(subscriber: subscriber) 之前调用。这个很重要！如果upload在返回之前同步调用其回调之一会怎样？通过在调用上传之前将订阅者传递给主题，我们确保即使upload同步调用其回调，订阅者也有机会收到通知。