在颤动中合并来自 Firestore 的流

Question

我一直在尝试使用 StreamBuilder 或类似的方式收听 Firestone 的多个专辑。 当我只使用一个 Stream 时，我的原始代码是：

import 'package:flutter/cupertino.dart';
import 'package:flutter/material.dart';
import 'package:cloud_firestore/cloud_firestore.dart';

class List extends StatefulWidget{

  ///The reference to the collection is like
  ///Firestore.instance.collection("users").document(firebaseUser.uid).collection("list1").reference()
  final CollectionReference listReference;

  List(this.listReference);

  @override
  State createState() => new ListState();
}

class ListState extends State<List> {

  @override
  Widget build(BuildContext context){

    return new StreamBuilder(
        stream: widget.listReference.snapshots(),
        builder: (context, snapshot) {
          return new ListView.builder(
              itemCount: snapshot.data.documents.length,
              padding: const EdgeInsets.only(top: 2.0),
              itemExtent: 130.0,
              itemBuilder: (context, index) {
                DocumentSnapshot ds = snapshot.data.documents[index];
                return new Data(ds);
              }
          );
        });
  }
}

这段代码运行良好，但现在我想听多个合集。我遇到了一个solution，它不涉及 StreamBuilder 并使用动态列表。我的代码现在看起来像这样：

import 'dart:async';
import 'package:flutter/cupertino.dart';
import 'package:flutter/material.dart';
import 'package:cloud_firestore/cloud_firestore.dart';
import 'main.dart';
import 'package:async/async.dart';

class ListHandler extends StatefulWidget{

  final CollectionReference listReference;

  ListHandler(this.listReference);

  @override
  State createState() => new ListHandlerState();
}

class ListHandlerState extends State<ListHandler> {

  StreamController streamController;
  List<dynamic> dataList = [];

  @override
  void initState() {
    streamController = StreamController.broadcast();
    setupData();
    super.initState();
  }

  @override
  void dispose() {
    super.dispose();
    streamController?.close();
    streamController = null;
  }

  Future<Stream> getData() async{
      Stream stream1 = Firestore.instance.collection("users").document(firebaseUser.uid).collection("list1").snapshots();
      Stream stream2 = Firestore.instance.collection("users").document(firebaseUser.uid).collection("list2").snapshots();

      return StreamZip(([stream1, stream2])).asBroadcastStream();
  }

  setupData() async {
    Stream stream = await getData()..asBroadcastStream();
    stream.listen((snapshot) {
      setState(() {
        //Empty the list to avoid repetitions when the users updates the 
        //data in the snapshot
        dataList =[];
        List<DocumentSnapshot> list;
        for(int i=0; i < snapshot.length; i++){
          list = snapshot[i].documents;
          for (var item in list){
            dataList.add(item);
          }
        }
      });
    });
  }

  @override
  Widget build(BuildContext context){
    if(dataList.length == 0){
      return new Text("No data found");
    }

    return new ListView.builder(
        itemCount: dataList.length,
        padding: const EdgeInsets.only(top: 2.0),
        itemBuilder: (context, index) {
          DocumentSnapshot ds = dataList[index];
          return new Data(ds['title']);
        }
    );
  }
}

问题是 ListView 返回Data，即StatefulWidget，用户可以与之交互，从而在 Firestore 中更改数据，从而出现下一个错误：

[VERBOSE-2:dart_error.cc(16)] Unhandled exception:
setState() called after dispose(): ListHandlerState#81967(lifecycle state: defunct, not mounted)
This error happens if you call setState() on a State object for a widget that no longer appears in the widget tree (e.g., whose parent widget no longer includes the widget in its build). This error can occur when code calls setState() from a timer or an animation callback. The preferred solution is to cancel the timer or stop listening to the animation in the dispose() callback. Another solution is to check the "mounted" property of this object before calling setState() to ensure the object is still in the tree.
This error might indicate a memory leak if setState() is being called because another object is retaining a reference to this State object after it has been removed from the tree. To avoid memory leaks, consider breaking the reference to this object during dispose().

应用程序不会崩溃，它会执行预期的操作，但始终显示此错误。

有些人使用库 rxdart 来处理流，我尝试过类似下面的代码，但是当我把它放在 StreamBuilder 中时，只有以下元素：

