在 Swift Combine 中创建顺序发布者

Posted 2023-03-10

【中文标题】在 Swift Combine 中创建顺序发布者

【英文标题】：Creating a sequential publisher in Swift Combine

【发布时间】：2021-12-25 17:32:22

【问题描述】：

我想对一批事件进行去抖动处理，并在大约 1.5 秒的延迟后处理它们。这就是我所做的。

class ViewModel: ObservableObject 

    @Published var pending: [TaskInfo]
    private var cancellable: AnyCancellable? = nil

    init() 
        processPendingTasks()
    

    func queueTask(task: TaskInfo)  
        pending.append(task)
    

    private func processPendingTasks() 
        cancellable = $pendingTasks
                .debounce(for: 1.5, scheduler: RunLoop.main)
                .sink(receiveValue:  batch in 
                    // Iterate though elements and process events.
                )           
    

问题：这很好用，但我遇到的问题是它执行了不必要的视图更新，因为数组被标记为 @Published。

我在寻找什么： 理想的方法是流式设置，我可以在其中获取所有事件（以批处理方式），但 sink 应该在最后一个事件之后等待 1.5 秒已添加事件。

我尝试了PassthroughSubject，但它似乎只能让我知道最后 1.5 秒内发生的最后一个事件。

【参考方案1】：

一个可能的解决方案是PassthroughSubject 和collect 运算符的组合。在queueTask 中将任务发送给主题。

func queueTask(task: TaskInfo)  
    subject.send(task)

收到最后一条发送后 1.5 秒

subject.send(completion: .finished)

并订阅

subject
.collect()
.sink  [weak self] tasks in 
   self?.pending = tasks
  

如果传入任务的间隔小于 1.5 秒，您还可以使用 .timeout(1.5) 运算符，它会在超时间隔后终止管道。

【讨论】：

这是一个不错的解决方案。我试过了，现在只有一个问题。因为我打电话给.finished，所以接收器只被调用一次。我想要的是即使在调用接收器之后继续流，即在一个批次完成处理之后 - 它应该重新开始监听新批次。

正如你所写的在添加最后一个事件之后我假设传输在最后一个项目之后完成。或者，您可以使用 collect(n) 以 n 个块的形式发出项目。

是的，我最终使用了延迟为 1.5 的 .collect(.byTime) 模式。它适用于我的用例，尽管不完全符合我在问题中提出的要求。

collect() 是我正在寻找的缺失部分。谢谢你！