firebase - Merging streams together to create paginated ListView

Question

I'm building a ListView containing a list of Hour items stored in Firestore. I was able to create it with StreamBuilder, however given the fact that Hours collections contains thousands of documents, it is necessary to introduce pagination to reduce unnecessary reads.

My idea is to create a stream (mainStream) with queried items for the last 24 hours. If user scrolls down 20 items, I will expand the mainStream with 20 additional elements by merging it with a new stream. However, it doesn't seem to work - only first 24 items are in the stream, merging doesn't seem to have any impact.

Why I want to have paginated Stream, not a list? Hour items can be changed on other devices and I need streams to notice changes.

I'm stuck with it for a more than a day and other questions do not help - any help is much appreciated.

class TodayPage extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('Today'),
      ),
      body: _buildContents(context),
    );
  }

  Widget _buildContents(BuildContext context) {
    final database = Provider.of<Database>(context, listen: true);

    DateTime startingHour =
    dateHourFromDate(DateTime.now()).subtract(Duration(hours: 24));
    DateTime beforeHour;

    // Stream with items for last 24 hours
    // hoursStream returns Stream<List<Hour>> from startingHour till beforeHour
    Stream mainStream = database.hoursStream(startingHour: startingHour.toString());

    return StreamBuilder<List<Hour>>(
      stream: mainStream,
      builder: (context, snapshot) {
        if (snapshot.data != null) {

          final List<Hour> hours = snapshot.data;
          final allHours = hours.map((hour) => hour.id).toList();

          return ListView.builder(

              // chacheExtent to make sure that itemBuilder will not build items for which we have not yet fetched data in the new stream
              cacheExtent: 5,
              itemBuilder: (context, index) {

                // if itemBuilder is reaching till the end of items in mainStream, extend mainStream by items for the following 24 hours
                if (index % 20 == 0) {
                  beforeHour = startingHour;
                  startingHour = startingHour.subtract(Duration(hours: 20));

                  // doesn't seem to work - snapshot contains elements only from the original mainStream
                  mainStream.mergeWith([
                    database.hoursStream(
                        beforeHour: beforeHour.toString(),
                        startingHour: startingHour.toString())
                  ]);
                }

                return Container(); // placeholder for a widget with content based on allHours

              });
        } else {
          return Center(child: CircularProgressIndicator());
        }
      },
    );
  }
}

Implementation of hoursStream:

  Stream<List<Hour>> hoursStream({
    String startingHour,
    String beforeHour,
  }) =>
      _service.collectionStream(
          path: APIPath.hours(uid),
          builder: (data, documentId) => Hour.fromMap(data, documentId),
          queryBuilder: (query) => query
              .where('id', isGreaterThanOrEqualTo: startingHour)
              .where('id', isLessThan: beforeHour));

and finally collectionStream

  Stream<List<T>> collectionStream<T>({
    @required String path,
    @required T Function(Map<String, dynamic> data, String documentId) builder,
    Query Function(Query query) queryBuilder,
    int Function(T lhs, T rhs) sort,
  }) {
    Query query = FirebaseFirestore.instance.collection(path);
    if (queryBuilder != null) {
      query = queryBuilder(query);
    }
    final snapshots = query.snapshots();
    return snapshots.map((snapshot) {
      final result = snapshot.docs
          .map((snapshot) => builder(snapshot.data(), snapshot.id))
          .where((value) => value != null)
          .toList();
      if (sort != null) {
        result.sort(sort);
      }
      return result;
    });
  }

Answer 1

As far as I know, Stream does not have a mergeWith method, so I don't know if it's an extension you created or from library, but one thing you are forgetting is the setState.

if (index % 20 == 0) {
  beforeHour = startingHour;
  startingHour = startingHour.subtract(Duration(hours: 20));

  setState(() {
    mainStream.mergeWith([
      database.hoursStream(
          beforeHour: beforeHour.toString(),
          startingHour: startingHour.toString())
    ]);
  });
}

By the way, I like what you are doing with your collectionStream method. Very beautiful coding.