Flutter merge two firestore streams into a single stream

Question

I simply want to perform an 'OR' operation and get the both results of two queries into one stream.

Here's my code with a single stream

 StreamBuilder(
    stream: Firestore.instance
        .collection('list')
        .where('id', isEqualTo: 'false')
        .orderBy('timestamp')
        .snapshots(),
    builder: (context, snapshot) {
      if (!snapshot.hasData)
        return Column(
          mainAxisAlignment: MainAxisAlignment.center,
          children: <Widget>[
            Center(
              child: CircularProgressIndicator(),
            )
          ],
        );
      if (snapshot.data.documents.length == 0)
        return const Center(
          child: Text(
            "Not Available",
            style: TextStyle(fontSize: 30.0, color: Colors.grey),
          ),
        );
      return ListView.builder(
        padding: EdgeInsets.all(5.0),
        key: Key(randomString(20)),
        itemCount: snapshot.data.documents.length,
        itemBuilder: (BuildContext context, int index) {
          return ListCard(snapshot.data.documents[index]);
        },
      );
    }),

Instead of a single stream now I want to feed two stream to the same stream builder.

I tried StreamGroup but it's not working since Widgets rebuild

StreamGroup.merge([streamOne, streamTwo]).asBroadcastStream();

I tried followed method also

 Stream<List<DocumentSnapshot>> searchResult()  {
List<Stream<List<DocumentSnapshot>>> streamList = [];

Firestore.instance
    .collection('room-list')
    .where('id', isEqualTo: 'false')
    .snapshots()
    .forEach((snap) {
  streamList.add(Observable.just(snap.documents));
});

Firestore.instance
    .collection('room-list')
    .where('id', isEqualTo: 'pending')
    .snapshots()
    .forEach((snap) {
  streamList.add(Observable.just(snap.documents));
});

var x = Observable.merge(streamList)
    .scan<List<DocumentSnapshot>>((acc, curr, i) {
  return acc ?? <DocumentSnapshot>[]
    ..addAll(curr);
});
return x;
}

Here I get the error there should be at least a single stream to merge. Its because Observable.merge(streamList) is called before items are added to streamList.

I simply want to get the both results of two queries into one stream.

Solution 1

This should work.

//Change your streams here
    Stream<List<QuerySnapshot>> getData() {
        Stream stream1 = Firestore.instance.collection('list').where('id', isEqualTo: 'false').orderBy('timestamp').snapshots();
        Stream stream2 = Firestore.instance.collection('list').where('id', isEqualTo: 'true').orderBy('timestamp').snapshots();
        return StreamZip([stream1, stream2]);
      }


  @override
  Widget build(BuildContext context) {
    return new Scaffold(
      body: StreamBuilder(
          stream: getData(),
          builder: (BuildContext context, AsyncSnapshot<List<QuerySnapshot>> snapshot1) {

            List<QuerySnapshot> querySnapshotData =  snapshot1.data.toList();

            //copy document snapshots from second stream to first so querySnapshotData[0].documents will have all documents from both query snapshots
            querySnapshotData[0].documents.addAll(querySnapshotData[1].documents);

            if (querySnapshotData[0].documents.isEmpty)
              return Column(
                mainAxisAlignment: MainAxisAlignment.center,
                children: <Widget>[
                  Center(
                    child: CircularProgressIndicator(),
                  )
                ],
              );
            if (querySnapshotData[0].documents.length == 0)
              return const Center(
                child: Text(
                  "Not Available",
                  style: TextStyle(fontSize: 30.0, color: Colors.grey),
                ),
              );

            return new ListView(
                children: querySnapshotData[0].documents.map((DocumentSnapshot document){
                 // put your logic here. You will have access to document from both streams as "document" here
                  return new ListCard(document);
                }).toList()
            );
          }
      ),
    );
  }

Hope this helps!!!

Solution 2

I’m not sure why you’re using forEach and Observable.just().

You can just merge two firestore streams directly like:

Observable.merge([stream1, stream2]).pipe(combineStream);

Wherre stream1/2 is just your firestore snapshot.

Solution 3

Well I am late, but just gonna put it out there.

You can add whereIn clause in your query like this:

Firestore.instance.collection("collection_name").where("field",whereIn:["false","true"]).snapshots();

Solution 4

I used RxDart package to combine two streams as shown below

RxDart - CombineLatestStream

final Stream<DocumentSnapshot> user = Firestore.instance
        .collection("users")
        .document(firebaseUser.uid)
        .snapshots();

    final Stream<QuerySnapshot> cards =
        Firestore.instance.collection("cards").snapshots();

    CombineLatestStream.list([user, cards]).listen((data) {
      add(LoadedHomeEvent(
        data.elementAt(0),
        data.elementAt(1),
      ));
    });

View more solutions

Comments

[Praneeth Dhanushka Fernando]

From where does combineStream come?

[Jason Scott]

Just create a new PublishSubject or StreamContoller for combineStream.

[Praneeth Dhanushka Fernando]

I tried your suggestion and it failed. The issue since stream1 is a firestore querysnapshot when I tried to call merge it says i'm trying to merge an empty stream. It is because stream1 and stream2 are not initialized yet as they are waiting for firestore. How can i make sure stream1 and sream2 are initialized before called merge?

[m-bhole]

@PraneethDhanushkaFernando please try this and let me know if it worked. Thanks

[Praneeth Dhanushka Fernando]

StreamZip cannot be used for this bro. It used to combined streams that belongs with each other like Title and Post

[m-bhole]

in that case you will need nested streambuilder

[Chichebe]

This worked for my particular case. I've been at it all day still I stumbled on your answer. Thanks @m-bhole, from the Future :)

[lenz]

make sure to add async to your dependencies

[Hassan Ansari]

This is Not Working at all!

[Abhishek Kumar]

They may have changed the methods.

[Kamlesh]

You have used CombineLatestStream.... then what is further code? Kindly share full of code so that other users and I can understand what is happening? Its just a request :) Thanks.

[vertigo71]

It has not worked for me. When there is another event in one of the Streams, StreamBuider was not called. What worked for me: github.com/flutter/flutter/issues/23173 Rx.merge + scan