合并 - 从容器中移除管道

Question

假设以下简化代码-sn-p

import Foundation
import Combine

public class NetworkFetch {
  fileprivate var networkPipelines : Set<AnyCancellable> = []
  
  public func loadDataFor(url : URL)
  { 
    URLSession.shared.dataTaskPublisher(for: url)
      .map { $0.data }
      .decode(type: City.self, decoder: JSONDecoder())
      .eraseToAnyPublisher()
      .sink(receiveCompletion: {_ in print("Finish")},
            receiveValue: { v in
              print("\(c)\n")
            }
      )
      .store(in: &networkPipelines)
  }
}

对于 loadDataFor 的每次调用，都会生成一个新的 combine-pipeline 并将其添加到 networkPipelines 容器中。这个容器会随着时间的推移而增长。

一旦 URLSession-pipeline 获取所有数据，从该容器中删除此类 URLSession-pipeline 的正确方法是什么？

Answer 1

您可以做的一件事是从sink 内部删除您自己的订阅：

但也许更好的方法是订阅 PassthroughSubject 一次并通过它发送请求的 URL 和回调：

private let subject = PassthroughSubject<(URL, (City) -> Void)), Never>()
private var c: Set<AnyCancellable> = []

init() {
   subject
      .flatMap { (url, callback) in
          URLSession.shared.dataTaskPublisher(for: url)
             .map(\.data)
             .decode(type: City.self, decoder: JSONDecoder())
             .zip(Just(callback).setFailureType(to: Error.self))
      }
      .sink(receiveCompletion: {_ in print("Finish")},
            receiveValue: { (city, callback) in
              callback(city)
            }
      )
      .store(in: &c)
}

public func loadDataFor(url : URL, callback: @escaping (City) -> Void) {
   subject.send(url, callback)
}

因此，单个订阅可以通过发送一对请求的URL 和通过主题的回调来处理多个请求。