如何使用 2 个流，其中流 2 取决于流 1？答案

【问题标题】：How can I work with 2 streams where stream two is depending on stream one?如何使用 2 个流，其中流 2 取决于流 1？
【发布时间】：2020-08-21 01:53:06
【问题描述】：

我有 2 个流

  Stream<List<GroupModel>> groupStream() {
    final CollectionReference groupCollection = fireStore.collection('groups');

    return groupCollection
        .where('members.${user.id}', isEqualTo: true)
        .snapshots()
        .map((snapshot) => snapshot.documents.map((document) => GroupModel.fromFirestore(document)).toList());
  }

  Stream<List<GroupSubscriptionModel>> groupSubscriptionsStream(List<GroupModel> groups) {
    final chunks = chunkSizeCollection(groups, 10);
    List<Stream<List<GroupSubscriptionModel>>> streams = List<Stream<List<GroupSubscriptionModel>>>();
    chunks.forEach((chunck) => streams.add(fireStore
        .collection('users/${userID}/userGroupSubscriptions')
        .where(FieldPath.documentId, whereIn: chunck)
        .snapshots()
        .map((snapshot) =>
            snapshot.documents.map((document) => GroupSubscriptionModel.fromFirestore(document)).toList())));
    return ZipStream(streams, (value) => value.last);
  }

我想要的是获取所有组，然后从用户那里获取订阅，这基本上说明用户是否订阅了该组。然后在我的 UI 中，我根据用户是否订阅来切换一个图标。

我的问题是我的groupSubscriptionsStream 依赖于组中的 id 来获取相应的文档。因为我不能只流式传输集合中的所有文档。如果可以的话，我没有任何问题。

我正在使用带有提供程序的 bloc 来为我的小部件提供流控制器。但是我的StreamSubscriptionsBloc 需要List<GroupModel> 才能将流添加到它的控制器

class GroupSubscriptionsBloc{

  final StreamController<List<GroupSubscriptionModel>> _controller = StreamController<List<GroupSubscriptionModel>>();
  Stream<List<GroupSubscriptionModel>> get stream => _controller.stream;

  GroupSubscriptionsBloc(DatabaseService database, List<GroupModel> groups)
  {
    _controller.addStream(database.groupSubscriptionsStream(groups));
  }

  void dispose(){
    _controller.close();
  }
}

所以在小部件中，我有一个静态方法，可为小部件提供两个块

  static Widget create(BuildContext context) {
    final database = Provider.of<DatabaseService>(context);
    return MultiProvider(
      providers: [
        Provider(
          create: (_) => GroupBloc(database),
          dispose: (BuildContext context, GroupBloc bloc) => bloc.dispose(),
        ),
        Provider(
          create: (_) => GroupSubscriptionsBloc(database, null),
          dispose: (BuildContext context, GroupSubscriptionsBloc bloc) => bloc.dispose(),
        ),
      ],
      child: TheGroupOverviewPage(),
    );
  }

但正如您所见，我目前将 null 传递给订阅集团，因为我没有组。所以我无法将流添加到他的控制器。

但我确实想在 TheGroupOverviewPage 小部件中使用这两个块，因为在不同的小部件中这样做是没有意义的。

所以问题是我如何获得List<GroupModel>？

问题是，如果我能以某种方式组合两个流并将它们映射到我的 GroupModel 以便我将 isSubscribed 切换为 true，我什至不需要两个流。

class GroupModel
{
  final String _id;
  final String _displayName;
  final Map<String, bool> _members;
  final Map<String, bool> _admins;
  bool isSubscribed = false;

  String get id => _id;
  String get displayName => _displayName;
  Map<String, bool> get members => _members;
  Map<String, bool> get admins => _admins;

  GroupModel._internal(this._id, this._displayName, this._members, this._admins);

  factory GroupModel.fromFirestore(DocumentSnapshot document)
  {
    return GroupModel._internal(
      document.documentID ?? '',
      document['displayName'] ?? '',
      document['members'] != null ? Map.from(document['members']) :  Map<String,bool>(),
      document['admins'] != null ? Map.from(document['admins']) : Map<String,bool>(),
    );
  }
}

我知道您可以组合流，但在这种情况下，第二个流取决于第一个流。那么这甚至是一种选择吗？这将是最有意义的，因为它将输出List<GroupModel>，其中 isSubscribed 由第二个流设置。而且我可以将所有组组相关的内容保存在一个流中。

