How to await inside a stream while querying data from Firebase firestore

Issue

For context I'm using Getx state management for flutter and i need to call list.bindStream(availabilityStream()) on my Rx<List<Availability>> object. here is my availabilityStream method

  static Stream<List<Availability>> availabilityStream() {
    return FirebaseFirestore.instance
        .collection('availability')
        .where('language',
            isEqualTo: GetStorageController.instance.language.value)
        .snapshots()
        .map((QuerySnapshot query) {
      List<Availability> results = [];
      for (var availablity in query.docs) {
        availablity["cluster"].get().then((DocumentSnapshot document) {
          if (document.exists) {
            print("Just reached here!");
            //! Ignore doc if cluster link is broken
            final model = Availability.fromDocumentSnapshot(
                availabilityData: availablity, clusterData: document);
            results.add(model);
          }
        });
      }
      print("result returned");
      return results;
    });
  } 

the cluster field on my availability collection is a reference field to another collection. The problem here is i need to await the .get() call to my firestore or the function returns before the data gets returned. I can't await inside the map function or the return type of Stream<List> changes. so how can i await my function call here?

Solution

using the advice i got from the comments I've used Stream.asyncMap to wait for all my network call futures to complete.

Here is my updated Repository

class AvailabilityRepository {
  static Future<Availability> getAvailabilityAndCluster(
      QueryDocumentSnapshot availability) async {
    return await availability["cluster"]
        .get()
        .then((DocumentSnapshot document) {
      if (document.exists) {
        //! Ignore doc if cluster link is broken
        final model = Availability.fromDocumentSnapshot(
            availabilityData: availability, clusterData: document);
        return model;
      }
    });
  }


  static Stream<List<Availability>> availabilityStream() {
    return FirebaseFirestore.instance
        .collection('availability')
        .where('language',
            isEqualTo: GetStorageController.instance.language.value)
        .snapshots()
        .asyncMap((snapshot) => Future.wait(
            snapshot.docs.map((e) => getAvailabilityAndCluster(e))));
  }
}

How i think this works is that the normal .map function returns multiple promises form the getAvailabilityAndCluster() method then all of the processes that execute asynchronously are all put to Future.wait() which is one big promise that waits all the promises inside it to complete. Then this is passed onto .asyncMap() which waits for the Future.wait() to complete before continuing with its result.

Answered By - m.prime