How exactly to merge multiple streams in firebase firestore

Question

Before you say this is a duplicate question or that I should use nested stream builders, please hear me out.

I am designing a social media type application. And I want users to receive updates whenever someone they are following posts something in their "my Followers Posts" collection. In the app, the app will check firebase rtdb for the current user's following list(the people he is following) and make a list of their uids.

I plan on using said list to create a list of streams (ordered by time, of course) and merge them into one stream that shall then be fed into a stream builder on the private feed page.

On this page, the user will be able to easily follow what their people of interest have been posting.

I figured such a system is a lot more cost efficient than every user having a document in the "private Feed" collection and whenever someone posts something, the app reads their list of followers and then promptly posts an update in each and every one of their private feeds. Because... Picture someone with 2 million followers. That's 2 million writes instantly. And later on, 2 million reads. I figured it's a lot more cost efficient for the poster to just put the post in their "publicFeed" and the different followers simply listen in onto that feed and keep up to tabs with them.

But.. This requires implementing a merging of multiple streams (more than 2). How do I do this?

I have tried reading into RxDart but it is total Greek to me. I am relatively a beginner in dart. I've only been coding for about 5 months now.

Manyplies

Answer 1

You can use StreamGroup from the async package: https://pub.dev/documentation/async/latest/async/StreamGroup-class.html to group events from multiple streams - it's well documented and maintained by the dart team. If you have no RxDart experience this is a good choice. It does not have all the features of rx but for a beginner it should be easier to wrap your head around this

Theosophy

Could you please give me a simple example on how to do this. I am learning completely by youtube tutorial, so, I'm not yet that good at decoding the way documentations are written. I learn better from examples. Assume you have data at the new Cars collection and some other data at the newClothesCollection and some other data at the newKeyBoards collection. Could you please write some demo code on how you would use async to merge these three and feed them into a stream builder? – Manyplies

I have managed to merge the streams using this and it actually works. However, i am facing a new and entirely different issue regarding displaying the merged stream in a stream builder. I am going to post it as a new question though. Thank you. – Manyplies

Answer 2

I recently had a similar case, and what I suggest You to do is this (I'm using cloud firestore, but I'm sure You have your streams already written so the important part is the usage of multiple streams):

You have to add this plugin to pub spec.yaml: https://pub.dev/packages/rxdart

Here is the repository for (in your case posts, let's say newPosts, oldPosts):

class PostRepository {

  static CollectionReference get collection => yourCollectionRef;
    
  static Stream<List<Post>> newPosts() {
    Query query = collection
        .where('Your condition like was viewed', isEqualTo: false)
        .orderBy('updateDate', descending: true)
        .limit(50);
    return query.snapshots().map<List<Post>>((querySnapshot) {
      final _newPosts = querySnapshot.documents.map((doc) {
        final post = Post.fromDoc(doc);
        return post;
      }).where((p) => p != null);

      return _newPosts
    });
  }

  static Stream<List<Post>> oldPosts() {
    Query query = collection
        .where('Your condition like was viewed', isEqualTo: true)
        .orderBy('updateDate', descending: true)
        .limit(50);
    return query.snapshots().map<List<Post>>((querySnapshot) {
      final _oldPosts = querySnapshot.documents.map((doc) {
        final post = Post.fromDoc(doc);
        return post;
      }).where((p) => p != null);

      return _oldPosts
    });
  }
}

Then to get multiple streams (those two from above combined), do like this in your widget class:

IMPORTANT! you have to import this - import 'package:rxdart/streams.dart';

List<Post> newPosts;
List<Post> oldPosts;

Widget _pageContent() {
  return SingleChildScrollView(
    child: Column(
      children: [
        ListView.builder(
          shrinkWrap: true,
          physics: NeverScrollableScrollPhysics(),
          itemCount: newPosts.length,
          itemBuilder: (context, index) {
            return ListTile(
              title: Text(newPosts[index].title)
            );
          }
        ),
        ListView.builder(
          shrinkWrap: true,
          physics: NeverScrollableScrollPhysics(),
          itemCount: oldPosts.length,
          itemBuilder: (context, index) {
            return ListTile(
              title: Text(oldPosts[index].title)
            );
          }
        )
      ]
    )
  );
}

Widget _posts() {
  return StreamBuilder(
    stream: CombineLatestStream.list([
      PostRepository.getNewPosts(),
      PostRepository.getOldPosts()
    ]),
    builder: (context, snap) {
     if (snap.hasError) {

        debugPrint('${snap.error}');
        return ErrorContent(snap.error);

      } else if (!snap.hasData) {

        return Center(
          child: CircularProgressIndicator(),
        );

      }

      newPosts = snap.data[0];
      oldPosts = snap.data[1];

      return _pageContent();
    }
  );
}

I wrote the code like from head, so there may be some small errors, but I hope You get the point, enjoy :)

Cence

Thank you a lot. Though I don't think it's possible to put a list view inside a scroll view. I've done that before and it gives some error along the lines of "you can't put a scrollable inside another scrollable, dummy". I am going to try out your solution though and I'll give you feedback. – Manyplies

Ok. If u use shrinkwrap and neverscrollablescrollphysics as in example your 2 lists should behave perfectly as U want :) – Cence

Thank you a lot. I managed to merge the streams using stream group, but im currently facing a new issue with displaying this merged stream in a stream builder. But im posting it as an entirely new question. Thank you tho. – Manyplies

Notice that if you want to combine collection streams (Stream<List<T>> and not a Stream<T>, you need to use a custom combiner function - in this example he calls CombineLatestStream.list which makes use of a default combiner which is good for Stream<T>. In my case I couldn't make it work until I realized when using Stream<List<T>> you need to provide a custom combiner to handle it (get a List<List<T>> then flatten it to get a Stream<List<T>> as required). So you need to use the regular CombineLatestStream(streamList, combinerFunc). – Mccabe