如何实现触发特定值后开启/关闭Combine的collect()收集功能？

要实现“收到特定值后开启collect输出，收到另一个值后关闭”的需求，核心是用状态跟踪过滤collect的结果——上游单个值无法直接控制时间窗口的collect结果，所以需要在collect之后结合状态判断来决定是否输出。

基础实现：仅在收到5后开启输出 #

修改你的代码，通过scan维护触发状态，配合filter和map控制输出：

import UIKit
import Foundation
import Combine

var subj = PassthroughSubject<Int, Never>()

for iter in 0...10 {
    DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(iter)) {
        print("send \(iter)")
        subj.send(iter)
    }
}

let queue = DispatchQueue(label: "Q")

let cancellable = subj
    .collect(.byTime(queue, .seconds(2)))
    // 用scan跟踪触发状态，元组格式：(是否已触发, 当前收集的数组)
    .scan((isTriggered: false, values: [Int]())) { currentState, newValues in
        // 一旦触发（收到过5）就保持激活状态
        let newTriggered = currentState.isTriggered || newValues.contains(5)
        return (isTriggered: newTriggered, values: newValues)
    }
    // 只输出已激活状态下的收集结果
    .filter { $0.isTriggered }
    // 提取出最终要输出的数组
    .map { $0.values }
    .sink(receiveCompletion: {
        print("finish")
        print($0)
    }, receiveValue: {
        print($0)
    })

修改后的输出：

send 0
send 1
send 2
send 3
send 4
send 5 
send 6
[4, 5]
send 7
send 8
[6, 7]
send 9
send 10
[8, 9, 10]
finish
finished

前两个收集窗口[0,1]和[2,3]被过滤，从包含5的窗口开始输出后续所有结果。

扩展实现：支持开启和关闭 #

如果需要在收到另一个值（比如8）时关闭输出，可以用枚举定义状态，更灵活地控制生命周期：

import UIKit
import Foundation
import Combine

var subj = PassthroughSubject<Int, Never>()

for iter in 0...10 {
    DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(iter)) {
        print("send \(iter)")
        subj.send(iter)
    }
}

let queue = DispatchQueue(label: "Q")

// 定义收集状态枚举
enum CollectState {
    case inactive // 未开启
    case active // 已开启
    case stopped // 已关闭
}

let cancellable = subj
    .collect(.byTime(queue, .seconds(2)))
    .scan((state: CollectState.inactive, values: [Int]())) { currentState, newValues in
        var newState = currentState.state
        switch newState {
        case .inactive:
            // 收到5时切换为激活状态
            if newValues.contains(5) {
                newState = .active
            }
        case .active:
            // 收到8时切换为关闭状态
            if newValues.contains(8) {
                newState = .stopped
            }
        case .stopped:
            // 保持关闭状态
            break
        }
        return (state: newState, values: newValues)
    }
    // 只输出激活状态下的结果
    .filter { $0.state == .active }
    .map { $0.values }
    .sink(receiveCompletion: {
        print("finish")
        print($0)
    }, receiveValue: {
        print($0)
    })

输出结果：

send 0
send 1
send 2
send 3
send 4
send 5 
send 6
[4, 5]
send 7
send 8
[6, 7]
send 9
send 10
finish
finished

收到8后状态切换为关闭，后续的[8,9,10]不再输出。

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