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 subscription = _realtime.subscribe([
    'databases.$databaseId.tables.$path.rows',
  ]);
  _subscriptions[path] = subscription;
  return subscription.stream
      .where((message) => message.events.any((e) => e.endsWith(".create")))
      .map((message) {
        final data = Map<String, dynamic>.from(message.payload);
        final id = data.remove('\$id') as String;
        return ChatBackendMessage(id: id, data: _normalize(data));
      });
}