import 'dart:async';
import 'package:flutter/cupertino.dart';
import 'package:flutter/material.dart';
import 'package:cloud_firestore/cloud_firestore.dart';
import 'main.dart';
import 'showInfo.dart';
import 'package:rxdart/rxdart.dart';

class ListHandler extends StatefulWidget{

  @override
  State createState() => new ListHandlerState();
}

class ListHandlerState extends State<ListHandler> {

  Stream getData() {
    Stream stream1 = Firestore.instance.collection("users").document(firebaseUser.uid).collection("list1").snapshots();
    Stream stream2 = Firestore.instance.collection("users").document(firebaseUser.uid).collection("list2").snapshots();

    return Observable.merge(([stream2, stream1]));
  }

  @override
  Widget build(BuildContext context){
    return new StreamBuilder(
        stream: getData(),
        builder: (context, snapshot) {
          if(!snapshot.hasData){
            print(snapshot);
            return new Text("loading");
          }
          return new ListView.builder(
              itemCount: snapshot.data.documents.length,
              padding: const EdgeInsets.only(top: 2.0),
              itemBuilder: (context, index) {
                DocumentSnapshot ds = snapshot.data.documents[index];
                return new Data(ds);
              }
          );
        });
  }
}

这是我第一次与 Streams 合作，我不太了解它们，希望您能就如何做。

Answer 1

使用rxdart 中的CombineLatestStream 组合流。

StreamBuilder 将在每次其中一个流发出新事件时构建。

结果snapshot.data 是每个流的最新元素列表。

例子：

StreamBuilder(
stream: CombineLatestStream.list([
  stream0,
  stream1,
]),
builder: (context, snapshot) {
  final data0 = snapshot.data[0];
  final data1 = snapshot.data[1];
})

Answer 2

最后一个例子应该可以工作。 在您观察流期间，可能第二个流没有发出任何值。

async 包中的StreamGroup.merge() 也应该可以工作。

StreamZip 为每个流创建一对值。当它们以不同的速率发出值时，一个流会等待发出，直到另一个流发出一个值。 这可能不是你想要的。

Answer 3

问题不在于合并，而在于 StreamBuilder 根据 LATEST 快照更新 UI，换句话说，它不会堆叠快照，它只是拾取最后发出的事件，换句话说，流被合并并且合并的流确实包含所有合并流的数据，但是 streamBuilder 只会显示最后一个流发出的事件，解决方法是：

StreamBuilder<List<QuerySnapshot>>(stream: streamGroup, builder: (BuildContext context, 
    AsyncSnapshot<List<QuerySnapshot>> snapshotList){
                  if(!snapshotList.hasData){
                    return MyLoadingWidget();
                  }
                  // note that snapshotList.data is the actual list of querysnapshots, snapshotList alone is just an AsyncSnapshot

                  int lengthOfDocs=0;
                  int querySnapShotCounter = 0;
                  snapshotList.data.forEach((snap){lengthOfDocs = lengthOfDocs + snap.documents.length;});
                  int counter = 0;
                  return ListView.builder(
                    itemCount: lengthOfDocs,
                    itemBuilder: (_,int index){
                      try{DocumentSnapshot doc = snapshotList.data[querySnapShotCounter].documents[counter];
                      counter = counter + 1 ;
                       return new Container(child: Text(doc.data["name"]));
                      }
                      catch(RangeError){
                        querySnapShotCounter = querySnapShotCounter+1;
                        counter = 0;
                        DocumentSnapshot doc = snapshotList.data[querySnapShotCounter].documents[counter];
                        counter = counter + 1 ;
                         return new Container(child: Text(doc.data["name"]));
                      }

                    },
                  );
                },

Answer 4

你可能想试试 rxDart 包中的 concatWith：

返回一个从当前流中发出所有项目的流，然后 从给定的流中发出所有项目，一个接一个。

https://pub.dev/packages/rxdart

import 'package:rxdart/rxdart.dart';

      Stream.fromIterable(['a', 'b', 'c']).concatWith([
        Stream.fromIterable(['d', 'e', 'f'])
      ]).listen(print);

这将打印：

一个

b

c

d

e

f