将流与 rxdart 组合以在单个 StreamBuilder 中使用

Question

几天来，我试图找到这个问题的答案。 许多示例让我感到困惑，我真的不明白如何使用 rxdart 来实现它们，尽管它们看起来很简单。

我正在尝试将 Firestore 中的 4 个集合合并到一个 StreamBuilder 中。

Observable<QuerySnapshot> firstList = Firestore.instance.collection(bla).snapshots();

Observable<QuerySnapshot> secondList = Firestore.instance.collection(blabla).snapshots();

Observable<QuerySnapshot> thirdList= Firestore.instance.collection(blablabla).snapshots();

Observable<QuerySnapshot> fourthList= Firestore.instance.collection(blablablabla).snapshots();

最新编辑 我已经从 Observable 更改为 Stream，就像我尝试使用 rxdart 之前一样。

Stream<QuerySnapshot> firstList = Firestore.instance.collection(bla).snapshots();

Stream<QuerySnapshot> secondList = Firestore.instance.collection(blabla).snapshots();

Stream<QuerySnapshot> thirdList= Firestore.instance.collection(blablabla).snapshots();

Stream<QuerySnapshot> fourthList= Firestore.instance.collection(blablablabla).snapshots();

我正在使用 rxdart 来组合它们：

Observable<SomeList> allList = Observable.combineLatest4(
        firstList,
        secondList,
        thirdList,
        fourthList, (a, b, c, d) => SomeList(a, b, c, d));

这是我上面使用的类：

class SomeList{
  QuerySnapshot firstList;
  QuerySnapshot secondList;
  QuerySnapshot thirdList;
  QuerySnapshot fourthList;

  SomeList(this.firstList, this.secondList, this.thirdList,
      this.fourthList);
}

在构建方法中，我返回一个 StreamBuilder，如下所示：

Widget build(BuildContext context) {
    return Scaffold(
        appBar: _bla(),
        body: SingleChildScrollView(
        child: Column(
            children: <Widget>[
                return new StreamBuilder(
                    stream: allList, // here is my main issue
                    ...
                    ),
                ]
            ),
        ),  
    );
}   

当我运行代码时出现异常：

Class 'SomeList' has no instance getter 'documents'.
Receiver: Instance of 'SomeList'
Tried calling: documents

type '_BroadcastStream<QuerySnapshot>' is not a subtype of type 'Observable<QuerySnapshot>'

所以我尝试在初始化allList的时候去掉asBroadcastStream()的调用，但是报错是一样的。

在发布我的问题之前，我已经研究了这些示例：
Update streams based on filters in Dart/Flutter
Merge Firestore streams using rxDart
https://www.burkharts.net/apps/blog/rxdart-magical-transformations-of-streams/
https://medium.com/@ayushpguptaapg/using-streams-in-flutter-62fed41662e4

将这些流组合成一个流，然后在 StreamBuilder 中使用的正确方法是什么？ 可以举个小例子吗？

谢谢！

Answer 1

snapshots() 返回简单的 dart 异步 Stream，而不是 rxdart 的 Observable。你可以通过包装它来创建 Observable：

Observable<QuerySnapshot> firstList = Observable(Firestore.instance.collection(bla).snapshots());

UPD：Observable 扩展了 Stream 类，它可以像任何常规 Stream 一样使用。 RxDart 没有引入新的 API（与任何其他语言的实现不同）。相反，它使用的是 dart 的默认异步库，该库已经包含 Stream 和 StreamController 之类的类。 Observable 类只是扩展了 Stream 类，因此您可以使用 Observable 实例而不是默认的 dart Stream（例如在 StreamBuilder 中）。 RxDart 的 Observable 也可以使用默认流。

您遇到的另一个错误是因为您尝试从 SomeList 实例访问“文档”字段，但您的类没有这样的字段。