subscribeToNewMessages method

@override Stream<ChatBackendMessage> subscribeToNewMessages(String path)

Subscribes to newly created messages for the conversation at path.

The returned stream emits one ChatBackendMessage per new message. Implementations must not emit messages already returned by fetchMessages.

Implementation

@override
Stream<ChatBackendMessage> subscribeToNewMessages(String path) {
  final controller = StreamController<ChatBackendMessage>();
  final subscription = _collectionFor(path)
      .orderBy("timestamp")
      .limitToLast(1)
      .snapshots(includeMetadataChanges: true)
      .skip(1)
      .listen((event) {
        if (event.metadata.hasPendingWrites) return;
        for (var change in event.docChanges) {
          controller.add(
            ChatBackendMessage(
              id: change.doc.id,
              data: _normalize(change.doc.data()! as Map<String, dynamic>),
            ),
          );
        }
      });
  _listeners[path] = subscription;
  controller.onCancel = subscription.cancel;
  return controller.stream;
}