这就是我目前在TheGroupOverview 小部件中构建流的方式

    body: StreamBuilder(
      initialData: List<GroupModel>(),
      stream: groupBloc.groupStream,
      builder: (BuildContext context, AsyncSnapshot<List<GroupModel>> groupsSnapshot) {
        return ConditionalWidget(
          condition: groupsSnapshot.connectionState == ConnectionState.active,
          widgetTrue: StreamBuilder(
            initialData: List<GroupSubscriptionModel>(),
            stream: groupSubscriptionsBloc.stream,
            builder: (BuildContext context, AsyncSnapshot<List<GroupSubscriptionModel>> subscriptionSnapshot) {

              groupsSnapshot.data?.forEach((group) => group.isSubscribed =
                  subscriptionSnapshot.data?.any((subscription) => subscription.belongsTo(group)));

              return ConditionalWidget(
                condition: subscriptionSnapshot.connectionState == ConnectionState.active,
                widgetTrue: Builder(
                  builder: (BuildContext context) {
                    return GroupList(
                      groupsSnapshot.data,
                      padding: const EdgeInsets.only(top: 25.0, left: 20.0, right: 20.0),
                      onNotificationIconTap: _onGroupNotificationIconTap,
                    );
                  },
                ),
                widgetFalse: PlatformCircularProgressIndicator(),
              );
            },
          ),
          widgetFalse: PlatformCircularProgressIndicator(),
        );
      },
    ),

编辑 - WIP

    return groupCollection
        .where('members.${user.id}', isEqualTo: true)
        .snapshots()
        .flatMap((groups) => _groupSubscriptionsStream(groups) , //here should merging happen?));
  }

  Stream<QuerySnapshot> _groupSubscriptionsStream(QuerySnapshot groups) {
    final chunks = chunkSizeCollection(groups.documents.map((group) => group.documentID).toList(), 10);
    List<Stream<QuerySnapshot>> streams = List<Stream<QuerySnapshot>>();
    chunks.forEach((chunck) => streams.add(fireStore
        .collection(APIRoutes.userSubscriptions(user.id))
        .where(FieldPath.documentId, whereIn: chunck)
        .snapshots()));
    return ZipStream(streams, (value) => value.last);

【问题讨论】：

检查pub.dev/documentation/stream_transform/latest/stream_transform/… / pub.dev/documentation/rxdart/latest/rx/SwitchMapExtension/… / pub.dev/documentation/rxdart/latest/rx/FlatMapExtension/…
所以flatmap可以同时映射两个流，那我需要switchmap吗？我真的不明白 switchmap 的作用。你能再解释一下吗？
啊我想我明白了，我在组流上使用switchmap，然后在最后使用flat map对吗？
文档说：“switchMap 运算符类似于 flatMap 和 concatMap 方法，但它只从最近创建的 Stream 发出项目。” - 我认为最好理解它们的方法是在一些测试假流中使用它们
运行此代码：Stream.periodic(Duration(seconds: 3), (i) { print('master stream emitted ${10 * i}'); return 10 * i; }) .take(5) .flatMap((i) => Stream.periodic(Duration(milliseconds: 900), (ii) => '$i + $ii').take(10)) .listen(print) .onDone(() => print('the end')); 并查看输出到最后，现在看看区别替换 flatMap 为 1) switchMap 2) asyncExpand - 所有这些方法都需要传递“父”流的当前项的回调函数 - .flatMap((i) => ...

标签： flutter dart google-cloud-firestore stream

【解决方案1】：

我最终更改了我的 firestore 文档，因此我可以进行这样的查询

    return _groupSubscriptionsStream().switchMap((subscriptions) => groupCollection
        .where('members.${user.id}', isEqualTo: true)
        .snapshots()
        .map((snapshot) => snapshot.documents
            .map((document) => GroupModel.fromFirestore(document).checkSubscriptions(subscriptions))));
  }

  Stream<List<GroupSubscriptionModel>> _groupSubscriptionsStream() {
    return fireStore
        .collection(APIRoutes.userSubscriptions(user.id))
        .where('isSubscribed', isEqualTo: true)
        .snapshots()
        .map((snapshot) =>
            snapshot.documents.map((document) => GroupSubscriptionModel.fromFirestore(document)).toList());
  }

所以我现在有一个 isSubscribed 字段，该字段始终适用于查询。这确实很好用。

【讨论】：