Stream inside a Stream using Providers

Question

So I have created a BLOC structure with a Stream as given below. The Fetcher would receive changes to a list of Chatroom ids. Then using the transformer, it would add the data in the stream to a Cache map and pipe it to the output.

Now the catch here is that each Chatroom IDs will be used to create a stream instance, so subscribe to any changes in the Chatroom data. So the Cache map basically has the Chatroom ID mapped to its corresponding Stream. ChatRoomProvider is binds the bloc with the app.

   class ChatRoomBloc {
    // this is similar to the Streambuilder and Itemsbuilder we have in the Stories bloc
      final _chatroomsFetcher = PublishSubject<String>();
      final _chatroomsOutput =
          BehaviorSubject<Map<String, Observable<ChatroomModel>>>();

// Getter to Stream
  Observable<Map<String, Observable<ChatroomModel>>> get chatroomStream =>
      _chatroomsOutput.stream;

  ChatRoomBloc() {
    chatRoomPath.listen((chatrooms) => chatrooms.documents
        .forEach((f) => _chatroomsFetcher.sink.add(f.documentID)));
    _chatroomsFetcher.stream
        .transform(_chatroomsTransformer())
        .pipe(_chatroomsOutput);
  }

  ScanStreamTransformer<String, Map<String, Observable<ChatroomModel>>>
      _chatroomsTransformer() {
    return ScanStreamTransformer(
        (Map<String, Observable<ChatroomModel>> cache, String id, index) {
      // adding the iteam to cache map
      cache[id] = chatRoomInfo(id);
      print('cache ${cache.toString()}');
      return cache;
    }, <String, Observable<ChatroomModel>>{});
  }

  dispose() {
    _chatroomsFetcher.close();
    _chatroomsOutput.close();
  }
}

Observable<ChatroomModel> chatRoomInfo(String _chatrooms) {
  final _chatroomInfo = PublishSubject<ChatroomModel>();

  Firestore.instance
      .collection('chatRooms')
      .document(_chatrooms)
      .snapshots()
      .listen((chatroomInfo) =>
          _chatroomInfo.sink.add(ChatroomModel.fromJson(chatroomInfo.data)));

  dispose() {
    _chatroomInfo.close();
  }

  return _chatroomInfo.stream;
}

Then I create a Streambuilder with a List view to list the IDs and any data from their corresponding streams as given below.

class FeedList extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    final chatroomBloc = ChatRoomProvider.of(context);
    return Scaffold(
      appBar: AppBar(
        title: Text('Chat Room'),
      ),
      body: buildList(chatroomBloc),
    );
  }
  Widget buildList(ChatRoomBloc chatroomBloc) {
    return StreamBuilder(
        // Stream only top ids to display
        stream: chatroomBloc.chatroomStream,
        builder: (context,
            AsyncSnapshot<Map<String, Observable<ChatroomModel>>> snapshot) {
          if (!snapshot.hasData) { // no data yet
            return Center(child: CircularProgressIndicator());
          }
          return ListView.builder(
            itemCount: snapshot.data.length,
            itemBuilder: (context, int index) {
              print('index $index and ${snapshot.data}');
              return buildTile(snapshot.data[index]);
            },
          );
        });
  }

  Widget buildTile(Observable<ChatroomModel> chatroomInfoStream) {
    return StreamBuilder(
        stream: chatroomInfoStream,
        builder: (context, AsyncSnapshot<ChatroomModel> chatroomSnapshot) {
          if (!chatroomSnapshot.hasData) {
            return Center(
              child: CircularProgressIndicator(),
            );
          }
          print('${chatroomSnapshot.data.name}');
          print('${chatroomSnapshot.data.members.toString()}');
          return Column(children: [
            ListTile(
              title: Text('${chatroomSnapshot.data.name}'),
              trailing: Column(
                children: <Widget>[
                  Icon(Icons.comment),
                ],
              ),
            ),
            Divider(
              height: 8.0,
            ),
          ]);
        });
  }
}

The output I am getting is given below. The Streambuilder is stuck at CircularProgressIndicator in the buildTile method. I think it means that the instances are getting created and added in the cache map, but they are lot listening to the right instances or there is something wrong in the way I wired up the streams. Can you please help ?

I/flutter (12856): cache {H8j0EHhu2QpicgFDGXYZ: Instance of 'PublishSubject<ChatroomModel>'} 
I/flutter (12856): cache {H8j0EHhu2QpicgFDGXYZ: Instance of 'PublishSubject<ChatroomModel>', QAhKYk1cfoq8N8O6WY2N: Instance of 'PublishSubject<ChatroomModel>'} 
I/flutter (12856): index 0 and {H8j0EHhu2QpicgFDGXYZ: Instance of 'PublishSubject<ChatroomModel>', QAhKYk1cfoq8N8O6WY2N: Instance of 'PublishSubject<ChatroomModel>'} 
I/flutter (12856): index 1 and {H8j0EHhu2QpicgFDGXYZ: Instance of 'PublishSubject<ChatroomModel>', QAhKYk1cfoq8N8O6WY2N: Instance of 'PublishSubject<ChatroomModel>'}

Solution

As a quick fix, maybe try:

final _chatroomInfo = BehaviorSubject<ChatroomModel>();

---

On a second note:

The code in its current state is hard to read and understand, it's unmaintainable and inefficient. I'm not sure what you are actually trying to do.

It's a bad idea to nest StreamBuilders. It will delay the display of the chat list by at least 2 frames, because every StreamBuilder renders at least one empty frame (data = null).

Listening to a stream and feeding the result into a Subject will also add delays.

If possible, try to remove all subjects. Instead, use rx operators.

The BLoC should provide a single output stream that provides all the data that is required to render the chat list.

Comments

[vzurd]

The quick fix did not work. Thanks @boformer. I will try to make int into one output stream as you suggested.