Combine streams from Firestore in flutter

Question

I have been trying to listen to more than one collection from Firestone using a StreamBuilder or something similar. My original code when I was working with only one Stream was:

import 'package:flutter/cupertino.dart';
import 'package:flutter/material.dart';
import 'package:cloud_firestore/cloud_firestore.dart';

class List extends StatefulWidget{

  ///The reference to the collection is like
  ///Firestore.instance.collection("users").document(firebaseUser.uid).collection("list1").reference()
  final CollectionReference listReference;

  List(this.listReference);

  @override
  State createState() => new ListState();
}

class ListState extends State<List> {

  @override
  Widget build(BuildContext context){

    return new StreamBuilder(
        stream: widget.listReference.snapshots(),
        builder: (context, snapshot) {
          return new ListView.builder(
              itemCount: snapshot.data.documents.length,
              padding: const EdgeInsets.only(top: 2.0),
              itemExtent: 130.0,
              itemBuilder: (context, index) {
                DocumentSnapshot ds = snapshot.data.documents[index];
                return new Data(ds);
              }
          );
        });
  }
}

This code works fine, but now I want to listen to more than one collection. I have come across a solution that doesn't involve a StreamBuilder and works with a dynamic list. My code now looks like this:

import 'dart:async';
import 'package:flutter/cupertino.dart';
import 'package:flutter/material.dart';
import 'package:cloud_firestore/cloud_firestore.dart';
import 'main.dart';
import 'package:async/async.dart';

class ListHandler extends StatefulWidget{

  final CollectionReference listReference;

  ListHandler(this.listReference);

  @override
  State createState() => new ListHandlerState();
}

class ListHandlerState extends State<ListHandler> {

  StreamController streamController;
  List<dynamic> dataList = [];

  @override
  void initState() {
    streamController = StreamController.broadcast();
    setupData();
    super.initState();
  }

  @override
  void dispose() {
    super.dispose();
    streamController?.close();
    streamController = null;
  }

  Future<Stream> getData() async{
      Stream stream1 = Firestore.instance.collection("users").document(firebaseUser.uid).collection("list1").snapshots();
      Stream stream2 = Firestore.instance.collection("users").document(firebaseUser.uid).collection("list2").snapshots();

      return StreamZip(([stream1, stream2])).asBroadcastStream();
  }

  setupData() async {
    Stream stream = await getData()..asBroadcastStream();
    stream.listen((snapshot) {
      setState(() {
        //Empty the list to avoid repetitions when the users updates the 
        //data in the snapshot
        dataList =[];
        List<DocumentSnapshot> list;
        for(int i=0; i < snapshot.length; i++){
          list = snapshot[i].documents;
          for (var item in list){
            dataList.add(item);
          }
        }
      });
    });
  }

  @override
  Widget build(BuildContext context){
    if(dataList.length == 0){
      return new Text("No data found");
    }

    return new ListView.builder(
        itemCount: dataList.length,
        padding: const EdgeInsets.only(top: 2.0),
        itemBuilder: (context, index) {
          DocumentSnapshot ds = dataList[index];
          return new Data(ds['title']);
        }
    );
  }
}

The thing is that the ListView returns Data that is a StatefulWidget and the user can interact with it making the data change in Firestore making the next error appear:

[VERBOSE-2:dart_error.cc(16)] Unhandled exception:
setState() called after dispose(): ListHandlerState#81967(lifecycle state: defunct, not mounted)
This error happens if you call setState() on a State object for a widget that no longer appears in the widget tree (e.g., whose parent widget no longer includes the widget in its build). This error can occur when code calls setState() from a timer or an animation callback. The preferred solution is to cancel the timer or stop listening to the animation in the dispose() callback. Another solution is to check the "mounted" property of this object before calling setState() to ensure the object is still in the tree.
This error might indicate a memory leak if setState() is being called because another object is retaining a reference to this State object after it has been removed from the tree. To avoid memory leaks, consider breaking the reference to this object during dispose().

The app does not crash, and it does what is expected but it always shows this error.

Some people use the library rxdart to work with streams and I have tried doing something like the code below but when I put it in the StreamBuilder only elements from on of the :

