当内部发布者使用 subscribe(on:) 时，CombineLatest 运算符不会发出答案

【问题标题】：CombineLatest operator is not emitting when inners publishers use subscribe(on:)当内部发布者使用 subscribe(on:) 时，CombineLatest 运算符不会发出
【发布时间】：2021-07-16 21:11:44
【问题描述】：

我观察到有关 CombineLatest 的意外行为，如果内部发布者有 subscribe(on:)，则 CombineLatest 流不会发出任何值。

注意事项：

Zip 操作员正在工作
将 subscribe(on:) / receive(on:) 移动到 combineLatest 流也可以。但是在这个特定的用例中，内部发布者正在定义他们的订阅/接收因为在其他地方被（重新）使用了。
仅将 subscribe(on:)/receive(on:) 添加到其中一个内部发布者也可以，所以问题在于两者都有。

    func makePublisher() -> AnyPublisher<Int, Never> {
        Deferred {
            Future { promise in
                DispatchQueue.global(qos: .background).asyncAfter(deadline: .now() + 3) {
                    promise(.success(Int.random(in: 0...3)))
                }
            }
        }
        .subscribe(on: DispatchQueue.global())
        .receive(on: DispatchQueue.main)
        .eraseToAnyPublisher()
    }
    
    var cancellables = Set<AnyCancellable>()
    Publishers.CombineLatest(
        makePublisher(),
        makePublisher()
    )
    .sink { completion in
        print(completion)
    } receiveValue: { (a, b) in
        print(a, b)
    }.store(in: &cancellables)

这是一个组合错误还是预期的行为？您知道如何设置这种内部人员可以定义自己的订阅调度程序的流吗？

【问题讨论】：

你有什么问题？
@NewDev 刚刚更新了问题

标签： ios swift combine combinelatest

【解决方案1】：

是的，这是一个错误。我们可以将测试用例简化为：

import Combine
import Dispatch

let pub = Just("x")
    .subscribe(on: DispatchQueue.main)

let ticket = pub.combineLatest(pub)
    .sink(
        receiveCompletion: { print($0) },
        receiveValue: { print($0) })

这从不打印任何东西。但是，如果您注释掉 subscribe(on:) 运算符，它会打印出预期的内容。如果您将subscribe(on:) 保留在其中，但插入一些print() 运算符，您将看到CombineLatest 运算符永远不会向上游发送任何请求。

我建议你复制CombineX reimplementation of CombineLatest 和它需要编译的实用程序（我认为Lock 和LockedAtomic 的CombineX 实现）。我不知道 CombineX 版本也可以，但如果它有问题，至少你有源并且可以尝试修复它。

【讨论】：