如何压缩超过 4 个出版商

Question

我正在为我的 API 请求使用 Swift Combine。现在我面临的情况是，我想要将 4 个以上的并行请求压缩在一起。在我使用 Zip4() 运算符将 4 个请求压缩在一起之前。我可以想象您分多个步骤进行压缩，但我不知道如何为其编写 receiveValue。

这是我当前代码的简化，包含 4 个并行请求：

    Publishers.Zip4(request1, request2, request3, request4)
        .sink(receiveCompletion: { completion in
            // completion code if all 4 requests completed
        }, receiveValue: { request1Response, request2Response, request3Response, request4Response in
            // do something with request1Response
            // do something with request2Response
            // do something with request3Response
            // do something with request4Response
        }
    )
        .store(in: &state.subscriptions)

Answer 1

阻止您压缩任意数量的发布者的事情是非常不幸的事实，即 Apple 选择使 zip 运算符的输出为tuple。元组非常不灵活，而且功能有限。你不能有一个包含 10 个元素的元组；并且您甚至不能将元素附加到元组，因为这会导致您获得不同的类型。因此，我们需要的是一个新的操作符，它可以完成相同的工作，zip但会发出一些更强大、更灵活的结果，例如数组。

我们可以做一个！幸运的是，zip操作符本身有一个transform参数，可以让我们指定我们想要的输出类型。

好的，为了说明，我将把十个出版商压缩在一起。首先，我将制作一个包含 10 个发布者的数组；他们将仅仅是出版商，但这足以说明这一点，并证明我没有作弊，我将任意延迟给他们每个人：

let justs = (1...10).map {
    Just($0)
        .delay(for: .seconds(Int.random(in:1...3)), scheduler: DispatchQueue.main)
        .eraseToAnyPublisher() }

好的，现在我有一个发布者数组，我将把它们压缩在一起：

let result = justs.dropFirst().reduce(into: AnyPublisher(justs[0].map{[$0]})) { 
    res, just in
    res = res.zip(just) {
        i1, i2 -> [Int] in
        return i1 + [i2]
    }.eraseToAnyPublisher()
}

注意zip运算符后面的尾随闭包！这确保我的输出将是一个Array<Int>而不是元组。与元组不同，我可以创建任意大小的数组，只需在每次循环中添加元素。

好的，result现在是一个 Zip 发布者，它将10 个发布者压缩在一起。为了证明这一点，我将向它附加一个订阅者并打印输出：

result.sink {print($0)}.store(in: &self.storage)

我们运行代码。有一个令人心碎的停顿——这是正确的，因为每个 Just 发布者都有不同的随机延迟，而 zip 的规则是他们都需要在我们得到任何输出之前发布。他们迟早都会这样做，并且输出会出现在控制台中：

[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

完全正确的答案！我已经证明我确实将十个出版商压缩在一起以产生由每个出版商的单一贡献组成的输出。

将任意数量的数据任务发布者（或您正在使用的任何内容）压缩在一起并没有什么不同。

（对于一个相关的问题，在那里我学习了如何序列化任意数量的数据任务发布者，请参阅结合框架序列化异步操作。）

Answer 2

认为你可以这样完成：

let zipped1 = Publishers.Zip4(request1, request2, request3, request4)    
let zipped2 = Publishers.Zip4(request5, request6, request7, request8)

Publishers.Zip(zipped1, zipped2)
    .sink(receiveCompletion: { completion in
        // completion code if all 8 requests completed
    }, receiveValue: { response1, response2 in
        // do something with response1.0
        // do something with response1.1
        // do something with response1.2, response1.3, response2.0, response2.1, response2.2, response2.3
    }
)
    .store(in: &state.subscriptions)

Answer 3

基于马特的回答：

extension Publishers {
    struct ZipMany<Element, F: Error>: Publisher {
        typealias Output = [Element]
        typealias Failure = F

        private let upstreams: [AnyPublisher<Element, F>]

        init(_ upstreams: [AnyPublisher<Element, F>]) {
            self.upstreams = upstreams
        }

        func receive<S: Subscriber>(subscriber: S) where Self.Failure == S.Failure, Self.Output == S.Input {
            let initial = Just<[Element]>([])
                .setFailureType(to: F.self)
                .eraseToAnyPublisher()

            let zipped = upstreams.reduce(into: initial) { result, upstream in
                result = result.zip(upstream) { elements, element in
                    elements + [element]
                }
                .eraseToAnyPublisher()
            }

            zipped.subscribe(subscriber)
        }
    }
}

单元测试可以使用以下内容作为输入：

let upstreams: [AnyPublisher<String, Never>] = [
    Just("first")
        .receive(on: DispatchQueue.main)
        .eraseToAnyPublisher(),
    Just("second").eraseToAnyPublisher()
]

将该.receive(on:)事件的发射放在主队列的末尾，以便它将在"second".