Flutter Firestore问题：无法删除广播流中的消息条目

开发一款聊天应用，需求是消息被监听器接收后立即删除该消息。作为Flutter+Firestore新手，目前遇到问题：已删除的消息仍会被重复接收，以下是相关代码及输出：

服务文件 #

@override
Stream<Message> messages({required User activeUser}) {
  _startReceivingMessages(activeUser);
  return _controller.stream;
}

_startReceivingMessages(User activeUser) {
  _changeFeed = _messageCollection
      .where('to', isEqualTo: activeUser.id)
      .snapshots(includeMetadataChanges: false)
      .asBroadcastStream()
      .listen((snapData) {
    if (snapData.docs.isEmpty) return;
    snapData.docs.forEach((feedData) async {
      final message = _messageFromFeed(feedData);
      _controller.sink.add(message);
      await _removeDeliverredMessage(message);
    });
  });
}

Message _messageFromFeed(feedData) {
  return feedData.data();
}

_removeDeliverredMessage(Message message) async {
  log("try deleting start");
  // _changeFeed!.pause();
  await _messageCollection.doc(message.id).delete();
  // _changeFeed!.resume();
  log("try deleting stop");
}

//collections
CollectionReference<Message> get _messageCollection =>
    firestore.collection('messages').withConverter<Message>(            
            fromFirestore: ((snapshot, _) {
              return Message.fromJson({
                'id': snapshot.id,
                'to': snapshot.data()!['to'],
                'from': snapshot.data()!['from'],
                'contents': encryption.decrypt(snapshot.data()!['contents']),
                'timestamp': snapshot.data()!['timestamp'].toDate()
              });
            }),
            toFirestore: ((message, _) => message.toJson()),
          );

测试文件 #

test('receive suscribe and recieve', () async {
    Message message = Message(
      from: user2.id!,
      to: user1.id!,
      timestamp: DateTime.now(),
      contents: "Mesaage from Rahul",
    );

    Message secondMessage = Message(
      from: user2.id!,
      to: user1.id!,
      timestamp: DateTime.now(),
      contents: "Another Mesaage from Rahul",
    );

    await sut.send(message);
    sut.messages(activeUser: user1).listen((message) {
      log("Messages :::");
      log(message.toJson().toString());
    });
    await sut.send(secondMessage);
  });

输出日志 #

Connecting to VM Service at http://127.0.0.1:59909/5QohWpYTglo=/ws
[log] ================LISTERN Message====================
[log] try deleting start
[log] Messages :::
[log] {from: 1111, to: 1234, contents: Mesaage from Rahul, timestamp: 2022-09-17 14:50:34.097438}
[log] try deleting stop
[log] ================LISTERN Message====================
2
[log] try deleting start
[log] Messages :::
[log] {from: 1111, to: 1234, contents: Mesaage from Rahul, timestamp: 2022-09-17 14:50:34.097438}
2
[log] try deleting stop
[log] Messages :::
[log] {from: 1111, to: 1234, contents: Another Mesaage from Rahul, timestamp: 2022-09-17 14:50:34.097715}
✓ receive suscribe and recieve
Exited

核心问题 #

1. 重复创建监听器：每次调用messages()都会执行_startReceivingMessages()，导致多个监听器同时监听同一个查询，重复处理消息。
2. 未区分文档变更类型：Firestore的snapshots()返回的快照包含所有文档状态（新增、修改、删除），代码直接遍历所有文档，删除操作触发的快照仍会把已删除的文档再次处理。
3. 异步循环时序问题：forEach中使用async会导致循环不等待删除操作完成，多个删除操作并行，可能引发快照重复触发。

修改方案 #

1. 避免重复创建监听器 #

在_startReceivingMessages()中先检查_changeFeed是否已存在，已存在则不再创建新的监听器：

void _startReceivingMessages(User activeUser) {
  // 避免重复创建监听器
  if (_changeFeed != null && !_changeFeed!.isPaused) return;
  
  _changeFeed = _messageCollection
      .where('to', isEqualTo: activeUser.id)
      .snapshots(includeMetadataChanges: false)
      .asBroadcastStream()
      .listen((snapData) async {
    // 遍历文档变更，只处理新增的消息
    for (final change in snapData.docChanges) {
      if (change.type == DocumentChangeType.added) {
        final feedData = change.doc;
        final message = _messageFromFeed(feedData);
        _controller.sink.add(message);
        await _removeDeliverredMessage(message);
      }
    }
  });
}

2. 只处理新增的文档变更 #

使用docChanges替代直接遍历docs，只处理DocumentChangeType.added类型的变更，这样删除操作触发的快照不会重复处理已删除的消息。

3. 用for循环替代forEach确保异步顺序 #

将forEach改为for循环，确保每个消息的删除操作完成后再处理下一个，避免并行操作导致的时序问题。

4. 优化测试代码（可选） #

测试中需要等待流事件处理完成，否则测试会提前结束，可使用Completer来等待消息接收：

test('receive subscribe and receive', () async {
  final completer = Completer<void>();
  int messageCount = 0;

  Message message = Message(
    from: user2.id!,
    to: user1.id!,
    timestamp: DateTime.now(),
    contents: "Message from Rahul",
  );

  Message secondMessage = Message(
    from: user2.id!,
    to: user1.id!,
    timestamp: DateTime.now(),
    contents: "Another Message from Rahul",
  );

  // 先订阅流
  sut.messages(activeUser: user1).listen((receivedMessage) {
    log("Messages :::");
    log(receivedMessage.toJson().toString());
    messageCount++;
    // 收到两条消息后完成测试
    if (messageCount == 2) {
      completer.complete();
    }
  });

  await sut.send(message);
  await sut.send(secondMessage);
  
  // 等待所有消息处理完成
  await completer.future;
});

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