看看这个要点：https://gist.github.com/IanKeen/934258d2d5193160391e14d25e54b084
有了上面的要点，你就可以做到：

func observe<T: Codable>(document: String, inCollection collection: String) -> AnyPublisher<T, Error> {
    let docRef = self.db.collection(collection).document(document)
    return AnyPublisher { subscriber in
        let docRef = self.db.collection(collection).document(document)
        let listener = docRef.addSnapshotListener { (snapshot, error) in
            guard let object = T(dictionary: snapshot?.data()), error == nil else {
                subscriber.send(completion: .failure(error ?? CRUDServiceError.encodingError))
                return
            }
            subscriber.send(object)
        }

        return AnyCancellable { listener.remove() }
    }
}

至于错误...发布服务器在发出错误后 * 不能 * 发出任何更多的值。这是契约的基本部分。然后，您能做的最好的事情是将错误转换为其他类型，并作为下一个事件而不是完成事件发出。
像这样的东西会做到这一点：

func observe<T: Codable>(document: String, inCollection collection: String) -> AnyPublisher<Result<T, Error>, Never> {
    let docRef = self.db.collection(collection).document(document)
    return AnyPublisher { subscriber in
        let docRef = self.db.collection(collection).document(document)
        let listener = docRef.addSnapshotListener { (snapshot, error) in
            guard let object = T(dictionary: snapshot?.data()), error == nil else {
                subscriber.send(.failure(error ?? CRUDServiceError.encodingError))
                return
            }
            subscriber.send(.success(object))
        }

        return AnyCancellable { listener.remove() }
    }
}

但是我怀疑一旦Firestore向回调函数发送了一个错误，它也会停止用新的快照调用回调函数...所以我不认为这实际上是有用的。