在Swift的联合发布服务器中包装异步代码

Question

我有一个名为QueryObserver的类，它可以在一段时间内产生多个结果，返回为回调(闭包)。你就这样用它：

let observer = QueryObserver<ModelType>(query: query) { result in
  switch result {
  case .success(let value):
    print("result: \(value)")
  case .failure(let error):
    print("error: \(error)")
  }
}

(QueryObserver实际上是Firestore的笨重的query.addSnapshotListener功能的包装器，以防您有疑问。)使用现代Result类型而不是具有多个可选参数的回调。)

在一个较旧的项目中，我使用ReactiveKit，并有一个扩展将所有这些转换为Signal，如下所示：

extension QueryObserver {
  public static func asSignal(query: Query) -> Signal<[T], Error> {
    return Signal { observer in
      let queryObserver = QueryObserver<T>(query: query) { result in
        switch result {
        case .success(let value):
          observer.receive(value)
        case .failure(let error):
          if let firestoreError = error as? FirestoreError, case .noSnapshot = firestoreError {
            observer.receive([])
          } else {
            observer.receive(completion: .failure(error))
          }
        }
      }

      return BlockDisposable {
        queryObserver.stopListening()
      }
    }
  }
}

然而，在一个全新的项目中，我使用的是联合并试图重写这个项目。到目前为止，我已经成功地写了这个，但它不起作用。这是有意义的：observer不会被任何东西所保留，所以它会立即发布，什么都不会发生。

extension QueryObserver {
  public static func asSignal(query: Query) -> AnyPublisher<[T], Error> {
    let signal = PassthroughSubject<[T], Error>()

    let observer = QueryObserver<T>(query: query) { result in
      switch result {
      case .success(let value):
        print("SUCCESS!")
        signal.send(value)
      case .failure(let error):
        if let firestoreError = error as? FirestoreError, case .noSnapshot = firestoreError {
          signal.send([])
        } else {
          signal.send(completion: .failure(error))
        }
      }
    }

    return signal.eraseToAnyPublisher()
  }
}

我如何使合并版本工作？如何包装现有的异步代码？我找到的唯一示例是使用Future进行一次性回调，但随着时间的推移，我将处理多个值。

基本上，我正在寻找ReactiveKit-to-组合版本的这。

Answer 1

查看https://github.com/DeclarativeHub/ReactiveKit/issues/251#issuecomment-575907641以获得一个方便的信号组合版本，使用如下所示：

let signal = Signal<Int, TestError> { subscriber in
    subscriber.receive(1)
    subscriber.receive(2)
    subscriber.receive(completion: .finished)
    return Combine.AnyCancellable {
        print("Cancelled")
    }
}