Flutter 3中如何将List<Stream<MyUser>>转换为Stream<List<MyUser>>

问题场景 #

我有一个从 Firestore 批量获取用户数据的方法，代码如下：

static Stream<List<MyUser>> getUsersByList(List<String> userIds) {
     final List<Stream<MyUser>> streams = [];
     for (String id in userIds) {
       streams.add(firebaseFirestore
           .collection('user_profile')
           .doc(id)
           .snapshots()
           .map((DocumentSnapshot snap) =>
               MyUser.fromMap(snap.data as Map<String, dynamic>)));
     }
 
     return 'here is a return data';
 
     // 旧版写法：return StreamZip<MyUser>(streams).asBroadcastStream();
}

需要将该方法返回值传入 StreamProvider：

StreamProvider.value(
value: UserProvider.getUsersByList(userIds),
initialData: [],
child: HomePage());

但 StreamProvider 要求传入 Stream<List> 类型的数据，旧版使用 StreamZip 实现 List<Stream<MyUser>> 到 Stream<List<MyUser>> 的转换，请问 Flutter 3 中该如何实现？

解决方案 #

方案1：使用 Dart 原生 StreamZip（无额外依赖） #

StreamZip 并未被移除，仅构造函数写法简化，直接使用即可：

import 'dart:async';

static Stream<List<MyUser>> getUsersByList(List<String> userIds) {
  final List<Stream<MyUser>> streams = [];
  for (String id in userIds) {
    streams.add(firebaseFirestore
        .collection('user_profile')
        .doc(id)
        .snapshots()
        .map((DocumentSnapshot snap) =>
            MyUser.fromMap(snap.data as Map<String, dynamic>)));
  }

  // 新版 StreamZip 写法，无需指定泛型
  return StreamZip(streams).asBroadcastStream();
}

行为说明：当所有输入 Stream 都发出至少一个值后，任意一个 Stream 更新时，会将所有 Stream 的最新值合并为 List 返回。

方案2：使用 RxDart 的 combineLatestList（更灵活） #

若项目已引入 rxdart 依赖，推荐使用 combineLatestList，API 更直观：

1. 先添加依赖到 pubspec.yaml：

dependencies:
  rxdart: ^0.27.7 # 请使用最新版本

2. 修改方法代码：

import 'package:rxdart/rxdart.dart';

static Stream<List<MyUser>> getUsersByList(List<String> userIds) {
  final List<Stream<MyUser>> streams = [];
  for (String id in userIds) {
    streams.add(firebaseFirestore
        .collection('user_profile')
        .doc(id)
        .snapshots()
        .map((DocumentSnapshot snap) =>
            MyUser.fromMap(snap.data as Map<String, dynamic>)));
  }

  return Rx.combineLatestList(streams).asBroadcastStream();
}

方案3：手动实现合并逻辑（完全自定义） #

如果需要完全控制合并行为，可手动实现 Stream 合并：

static Stream<List<MyUser>> getUsersByList(List<String> userIds) {
  if (userIds.isEmpty) return Stream.value([]);

  // 初始化缓存列表，存储每个Stream的最新用户数据
  final List<MyUser?> userCache = List.filled(userIds.length, null);
  int initializedStreams = 0;
  final controller = StreamController<List<MyUser>>.broadcast();

  for (int i = 0; i < userIds.length; i++) {
    firebaseFirestore
        .collection('user_profile')
        .doc(userIds[i])
        .snapshots()
        .map((snap) => MyUser.fromMap(snap.data as Map<String, dynamic>))
        .listen((user) {
          userCache[i] = user;
          // 只有所有Stream都返回过数据后，才推送完整列表
          if (userCache.every((u) => u != null)) {
            controller.add(userCache.cast<MyUser>().toList());
          }
        }, onError: (err) {
          controller.addError(err);
        }, onDone: () {
          // 处理Stream关闭的情况，可根据业务需求调整
          initializedStreams++;
          if (initializedStreams == userIds.length) {
            controller.close();
          }
        });
  }

  return controller.stream;
}

内容的提问来源于stack exchange，提问作者Maral-Erdene Tumursuh