Flutter Firebase中创建动态Stream列表实现多文档实时监听

问题描述 #

需要实现一个可动态构建并返回Stream列表的函数，支持自定义MultiStreamBuilder同时监听多个Firebase Firestore文档的变动，不同Stream可对应不同集合下的不同文档。原有实现仅支持传入固定Stream列表，无法适配动态生成Stream的场景。

最初编写的_buildStreamList函数代码如下：

List<Stream<dynamic>> _buildStreamList() async* {
  var chat_overview_object = await query_chat_overview_id();

  List<Stream<dynamic>> stream_list = [];

  for (var i = 0; i < chat_overview_object.length; i++) {
    stream_list.add(
      FirebaseFirestore.instance
          .collection('chat')
          .doc('chat_overview_data')
          .collection('data')
          .doc('chat_overview_id_1234')
          .snapshots(),
    );
  }
  
yield stream_list;
}

初始自定义MultiStreamBuilder实现代码如下：

var _data = [];

typedef MultiStreamWidgetBuilder<T> = Widget Function(BuildContext context);

// 任意一个Stream触发事件时都会重新调用builder的组件
class MultiStreamBuilder extends StatefulWidget {
  const MultiStreamBuilder({
    required this.streams,
    required this.builder,
    Key? key,
  }) : super(key: key);

  final List<Stream<dynamic>> streams;
  final MultiStreamWidgetBuilder builder;

  Widget build(BuildContext context) => builder(context);

  @override
  State<MultiStreamBuilder> createState() => _MultiStreamBuilderState();
}

class _MultiStreamBuilderState extends State<MultiStreamBuilder> {
  final List<StreamSubscription<dynamic>> _subscriptions = [];

  @override
  void initState() {
    super.initState();
    _subscribe();
  }

  @override
  void didUpdateWidget(MultiStreamBuilder oldWidget) {
    super.didUpdateWidget(oldWidget);
    if (oldWidget.streams != widget.streams) {
      // 卸载旧订阅、重新订阅新Stream列表
      // 这里直接全量重订阅，不需要做增量差集处理，该场景触发频率极低不影响性能
      _unsubscribe();
      _subscribe();
    }
  }

  @override
  Widget build(BuildContext context) => widget.build(context);

  @override
  void dispose() {
    _unsubscribe();
    super.dispose();
  }

  void _subscribe() {
    for (final s in widget.streams) {
      final subscription = s.listen(
        (dynamic data) {
          setState(() {
            _data.add(data);
            print('data: ' + _data.toString());
          });
        },
        onError: (Object error, StackTrace stackTrace) {
          setState(() {});
        },
        onDone: () {
          setState(() {});
        },
      );
      _subscriptions.add(subscription);
    }
  }

  void _unsubscribe() {
    for (final s in _subscriptions) {
      s.cancel();
    }
    _subscriptions.clear();
  }
}

class AppWidget extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    return MultiStreamBuilder(
      streams: _buildStreamList(),
       [
      FirebaseFirestore.instance
            .collection('chat')
            .doc('chat_overview_data')
            .collection('data')
            .doc('chat_overview_id_1233')
            .snapshots(),
        FirebaseFirestore.instance
            .collection('chat')
            .doc('chat_overview_data')
            .collection('data')
            .doc('chat_overview_id_1234')
            .snapshots(),
        FirebaseFirestore.instance
            .collection('chat')
            .doc('chat_overview_data')
            .collection('data')
            .doc('chat_overview_id_1232')
            .snapshots()
      ],
      builder: _buildMain,
    );
  }
}

Widget _buildMain(BuildContext context) {
  return MaterialApp(
      home: Row(
    children: [
      Text(
        '_data: ' + _data.toString(),
        style: TextStyle(fontSize: 15),
      )
    ],
  ));
}

最终可运行实现 #

// 多Stream组件builder定义
typedef MultiStreamWidgetBuilder<T> = Widget Function(BuildContext context);

// 任意一个Stream触发事件时都会重新调用builder的组件
class MultiStreamBuilder extends StatefulWidget {
  const MultiStreamBuilder({
    required this.streams,
    required this.builder,
    Key? key,
  }) : super(key: key);

  final List<Stream<dynamic>> streams;
  final MultiStreamWidgetBuilder builder;

  Widget build(BuildContext context) => builder(context);

  @override
  State<MultiStreamBuilder> createState() => _MultiStreamBuilderState();
}

class _MultiStreamBuilderState extends State<MultiStreamBuilder> {
  final List<StreamSubscription<dynamic>> _subscriptions = [];

  @override
  void initState() {
    super.initState();
    _subscribe();
  }

  @override
  void didUpdateWidget(MultiStreamBuilder oldWidget) {
    super.didUpdateWidget(oldWidget);
    if (oldWidget.streams != widget.streams) {
      // 卸载旧订阅、重新订阅新Stream列表
      // 这里直接全量重订阅，不需要做增量差集处理，该场景触发频率极低不影响性能
      _unsubscribe();
      _subscribe();
    }
  }

  @override
  Widget build(BuildContext context) => widget.build(context);

  @override
  void dispose() {
    _unsubscribe();
    super.dispose();
  }

  void _subscribe() {
    for (final s in widget.streams) {
      final subscription = s.listen(
        (dynamic data) {
          setState(() {
            // 匹配对应ID的会话对象
            var object = chat_overview_object.singleWhere(
                (element) => element.chat_overview_id[0].trim() == data.id);

            // 更新会话对象的字段
            object.last_message_send = data['last_message_send'];
            object.last_update = data['last_update'];
            object.members_id = data['members_id'];
            object.new_message = data['new_message'];
            object.user_display_name = data['user_display_name'];
          });

          // 累计加载计数
          index++;

          // 所有文档初始数据加载完成
          if (index == chat_overview_object.length) {
            // 更新加载状态
            data_load.value = false;
          }
        },
        onError: (Object error, StackTrace stackTrace) {
          setState(() {});
        },
        onDone: () {
          setState(() {});
        },
      );
      _subscriptions.add(subscription);
    }
  }

  void _unsubscribe() {
    for (final s in _subscriptions) {
      s.cancel();
    }
    _subscriptions.clear();
  }
}

使用注意事项 #

• 外层需嵌套FutureBuilder先执行query_chat_overview_id()异步查询拿到要监听的文档列表后，再生成Stream数组传入MultiStreamBuilder，避免动态构建Stream时异步未完成的问题
• chat_overview_object、加载计数index、加载状态data_load建议封装到独立的状态类中管理，不要使用全局变量，避免多页面冲突
• 文档ID匹配的singleWhere逻辑建议增加orElse参数处理匹配不到的异常场景，避免崩溃

内容的提问来源于stack exchange，提问作者user16754196