取消嵌套的组合发布者

Question

我正在尝试将我的应用切换为使用组合管道。我希望简化线程管理，但却让自己陷入了 Combine 的意外行为。

我假设尽管我订阅了 DispatchQueue.global()，但取消主管道会取消嵌套订阅。

这是我的游乐场：

import Cocoa
import Combine

let folders = ["folder1", "folder2", "folder3", "folder4"]

class OneByOnePublisher: Publisher {
    typealias Output = String
    typealias Failure = Never
    
    let input: [String]
    init(input: [String]) {
        self.input = input
    }
  
    func receive<Downstream: Subscriber>(subscriber: Downstream) where Downstream.Input == Output, Downstream.Failure == Failure {
        let subject = PassthroughSubject<String, Never>()
        subject.receive(subscriber: subscriber)
        for value in input {
            subject.send(value)
        }
        subject.send(completion: .finished)
    }
}

func uppercase(_ character: Character) -> String {
    print("Uppercasing \(character)")
    Thread.sleep(forTimeInterval: 0.5)
    return character.uppercased()
}

func uppercasePublisher(_ folder: String) -> AnyPublisher<String, Never> {
    return folder.publisher
//        .handleEvents(receiveCancel: { print("Received cancel in nested") })
        .map{uppercase($0)}
        .collect()
        .map{$0.joined()}
        .eraseToAnyPublisher()
}


let stringPublisher = PassthroughSubject<String, Never>()
let oneByOnePublisher = OneByOnePublisher(input: folders)


let cancelable = oneByOnePublisher
    .subscribe(on: DispatchQueue.global())
    .handleEvents(receiveCancel: { print("Received cancel in main") })
    .flatMap{uppercasePublisher($0)}
    .receive(on: DispatchQueue.main)
    .sink { (completion) in
        print("Received completion: \(completion)")
    } receiveValue: { (value) in
        print("Received value: \(value)")
    }

Thread.sleep(forTimeInterval: 2)
cancelable.cancel()
Thread.sleep(forTimeInterval: 2)
print("Done")

那个输出

Uppercasing f
Uppercasing o
Uppercasing l
Uppercasing d
Received cancel in main
Uppercasing e
Uppercasing r
Uppercasing 1
Done

但是，如果我取消注释该行

//        .handleEvents(receiveCancel: { print("Received cancel in nested") })

那么输出就是我一开始所期望的

Uppercasing f
Uppercasing o
Uppercasing l
Uppercasing d
Received cancel in nested
Received cancel in main
Done

我错过了什么？为什么在第一种情况下嵌套订阅不会立即取消？为什么添加 handleEvents() 会改变取消流程？

Answer 1

Publishers.Sequence 发布者可能存在错误，这是由folder.publisher 调用产生的错误。添加handleEvents 调用会隐藏此错误，因为调用会导致包装发布者似乎可以正确处理取消。

为了检验这个理论，让我们调整您的OneByOnePublisher 以适用于任何类型的序列（这也将使其适用于原始字符串数组）：

class OneByOnePublisher<Seq: Sequence>: Publisher {
    typealias Output = Seq.Element
    typealias Failure = Never
    
    let input: Seq
    init(input: Seq) {
        self.input = input
    }
  
    func receive<Downstream: Subscriber>(subscriber: Downstream) where Downstream.Input == Output, Downstream.Failure == Failure {
        let subject = PassthroughSubject<Output, Failure>()
        subject.receive(subscriber: subscriber)
        for value in input {
            subject.send(value)
        }
        subject.send(completion: .finished)
    }
}

extension Sequence {
    var oneByOne: OneByOnePublisher<Self> { OneByOnePublisher(input: self) }
}

现在如果我们更改 uppercasePublisher 以使用增强版发布者

func uppercasePublisher(_ folder: String) -> AnyPublisher<String, Never> {
    return folder.oneByOne
        // .handleEvents(receiveCancel: { NSLog("Received cancel in nested") })
        .map{uppercase($0)}
        .collect()
        .map{$0.joined()}
        .eraseToAnyPublisher()
}

我们可以看到，无论handleEvents 行是否被注释，取消都按预期工作。这表明原来的 folder.publisher 行是问题的根源，更具体的说 - 它是有问题的 Publishers.Sequence 实例。