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.

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

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 :)