Flutter Firestore real-time pagination using multiple listeners streamed combined

Question

I want to create a pagination screen, using Bloc, from the Firestore database. The screen should update on document changes.

My FirestoreProviderApi receives fetch requests from the Bloc unit. After fetching the documents from Firestore, it sends them back to the Bloc using a Stream.

The FirestoreProviderApi:

  final _pollOverviewStreamController = BehaviorSubject<QuerySnapshot>();

  @override
  Stream<QuerySnapshot> getOverviewPolls<QuerySnapshot>() =>
      _pollOverviewStreamController.asBroadcastStream().cast();

  @override
  Future<void> fetchFirstOverviewPolls() async {
    _overviewPollsRef.limit(5).snapshots().listen((querySnapshot) {
      _pollOverviewStreamController.add(querySnapshot);
    });
  }

  @override
  Future<void> fetchNextOverviewPolls() async {
    if (_pollOverviewStreamController.value.docs.isEmpty) return;

    final lastDoc = _pollOverviewStreamController.value.docs.last;
    _overviewPollsRef
        .startAfterDocument(lastDoc)
        .limit(5)
        .snapshots()
        .listen((querySnapshot) {
      _pollOverviewStreamController.add(querySnapshot);
    });
  }

On document changes, I want to update my list of documents.

Right now, whenever a change occurs, a new QuerySnapshot is appended to the Stream (instead of replacing the old one). Is there a way to combine multiple listeners to the same Stream, aggregating only the most up-to-date data?

Answer 1

Issue solved.

I found good documentation of how to implement real-time pagination using Firestore (been implemented it, works like a charm).

Youtube implementation: Flutter and Firestore real-time Pagination.

High-level steps:

1. Create a class-scoped document list state variable.
2. Stream each time the whole list.
3. Listen and iterate over snapshot changes, and update/add documents to the main list.
4. If the list is not empty, then fetch from the last document on the list.

My implementation:

final List<DocumentSnapshot> _overviewPolls = [];

@override
  Future<void> fetchOverviewPolls(
      [int firstFetchLimit = 1, int nextFetchLimit = 1]) async {

    Query overviewPollsQuery = _overviewPollsRef.limit(firstFetchLimit);

    if (_overviewPolls.isNotEmpty) {
      overviewPollsQuery = overviewPollsQuery
          .startAfterDocument(_overviewPolls.last)
          .limit(nextFetchLimit);
    }

    overviewPollsQuery.snapshots().listen((querySnapshot) {
      if (querySnapshot.docs.isNotEmpty) {
        for (final doc in querySnapshot.docs) {
          int index = _overviewPolls.indexWhere((d) => d.id == doc.id);
          // if already exists - update poll
          if (index >= 0) {
            _overviewPolls[index] = doc;
          }
          // if new document - add poll
          else {
            _overviewPolls.add(doc);
          }
        }

        _pollOverviewStreamController.add(_overviewPolls);
      }
    });
  }

Using orderBy

Note that if you are using orderBy on querying your snapshot, any update to the order's field will cause a reordering of the whole list of items. Moreover, in some cases, it will fetch automatically more items (since we are using a limit for each snapshot, and the list of items is reordered, some snapshots may need to fetch more items to fill its limitation).