Swift 将 Future 与多个值结合起来? [英] Swift Combine Future with multiple values?

问题描述

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

I may be going about this the wrong way, but I have a function with which I want to emit multiple values over time. But I don’t want it to start emitting until something is subscribed to that object. I’m coming to combine from RxSwift, so I’m basically trying to duplicated Observable.create() in the RxSwift world. The closest I have found is returning a Future, but futures only succeed or fail (so they are basically like a Single in RxSwift.)

我在这里遗漏了一些基本的东西吗?我的最终目标是创建一个函数来处理视频文件并发出进度事件直到它完成，然后为完成的文件发出一个 URL.

Is there some fundamental thing I am missing here? My end goal is to make a function that processes a video file and emits progress events until it completes, then emits a URL for the completed file.

推荐答案

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

Generally you can use a PassthroughSubject to publish custom outputs. You can wrap a PassthroughSubject (or multiple PassthroughSubjects) in your own implementation of Publisher to ensure that only your process can send events through the subject.

为了示例目的，让我们模拟 VideoFrame 类型和一些输入帧:

Let's mock a VideoFrame type and some input frames for example purposes:

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

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

Now we want to write a function that synchronously processes these frames. Our function should report progress somehow, and at the end, it should return the output frames. To report progress, our function will take a PassthroughSubject<Double, Never>, and send its progress (as a fraction from 0 to 1) to the subject:

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 作为它的输出:

Okay, so now we want to turn this into a publisher. The publisher needs to output both progress and a final result. So we'll use this enum as its output:

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

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

Now we can define our Publisher type. Let's call it SyncPublisher, because when it receives a Subscriber, it immediately (synchronously) performs its entire computation.

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 像这样:

Now we can turn our process(_:progress:) function into a SyncPublisher like this:

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

run 闭包是 { process(inputFrames, progress: $0) }.请记住，这里的 $0 是一个 PassthroughSubject，正是 process(_:progress:) 想要作为其第二个参数的内容.

The run closure is { process(inputFrames, progress: $0) }. Remember that $0 here is a PassthroughSubject<Double, Never>, exactly what process(_:progress:) wants as its second argument.

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

When we subscribe to this pub, it will first create two subjects. One subject is the progress subject and gets passed to the closure. We'll use the other subject to publish either the final result and a .finished completion, or just a .failure completion if the run closure throws an error.

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

The reason we use two separate subjects is because it ensures that our publisher is well-behaved. If the run closure returns normally, the publisher publishes zero or more progress reports, followed by a single result, followed by .finished. If the run closure throws an error, the publisher publishes zero or more progress reports, followed by a .failed. There is no way for the run closure to make the publisher emit multiple results, or emit more progress reports after emitting the result.

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

At last, we can subscribe to pub to see if it works properly:

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

这篇关于Swift 将 Future 与多个值结合起来?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！