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

Question

我可能会以错误的方式处理这个问题，但我有一个函数，我想用它随着时间的推移发出多个值。但我不希望它在订阅该对象之前开始发出。I\xe2\x80\x99m 从 RxSwift 合并而来，所以我\xe2\x80\x99m 基本上试图在 RxSwift 世界中复制 Observable.create() 。我发现的最接近的是返回一个 Future，但 future 只会成功或失败（所以它们基本上就像 RxSwift 中的 Single 一样。）

\n\n

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

\n

Answer 1

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

VideoFrame为了示例目的，让我们模拟一个类型和一些输入框架：

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

现在我们要编写一个同步处理这些帧的函数。我们的函数应该以某种方式报告进度，最后，它应该返回输出帧。为了报告进度，我们的函数将采用 a 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，因为当它收到 a 时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，或者.failure如果run闭包抛出错误则仅发布完成。

我们使用两个单独的主题的原因是因为它可以确保我们的发布者表现良好。如果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