Swift将Future与多个值结合起来？

Question

我可能会以错误的方式进行此操作，但我有一个函数，我想随着时间的推移发出多个值。但我不希望它在订阅该对象之前开始发射。我是从 RxSwift 开始合并的，所以我基本上是在尝试在 RxSwift 世界中复制 Observable.create()。我发现最接近的是返回一个 Future，但 Futures 只会成功或失败（所以它们基本上就像 RxSwift 中的 Single。）

我在这里缺少一些基本的东西吗？我的最终目标是制作一个函数来处理视频文件并发出进度事件直到它完成，然后发出完成文件的 URL。

Answer 1

通常您可以使用PassthroughSubject 发布自定义输出。您可以在自己的Publisher 实现中包装一个PassthroughSubject（或多个PassthroughSubjects），以确保只有您的进程可以通过主题发送事件。

让我们模拟一个VideoFrame 类型和一些输入帧以作为示例：

typealias VideoFrame = String
let inputFrames: [VideoFrame] = ["a", "b", "c"]

现在我们要编写一个同步处理这些帧的函数。我们的函数应该以某种方式报告进度，最后，它应该返回输出帧。为了报告进度，我们的函数将采用PassthroughSubject<Double, Never>，并将其进度（作为从 0 到 1 的分数）发送给主题：

func process(_ inputFrames: [VideoFrame], progress: PassthroughSubject<Double, Never>) -> [VideoFrame] {
    var outputFrames: [VideoFrame] = []
    for input in inputFrames {
        progress.send(Double(outputFrames.count) / Double(inputFrames.count))
        outputFrames.append("output for \(input)")
    }
    return outputFrames
}

好的，现在我们想把它变成一个出版商。发布者需要输出进度和最终结果。所以我们将使用这个enum 作为它的输出：

public enum ProgressEvent<Value> {
    case progress(Double)
    case done(Value)
}

现在我们可以定义我们的Publisher 类型。我们称它为SyncPublisher，因为当它接收到Subscriber 时，它会立即（同步）执行其整个计算。

public struct SyncPublisher<Value>: Publisher {
    public init(_ run: @escaping (PassthroughSubject<Double, Never>) throws -> Value) {
        self.run = run
    }

    public var run: (PassthroughSubject<Double, Never>) throws -> Value

    public typealias Output = ProgressEvent<Value>
    public typealias Failure = Error

    public func receive<Downstream: Subscriber>(subscriber: Downstream) where Downstream.Input == Output, Downstream.Failure == Failure {
        let progressSubject = PassthroughSubject<Double, Never>()
        let doneSubject = PassthroughSubject<ProgressEvent<Value>, Error>()
        progressSubject
            .setFailureType(to: Error.self)
            .map { ProgressEvent<Value>.progress($0) }
            .append(doneSubject)
            .subscribe(subscriber)
        do {
            let value = try run(progressSubject)
            progressSubject.send(completion: .finished)
            doneSubject.send(.done(value))
            doneSubject.send(completion: .finished)
        } catch {
            progressSubject.send(completion: .finished)
            doneSubject.send(completion: .failure(error))
        }
    }
}

现在我们可以像这样将process(_:progress:) 函数变成SyncPublisher：

let inputFrames: [VideoFrame] = ["a", "b", "c"]
let pub = SyncPublisher<[VideoFrame]> { process(inputFrames, progress: $0) }

run 闭包是 { process(inputFrames, progress: $0) }。请记住，这里的$0 是PassthroughSubject<Double, Never>，这正是process(_:progress:) 想要的第二个参数。

当我们订阅这个pub时，它会首先创建两个主题。一个主题是进度主题并被传递给闭包。我们将使用另一个主题来发布最终结果和 .finished 完成，或者如果 run 闭包引发错误，则仅发布 .failure 完成。

我们使用两个单独的主题的原因是因为它可以确保我们的发布商表现良好。如果run 闭包正常返回，则发布者发布零个或多个进度报告，然后是单个结果，然后是.finished。如果run 闭包抛出错误，则发布者发布零个或多个进度报告，然后发布.failed。 run 闭包无法让发布者发出多个结果，或在发出结果后发出更多进度报告。

最后我们可以订阅pub看看是否正常：

pub
    .sink(
        receiveCompletion: { print("completion: \($0)") },
        receiveValue: { print("output: \($0)") })

这是输出：

output: progress(0.0)
output: progress(0.3333333333333333)
output: progress(0.6666666666666666)
output: done(["output for a", "output for b", "output for c"])
completion: finished