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.

Do you want to merge the two streams or do you only want to re-render when one of the streams emits a new value? — Günter Zöchbauer, Jul 06 '18 at 16:24
I want that my two streams act like one but getting information from two or more collections of Firebase (I asume I want to merge the two streams). — Roger Bosch Mateo, Jul 07 '18 at 09:28
@RogerBoschMateo Did you find any solution to merge 2 streams into 1 stream. Kindly share. Thanks. — Kamlesh, Jun 16 '21 at 09:24

score 25 · Answer 1 · answered May 04 '20 at 16:32

25

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];
})

answered May 04 '20 at 16:32

Erlend

1,711
1
22
25

1

This is the actual working answer. Upvoted – Watery Desert Jan 09 '21 at 16:25
@erlend Erlend - I have used it but getting error "type 'CombineLatestStream>, List>>>' is not a subtype of type 'Stream>?'" Kindly suggest, how to fix it. Thanks. – Kamlesh Jun 16 '21 at 09:34
same as I. did you fix it @erlend – Luqman Tuke Mar 03 '22 at 18:31
Getting the error: `The operator '[]' isn't defined for the type 'Object'. Try defining the operator '[]'` Which is weird because when I print out `snapshot.data` I get `[Instance of '_JsonQuerySnapshot', Instance of '_JsonQuerySnapshot']` Thoughts? – tazboy Mar 06 '22 at 15:08
I think you need to define in either `StreamBuilder` or `CombineLatestStream` what type of data you'd expect, or cast the data to the correct type. – Erlend Mar 06 '22 at 23:14
Worked for me using Firebase RTDB, trying to access multiple streams and their data – Tomas Ward Apr 28 '22 at 17:16

score 8 · Answer 2 · answered Jul 06 '18 at 16:23

8

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.

answered Jul 06 '18 at 16:23

Günter Zöchbauer

623,577
216
2,003
1,567

2

I edited a little bit the function getData(). What I get returning `Observable.merge(([stream1, stream2]))` is all the data from list2, but if I return `Observable.merge(([stream2, stream1]))` I get the data from list1 in the `StreamBuilder`. – Roger Bosch Mateo Jul 07 '18 at 08:53
1

If I use StreamGroup.merge(([stream2, stream1])) the same thing happens – Roger Bosch Mateo Jul 07 '18 at 10:45
That doesn't sound kike caused by the merging of the streams. – Günter Zöchbauer Jul 08 '18 at 12:10
Do you have some working code or some other tips? I can’t get it to work – Roger Bosch Mateo Jul 08 '18 at 17:08
I have the same issue as @RogerBoschMateo describes. Did you figure it out? – Ronen Rabinovici Oct 16 '18 at 10:41
Key phrase in the answer - "during the time you observed the stream". If these are broadcast streams and you merge them you'll still only see the events emitted during the time you are listening. – Nate Bosch Oct 16 '18 at 21:36

score 4 · Accepted Answer · answered Oct 18 '18 at 13:28

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"]));
                      }

                    },
                  );
                },

Thank you! I can't test the solution as I'm no longer working in that personal project, but at the end I came to the same conclusion as you said: the streamBuilder will only show the very Last stream emitted event. — Roger Bosch Mateo, Jan 22 '19 at 11:42

Scorb · Answer 4 · 2023-06-12T21:10:58.103

If you don't want a third party package, this logic is not too complex.

This method will combine N number of List streams into one list with the latest values.

Stream<List<T>> combineListStreams<T>(List<Stream<List<T>>> streams) {
  var controller = StreamController<List<T>>();
  Set activeStreams = {};
  Map<Stream<List<T>> , List<T>> lastValues = {};
  List<StreamSubscription> subscriptions = [];

  for (var stream in streams) {
    activeStreams.add(stream);
    var subscription = stream.listen(
      (val) {
        lastValues[stream] = val;
        List<T> out = [];
        for (var list in lastValues.values) {
          out.addAll(list);
        }
        controller.add(out);
      },
      onDone: () {
        activeStreams.remove(stream);
        if (activeStreams.isEmpty) {
          controller.close();
        }
      }
    );
    subscriptions.add(subscription);
  }
  controller.onCancel = () {
    for (var subscription in subscriptions) {
      subscription.cancel();
    }
  };
  return controller.stream;
}

Any way you could make this work for a Firestore stream? Each stream is a Stream>>. I tried changing some stuff up but couldn't get it to work. — tazboy, Mar 06 '22 at 15:01

live-love · Answer 5 · 2020-05-11T17:11:37.337

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

Combine streams from Firestore in flutter

5 Answers5

Linked