import 'dart:async';
import 'package:flutter/cupertino.dart';
import 'package:flutter/material.dart';
import 'package:cloud_firestore/cloud_firestore.dart';
import 'main.dart';
import 'showInfo.dart';
import 'package:rxdart/rxdart.dart';

class ListHandler extends StatefulWidget{

  @override
  State createState() => new ListHandlerState();
}

class ListHandlerState extends State<ListHandler> {

  Stream getData() {
    Stream stream1 = Firestore.instance.collection("users").document(firebaseUser.uid).collection("list1").snapshots();
    Stream stream2 = Firestore.instance.collection("users").document(firebaseUser.uid).collection("list2").snapshots();

    return Observable.merge(([stream2, stream1]));
  }

  @override
  Widget build(BuildContext context){
    return new StreamBuilder(
        stream: getData(),
        builder: (context, snapshot) {
          if(!snapshot.hasData){
            print(snapshot);
            return new Text("loading");
          }
          return new ListView.builder(
              itemCount: snapshot.data.documents.length,
              padding: const EdgeInsets.only(top: 2.0),
              itemBuilder: (context, index) {
                DocumentSnapshot ds = snapshot.data.documents[index];
                return new Data(ds);
              }
          );
        });
  }
}

This is my first time working with Streams and I don't understand them quite well and I would like your thoughts on what to do.

Answer 1

Use CombineLatestStream from rxdart to combine streams.

The StreamBuilder will build every time one of the streams emits a new event.

The result snapshot.data is a list of the latest element per stream.

Example:

StreamBuilder(
stream: CombineLatestStream.list([
  stream0,
  stream1,
]),
builder: (context, snapshot) {
  final data0 = snapshot.data[0];
  final data1 = snapshot.data[1];
})

Answer 2

You might want to try concatWith from the rxDart package:

Returns a Stream that emits all items from the current Stream, then emits all items from the given streams, one after the next.

https://pub.dev/packages/rxdart

import 'package:rxdart/rxdart.dart';

      Stream.fromIterable(['a', 'b', 'c']).concatWith([
        Stream.fromIterable(['d', 'e', 'f'])
      ]).listen(print);

This will print:

a

b

c

d

e

f

Answer 3

The last example should work. Perhaps the 2nd stream didn't emit any values during the time you observed the stream.

StreamGroup.merge() from the async package should also work.

StreamZip creates pairs of values one of each stream. When they emit values at a different rate, then one stream waits with emitting until the other emits a value. This is probably not what you want.

Answer 4

The problem is not in the merging, but in the StreamBuilder updating the UI based on the LATEST snapshot, in other words it doesn't stack snapshots it just picks up that last emitted an event, in other words the streams are merged and the merged stream does contain the data of all merged streams, however the streamBuilder will only show the very Last stream emitted event, a work around is this:

StreamBuilder<List<QuerySnapshot>>(stream: streamGroup, builder: (BuildContext context, 
    AsyncSnapshot<List<QuerySnapshot>> snapshotList){
                  if(!snapshotList.hasData){
                    return MyLoadingWidget();
                  }
                  // note that snapshotList.data is the actual list of querysnapshots, snapshotList alone is just an AsyncSnapshot

                  int lengthOfDocs=0;
                  int querySnapShotCounter = 0;
                  snapshotList.data.forEach((snap){lengthOfDocs = lengthOfDocs + snap.documents.length;});
                  int counter = 0;
                  return ListView.builder(
                    itemCount: lengthOfDocs,
                    itemBuilder: (_,int index){
                      try{DocumentSnapshot doc = snapshotList.data[querySnapShotCounter].documents[counter];
                      counter = counter + 1 ;
                       return new Container(child: Text(doc.data["name"]));
                      }
                      catch(RangeError){
                        querySnapShotCounter = querySnapShotCounter+1;
                        counter = 0;
                        DocumentSnapshot doc = snapshotList.data[querySnapShotCounter].documents[counter];
                        counter = counter + 1 ;
                         return new Container(child: Text(doc.data["name"]));
                      }

                    },
                  );
                },