如何在没有自定义订阅者的情况下调整 Combine 的发布者需求？答案

【问题标题】：How to adjust Combine's Publisher demand without a custom Subscriber?如何在没有自定义订阅者的情况下调整 Combine 的发布者需求？
【发布时间】：2020-03-22 02:43:40
【问题描述】：

在Combine 框架中有一个demand 的概念，它允许向publishers 发送背压信号。

假设我有一个简单的发布者：

let numbers = Publishers.Sequence<ClosedRange<Int>, Error>(sequence: 0...100)

我想下载使用这些数字作为参数的某些 URL。我还希望仅在上一次下载完成后才开始下一次下载。

一个天真的方法看起来像这样：

let subscription = numbers.sink(receiveCompletion: { _ in }, receiveValue: {
  let url = URL(string: "https://httpbin.org/get?value=\($0)")!
  URLSession.shared.dataTask(with: url) { 
    $0.map { print(String(data: $0, encoding: .utf8)!) }
  }.resume()
})

不幸的是，这不能满足在开始下一个下载之前等待上一个下载完成的要求。据我所知，sink 函数将返回AnyCancellable 类型的值，而不是Subscription 类型的值。如果是后者，我们可以在上传完成后根据具体需求调用subscription上的request函数。

控制sink 或任何其他标准组合Subscriber 提供的订阅需求的最佳方法是什么？

【问题讨论】：

我个人觉得在操作中处理会更好（以前称为NSOperation），您可以添加对相同类型的正在进行的操作的依赖。它还将封装大部分代码，并可以返回一个Result，它可以由Future处理
我知道还有其他方法可以实现这一点，但我的问题是关于 Combine 及其 API。似乎令人费解的是，对背压的支持被宣传为 Combine 的一个显着特征，但对于一个据说可以通过背压解决的常见需求，并没有简单的解决方案。

标签： swift reactive-programming combine

【解决方案1】：

事实证明，flatMap 运算符采用额外的 maxPublishers 参数，该参数采用 Subscribers.Demand 值。结合Future 发布者，这允许numbers 发布者等到未来能够处理给定值，然后再发送下一个值。

将此应用于原始代码，一个接一个地下载值将如下所示：

enum DownloadError: Error {
    case noData
}

let subscription = numbers.flatMap(maxPublishers: .max(1)) { number in
    Future { promise in
        let url = URL(string: "https://httpbin.org/get?value=\(number)")!
        URLSession.shared.dataTask(with: url) {
            switch ($0, $2) {
            case let (data?, nil):
                promise(.success(data))
            case let (nil, error?):
                promise(.failure(error))
            default:
                promise(.failure(DownloadError.noData))
            }
        }.resume()
    }
}.sink(
    receiveCompletion: { _ in print("errors should be handled here") },
    receiveValue: { print(String(data: $0, encoding: .utf8)!) }
)

【讨论】：