streamQuery<T> method

Stream<List<RealtimeDocument<T>>> streamQuery<T>(
  1. RealtimeQueryDefinition definition, {
  2. FromJson<T>? fromJson,
})

Implementation

Stream<List<RealtimeDocument<T>>> streamQuery<T>(
  RealtimeQueryDefinition definition, {
  FromJson<T>? fromJson,
}) {
  final liveQuery = _queries.putIfAbsent(
    definition.queryId,
    () => RealtimeLiveQuery(definition),
  );

  return Stream.multi((controller) {
    final streamSubscription = liveQuery.stream.listen(
      (documents) {
        controller.add(_toDocuments(documents, fromJson));
      },
      onError: controller.addError,
      onDone: controller.close,
    );

    liveQuery.listenerCount += 1;
    if (liveQuery.listenerCount == 1) {
      liveQuery.subscriptionEpoch += 1;
      unawaited(_subscribeSafely(liveQuery));
    }

    if (liveQuery.hasSnapshot) {
      controller.add(_toDocuments(liveQuery.documents, fromJson));
    }

    controller.onCancel = () async {
      await streamSubscription.cancel();
      liveQuery.listenerCount -= 1;

      if (liveQuery.listenerCount <= 0) {
        liveQuery.listenerCount = 0;
        final epoch = liveQuery.subscriptionEpoch;
        unawaited(_closeIfStillIdle(liveQuery, epoch));
      }
    };
  }, isBroadcast: true);
}