StreamGroup.merge only retuning one stream

Question

I'm obviously misunderstanding how StreamGroup.merge works or how stream works or both!. I have two streams that are querying the same field in firestore. And I'm using merge to make one stream called mergedStream. This is used in a streamBuilder to serve up to the app. Here's my code:

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));
  }

and the merged stream I use for my stream builder:

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

Switching the order [s2, s1] changes the stream shown. And as the page loads I momentarily see the other stream but it disappears and the other stream appears..How can I merge the streams...?

EDIT: I'm adding the flutter tag because reading around it might be a problem with StreamBuilder

EDIT: This has helped a bit:(link in comments, And was just a proof of concept that the streams merged) But the stream searchResult() doesn't update when there's a state changes in one of he streams...You have to refresh the browser. (Flutter web)

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;
  }

just to clarify you want to have an unified stream that have a List<Order> of both s1 and s2 in one big List<Order>? — EdwynZN
Exactly. The streams are merging. But StreamBuilder see it as 2 events. So I can't show both streams from StreamBuilder — flutter
Also having seen this github.com/flutter/flutter/issues/23173#issuecomment-430712769 by @GünterZöchbauer that has helped me a bit. — flutter
try mergedStream().listen((data) { print("DataReceived: " + data); }); — Jitesh Mohite

EdwynZN EdwynZN · Accepted Answer · 2020-07-18T18:31:38

StreamGroup.merge merge the streams into an unified streams, but it doesn't merge the content of each one, so what you see in fact is the content of stream1 or stream2 coming one after the other.

I would recommend using StreamZip which combines the contents of the 2 streams and then emmits it as a new value (The first list contains the first value emitted by each stream, the second contains the second value, and so on).

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>

If you want to use the package Rx I would recommend ZipStream or 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
}

The difference between Zip and Combine is that Zip waits for all the inner streams to emit before doing the operation (first emmited value of s1 and first of s2, and so on) whereas combine doesn't care about the order, if you emit one of s1 and 3 of s2 then it will do [s1.1, s2.1] [s1.1, s2.2] ...

StreamGroup.merge only retuning one stream

1 Answers