flutter - StreamGroup.merge仅重新调整一个流

Question

我显然误会了StreamGroup.merge的工作方式或流的工作方式或两者!我有两个流在Firestore中查询同一字段。我正在使用merge使一个流称为mergedStream。在streamBuilder中用于服务该应用程序。
这是我的代码:

Stream<List<Order>> stream1({Order order})  {
    return _service.collectionStream(
        path: APIPath.orders(),
        queryBuilder: (query) =>
            query.where('orderStatus', isEqualTo: 'Pending'),
        builder: (documentSnapshot) => Order.fromFirebase(documentSnapshot));
  }

  Stream<List<Order>> stream2({Order order})  {
     return _service.collectionStream(
        path: APIPath.orders(),
        queryBuilder: (query) =>
            query.where('orderStatus', isEqualTo: 'Preparing'),
        builder: (documentSnapshot) => Order.fromFirebase(documentSnapshot));
  }

以及用于流构建器的合并流:

Stream<List<Order>> mergedStream() {
    final s1 = preparingStream();
    final s2 = pendingStream();
    return StreamGroup.merge([s2, s1]);
  }

切换顺序[s2, s1]会更改显示的流。当页面加载时，我暂时看到另一个流，但它消失了，另一个流出现了。我如何合并这些流...？
编辑:我添加flutter标签，因为阅读它周围可能是StreamBuilder与problem
编辑:这有所帮助:(在评论中的链接，并且仅仅是流合并的概念证明)但是，当他的一个流中的状态发生变化时，流searchResult()不会更新...您必须刷新浏览器。 ( flutter 网)

Stream<List<Order>> searchResult() {
    List<Stream<List<Order>>> streamList = [stream1(), stream2()];

    var x = Rx.merge(streamList).scan<List<Order>>((acc, curr,i) {
      return acc ?? <Order>[]
        ..addAll(curr);
    });
    //this test shows that the stream gets contents from both query streams
    // x.map((convert){ convert.documents.forEach((f){print(f.data["name"]);});}).listen(print);

    return x;
  }

Answer 1

StreamGroup.merge将流合并为一个统一的流，但是它不会合并每个流的内容，因此，您实际上看到的是stream1或stream2的内容一个接一个。
我建议使用StreamZip，它将两个流的内容组合在一起，然后将其作为新值来发出(第一个列表包含每个流发出的第一个值，第二个列表包含第二个值，依此类推)。

Stream<List<List<Order>>> mergedStream() {
    final s1 = preparingStream();
    final s2 = pendingStream();
    return StreamZip([s2, s1]);
  }
//This will give you a List with all the emmited values of each stream, in this case a List of List<Order>

如果您想使用Rx软件包，我建议ZipStream或CombineLatestStream

import 'package:rxdart/rxdart.dart';


Stream<List<Order>> get combineList{
  final s1 = preparingStream();
  final s2 = pendingStream();
  return ZipStream.zip2<<List<Order>>, <List<Order>>, <List<Order>>>(
    s1, s2, (firstOrder, secondOrder) => <Order>[...firstOrder, ...secondOrder]); //here you combine both in a list
}

Zip和Combine的区别在于Zip在执行操作之前先等待所有内部流发出(s1的第一个发射值和s2的第一个发射值，依此类推)，而combine不在乎顺序，如果发出一个的s1和s2的3，那么它将执行[s1.1, s2.1] [s1.1, s2.2] ...