How can I merge multiple Streams into a higher level Stream?

Question

I have two streams, Stream<A> and Stream<B>. I have a constructor for a type C that takes an A and a B. How do I merge the two Streams into a Stream<C>?

Answer 1

import 'dart:async' show Stream;
import 'package:async/async.dart' show StreamGroup;

main() async {
  var s1 = stream(10);
  var s2 = stream(20);
  var s3 = StreamGroup.merge([s1, s2]);
  await for(int val in s3) {
    print(val);
  }
}

Stream<int> stream(int min) async* {
  int i = min;
  while(i < min + 10) {
    yield i++;
  }
}

See also http://news.dartlang.org/2016/03/unboxing-packages-async-part-2.html

prints

10
20
11
21
12
22
13
23
14
24
15
25
16
26
17
27
18
28
19
29

Answer 2

You can use StreamZip in package:async to combine two streams into one stream of pairs, then create the C objects from that.

import "package:async" show StreamZip;
...
Stream<C> createCs(Stream<A> as, Stream<B> bs) =>
  new StreamZip([as, bs]).map((ab) => new C(ab[0], ab[1]));

Answer 3

If you need to react when either Stream<A> or Stream<B> emits an event and use the latest value from both streams, use combineLatest.

Stream<C> merge(Stream<A> streamA, Stream<B> streamB) {
  return streamA
    .combineLatest(streamB, (a, b) => new C(a, b));
}

Answer 4

For people that need to combine more than two streams of different types and get all latest values on each update of any stream.

import 'package:stream_transform/stream_transform.dart';

Stream<List> combineLatest(Iterable<Stream> streams) {
  final Stream<Object> first = streams.first.cast<Object>();
  final List<Stream<Object>> others = [...streams.skip(1)];
  return first.combineLatestAll(others);
}

The combined stream will produce:

streamA:  a----b------------------c--------d---|
streamB:  --1---------2-----------------|
streamC:  -------&----------%---|
combined: -------b1&--b2&---b2%---c2%------d2%-|

Why not StreamZip? Because StreamZip would produce:

streamA:  a----b------------------c--------d---|
streamB:  --1---------2-----------------|
streamC:  -------&----------%---|
combined: -------a1&-------b2%--|

Usage:

Stream<T> sA;
Stream<K> sB;
Stream<Y> sC;
combineLatest([sA, sB, sC]).map((data) {
  T resA = data[0];
  K resB = data[1];
  Y resC = data[2];
  return D(resA, resB, resC);
});

Answer 5

Using rxdart, you can use CombineLatestStream to achieve what you want. Note that the new stream doesn't return any value until all streams emitted at least one event:

You can create the combined stream using CombineLatestStream.list():

import 'package:rxdart/rxdart.dart';

Stream<A> s1 = Stream.fromIterable([A()]);
Stream<B> s2 = Stream.fromIterable([B()]);
Stream<dynamic> s3 = CombineLatestStream.list<dynamic>([s1, s2])
  ..listen(
    (value) {
      A a = value[0];
      B b = value[1];
    },
  );

Since Dart doesn't support union types, a downside of CombineLatestStream.list() is that events from streams of different types should be casted afterwards (due to List<dynamic>). Another approach is to use CombineLatestStream.combine2() (e.g. with a combiner that creates a Tuple2) to keep the types.

Answer 6

To get combined two streams when the second takes a result from the first one use asyncExpand

  Stream<UserModel?> getCurrentUserModelStream() {
    return FirebaseAuth.instance.authStateChanges().asyncExpand<UserModel?>(
      (currentUser) {
        if (currentUser == null) {
          return Stream.value(null);
        }
        return FirebaseFirestore.instance
            .collection('users')
            .doc(currentUser.uid)
            .snapshots()
            .map((doc) {
          final userData = doc.data();
          if (userData == null) {
            return null;
          }
          return UserModel.fromJson(userData);
        });
      },
    );
  }