Swift Combine：收集条件成立值并包含触发终止的目标值

prefix(while:)的核心逻辑是仅传递满足条件的元素，当遇到第一个不满足条件的元素时，立即终止序列且不会传递该元素。这就是你代码中"LastPage"未被收集的原因——它触发了终止条件，但本身没有被下游接收。

解决方案1：使用scan+prefix+map组合处理 #

直接通过scan跟踪序列状态，确保触发终止的元素被包含在内：

import Foundation
import Combine

var subj = PassthroughSubject<String, Never>()

let cancel = subj
    // 用scan跟踪当前元素和是否继续接收后续元素的状态
    .scan((element: String(), shouldContinue: true)) { prev, current in
        let shouldContinue = prev.shouldContinue && (current != "LastPage")
        return (element: current, shouldContinue: shouldContinue)
    }
    // 保留所有元素直到：要么仍在继续状态，要么当前是触发终止的元素
    .prefix { $0.shouldContinue || $0.element == "LastPage" }
    // 提取实际元素，去掉状态标记
    .map { $0.element }
    .collect(.byTime(DispatchQueue(label: "Test"), .seconds(3)))
    .sink {
        print("complete: \($0)")
    } receiveValue: {
        print("received: \($0)")
    }

print("start")

let strings = [
    "!@#$",
    "ZXCV",
    "LastPage",
    "ASDF",
    "JKL:"
]

for i in (0..<strings.count) {
    DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(i)) {
        let s = strings[i]
        print("sending \(s)")
        subj.send(s)
    }
}

输出结果：

start
sending !@#$
sending ZXCV
sending LastPage
received: ["!@#$", "ZXCV", "LastPage"]
complete: finished
sending ASDF

解决方案2：自定义可复用的prefixIncludingWhile操作符 #

如果需要在多个地方复用这个逻辑，可以自定义一个操作符，封装上述逻辑：

import Foundation
import Combine

// 自定义Publisher，实现包含终止触发元素的prefix逻辑
extension Publishers {
    struct PrefixIncludingWhile<Upstream: Publisher>: Publisher {
        typealias Output = Upstream.Output
        typealias Failure = Upstream.Failure
        
        private let upstream: Upstream
        private let predicate: (Upstream.Output) -> Bool
        
        init(upstream: Upstream, predicate: @escaping (Upstream.Output) -> Bool) {
            self.upstream = upstream
            self.predicate = predicate
        }
        
        func receive<S>(subscriber: S) where S : Subscriber, Failure == S.Failure, Output == S.Input {
            upstream
                .scan((element: Output?.none, shouldContinue: true)) { prev, current in
                    let shouldContinue = prev.shouldContinue && self.predicate(current)
                    return (element: current, shouldContinue: shouldContinue)
                }
                .prefix { $0.shouldContinue || !self.predicate($0.element!) }
                .map { $0.element! }
                .receive(subscriber: subscriber)
        }
    }
}

// 给Publisher扩展便捷调用方法
extension Publisher {
    func prefixIncludingWhile(_ predicate: @escaping (Output) -> Bool) -> Publishers.PrefixIncludingWhile<Self> {
        Publishers.PrefixIncludingWhile(upstream: self, predicate: predicate)
    }
}

// 使用示例
var subj = PassthroughSubject<String, Never>()

let cancel = subj
    .prefixIncludingWhile { $0 != "LastPage" }
    .collect(.byTime(DispatchQueue(label: "Test"), .seconds(3)))
    .sink {
        print("complete: \($0)")
    } receiveValue: {
        print("received: \($0)")
    }

print("start")

let strings = [
    "!@#$",
    "ZXCV",
    "LastPage",
    "ASDF",
    "JKL:"
]

for i in (0..<strings.count) {
    DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(i)) {
        let s = strings[i]
        print("sending \(s)")
        subj.send(s)
    }
}

这个自定义操作符的用法和原生prefix(while:)几乎一致，但会自动包含触发终止条件的那个元素。

内容的提问来源于stack exchange，提问作者Yogurt