使用`.receive（on：DispatchQueue.main）`时未接收输入答案

【问题标题】：Not receiving inputs when using `.receive(on: DispatchQueue.main)`使用`.receive（on：DispatchQueue.main）`时未接收输入
【发布时间】：2019-06-20 00:56:43
【问题描述】：

我正在尝试使用.receive(on: DispatchQueue.main) 更改为下游的主线程，但是在使用.subscribe(:) 或.sink(receiveValue:) 时我没有收到输入。如果我不更改线程，我会收到正确的输入。

出版商

extension URLSessionWebSocketTask {
  struct ReceivePublisher: Publisher {
    typealias Output = Message
    typealias Failure = Error

    let task: URLSessionWebSocketTask

    func receive<S>(subscriber: S) where S: Subscriber, Output == S.Input, Failure == S.Failure {
      task.receive { result in
        switch result {
        case .success(let message): _ = subscriber.receive(message)
        case .failure(let error): subscriber.receive(completion: .failure(error))
        }
      }
    }
  }
}

extension URLSessionWebSocketTask {
  func receivePublisher() -> ReceivePublisher {
    ReceivePublisher(task: self)
  }
}

订阅者

extension ViewModel: Subscriber {
  typealias Input = URLSessionWebSocketTask.Message
  typealias Failure = Error

  func receive(subscription: Subscription) {}

  func receive(_ input: URLSessionWebSocketTask.Message) -> Subscribers.Demand {
    // Handle input here.
    // When using `.receive(on:)` this method is not called when should be.
    return .unlimited
  }

  func receive(completion: Subscribers.Completion<Error>) {}
}

socketTask.receivePublisher()
      .receive(on: DispatchQueue.main)
      .subscribe(viewModel)
socketTask.resume()

【问题讨论】：

放一些示例代码，以便有人可以帮助您。
尝试更详细地描述您想要实现的目标以及您面临的问题
How to change thread using combine Publisher?的可能重复
谢谢，但问题不在于如何做，而在于为什么在使用它时取消下游。换句话说，这个问题的答案现在可以在 beta 2 中使用，而在这里我认为这是给定的。
你保存可取消的吗？如果您返回 AnyCancellable 并且不保存它，那么它将被取消初始化并且订阅将被取消。

标签： ios swift combine

【解决方案1】：

subscribe<S>(_ subject: S) -> AnyCancellable 返回的AnyCancellable 在被取消初始化后将调用cancel()。因此，如果您不保存它，它将在调用块超出范围时被取消初始化。

在我从 WWDC 看到的视频和教程中，从未解决如何使用它。我所看到的是人们正在转向 RxSwift 的DisposeBag 解决方案。

更新 Beta 4： 现在，Combine 在AnyCancellable 上提供了一个名为：store(in:) 的方法，它的功能与我的旧解决方案差不多。您可以将AnyCancellables 存储在一组AnyCancellable 中：

var cancellables = Set<AnyCancellable>()
...
override func viewDidLoad() {
    super.viewDidLoad()
    ...
    socketTask.receivePublisher()
        .receive(on: DispatchQueue.main)
        .subscribe(viewModel)
        .store(in: &cancellables)
}

这样，当包含的类被取消初始化时，数组（以及所有AnyCancellables）将被取消初始化。

过时：

如果您想要一个适用于所有Cancellables 的解决方案，可以以更好的方式使用，您可以扩展Cancellable，如下所示：

extension Cancellable {

    func cancel(with cancellables: inout [AnyCancellable]) {
        if let cancellable = self as? AnyCancellable {
            cancellables.append(cancellable)
        } else {
            cancellables.append(AnyCancellable(self))
        }
    }

}

【讨论】：