Поток внутри потока с использованием провайдеров

Итак, я создал структуру BLOC с потоком, как показано ниже. Сборщик получит изменения в списке идентификаторов чата. Затем с помощью преобразователя он добавлял данные из потока в карту кэша и передавал их на выход.

Теперь загвоздка в том, что каждый идентификатор чата будет использоваться для создания экземпляра потока, поэтому подпишитесь на любые изменения в данных чата. Таким образом, карта кэша в основном имеет идентификатор чата, сопоставленный с соответствующим потоком. ChatRoomProvider связывает блок с приложением.

   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;
}

Затем я создаю StreamBuilder с представлением списка, чтобы перечислить идентификаторы и любые данные из соответствующих потоков, как указано ниже.

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,
            ),
          ]);
        });
  }
}

Результат, который я получаю, приведен ниже. Streambuilder застрял в CircularProgressIndicator в методе buildTile. Я думаю, это означает, что экземпляры создаются и добавляются в карту кеша, но они много слушают нужные экземпляры, или что-то не так в том, как я подключил потоки. Не могли бы вы помочь?

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>'}