如何人工完成一个流？

Question

我有一个整数流，我需要使用一个结构来处理，该结构使用

CurrentValueSubject

来知道它是否被允许进行处理。由于

CurrentValueSubject

未完成，如果我在

flatMap()

内进行处理，则整个流 (

fooTheBars

) 只有在整数流为空时才能完成。

如果我注释掉处理部分（

.flatMap { bar in fooer.doFoo(with: bar) }

），流会正常完成，因为不涉及

CurrentValueSubject

。但我需要它在处理完所有项目后正常完成，无论项目数量如何。

我看到的输出：

[0, 1, 2]
doing foo with 0
doing foo with 1
doing foo with 2
received value
received value
received value

我想要它输出什么：

[0, 1, 2]
doing foo with 0
doing foo with 1
doing foo with 2
received value
received value
received value
received completion // hooray, completion!

流为空时的输出（正常完成，这就是我想要的）：

[]
received completion // hooray, completion!

如果我注释掉处理（

.flatMap()

），一切都很好：

[0, 1, 2]
received value
received value
received value
received completion // hooray, completion!

如何修改评论下方的代码，使所有项目处理完成时整体流完成？像这样使用

CurrentValueSubject

是一种固有的坏模式吗？

struct Fooer {
    private let readyToFoo = CurrentValueSubject<Bool, Never>(true)

    func doFoo(with item: Int) -> AnyPublisher<Void, Never> {
        readyToFoo
            .filter { $0 }
            .map { _ in () }
            .handleEvents(receiveOutput: {
                print("doing foo with \(item)") // processing
            })
            .delay(for: .seconds(1), scheduler: DispatchQueue.main)
            .eraseToAnyPublisher()
    }
}

func getBars() -> AnyPublisher<Int, Never> {
    let bars = (0..<(Bool.random() ? 0 : 3))
    print(Array(bars))
    return bars.publisher
        .eraseToAnyPublisher()
}

// can't change anything above this comment

let fooer = Fooer()
let fooTheBars = getBars()
    .flatMap { bar in
        fooer.doFoo(with: bar)
    }
    .sink(receiveCompletion: { completion in
        print("received completion")
    }, receiveValue: { _ in
        print("received value")
    })