如何结合Swift AsyncThrowingStream使用Firestore监听器及代码改造

核心改动思路 #

利用你提供的AsyncThrowingStream扩展，替换原有的手动ListenerRegistration管理逻辑，通过Swift异步序列的方式处理Firestore快照更新，同时保留原代码的错误处理和数据过滤逻辑。

改造后完整代码 #

1. 更新类属性与取消订阅方法

替换原有的listenerRegistration为Task来管理异步订阅：

// 移除原有的listenerRegistration，替换为Task
private var subscriptionTask: Task<Void, Error>?

func unsubscribe() {
    // 取消任务即可自动终止流并移除监听器
    subscriptionTask?.cancel()
    subscriptionTask = nil
}

2. 重写订阅方法

用异步流的方式处理快照更新：

func subscribe() {
    unsubscribe() // 先取消已有订阅
    guard let userID = Auth.auth().currentUser?.uid else { return }
    
    let docRef = Firestore.db.collection("mdates")
    let query = docRef.whereField(MeetDateKeys.selectedId.rawValue, isEqualTo: userID)
    
    // 启动异步任务订阅流
    subscriptionTask = Task { [weak self] in
        do {
            // 遍历异步流中的每一批解码后的数据
            for try await decodedDates in query.addSnapshotListener1(MeetDate.self) {
                self?.reminders = decodedDates
                
                // 更新过滤后的日期分组
                self?.thisWeek = self?.dateThisWeekFilter(dates: decodedDates)
                self?.nextWeek = self?.dateNextWeekFilter(dates: decodedDates)
                self?.laterWeek = self?.dateLaterFilter(dates: decodedDates)
            }
        } catch {
            // 处理流的全局错误（如权限不足、快照获取失败）
            print("Firestore订阅出错：\(error.localizedDescription)")
        }
    }
}

3. 调整扩展以匹配原解码逻辑

原扩展中单个文档解码失败会终止整个流，我们修改扩展，让它和原代码一样忽略失败文档并打印错误：

extension Query {
    func addSnapshotListener1<T>(
        includeMetadataChanges: Bool = false
    ) -> AsyncThrowingStream<[T], Error> where T: Decodable {
        .init { continuation in
            let listener = addSnapshotListener(includeMetadataChanges: includeMetadataChanges) { result in
                do {
                    let snapshot = try result.get()
                    // 逐个解码文档，捕获错误并忽略失败项，保留原错误打印逻辑
                    let decodedDocs = snapshot.documents.compactMap { doc in
                        do {
                            return try doc.data(as: T.self, decoder: Firestore.Decoder())
                        } catch {
                            switch error {
                            case DecodingError.typeMismatch(_, let context):
                                print("\(error.localizedDescription): \(context.debugDescription)")
                            case DecodingError.valueNotFound(_, let context):
                                print("\(error.localizedDescription): \(context.debugDescription)")
                            case DecodingError.keyNotFound(_, let context):
                                print("\(error.localizedDescription): \(context.debugDescription)")
                            case DecodingError.dataCorrupted(let key):
                                print("\(error.localizedDescription): \(key.debugDescription)")
                            default:
                                print("Error decoding document: \(error.localizedDescription)")
                            }
                            return nil
                        }
                    }
                    continuation.yield(decodedDocs)
                } catch {
                    // 处理快照获取失败的全局错误
                    continuation.finish(throwing: error)
                }
            }

            continuation.onTermination = { @Sendable _ in
                listener.remove()
            }
        }
    }
}

关键改动说明 #

• 自动管理监听器：Task取消时会触发continuation.onTermination，自动调用listener.remove()，无需手动维护监听器实例。
• 异步序列迭代：使用for try await遍历异步流，这是Swift处理持续异步事件的标准范式。
• 保留原有行为：调整扩展中的解码逻辑，确保单个文档解码失败不会终止整个订阅，仅打印错误并忽略该文档，和原代码逻辑完全一致。

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