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 channel = _client.channel('public:$path')
    ..onPostgresChanges(
      event: PostgresChangeEvent.insert,
      schema: 'public',
      table: path,
      callback: (payload) {
        final data = Map<String, dynamic>.from(payload.newRecord);
        final id = data.remove("id").toString();
        controller.add(ChatBackendMessage(id: id, data: _normalize(data)));
      },
    );
  channel.subscribe();
  _channels[path] = channel;
  controller.onCancel = () => _client.removeChannel(channel);
  return controller.stream;
}