In Dart, how to ensure stream updates are finished before moving on?

Question

I am programming in Dart 2.1.0, trying to update some states by listening to a custom stream:

import 'dart:async';

void main() {
  final controller = StreamController<int>();
  final divisibleBy2 = controller.stream.map((i) => i % 2 == 0);

  var seen2x = false;
  divisibleBy2.listen((y) {
    seen2x = y;
  });

  controller.add(2);
  print(seen2x);
}

When I run it, it prints false. Apparently, the print function is called sooner than the listener function. The state is not updated in time.

By await-ing on controller.add(2), I can get the order straight:

import 'dart:async';

void main() async {
  final controller = StreamController<int>();
  final divisibleBy2 = controller.stream.map((i) => i % 2 == 0);

  var seen2x = false;
  divisibleBy2.listen((y) {
    seen2x = y;
  });

  await controller.add(2);
  print(seen2x);
}

Now it prints true. The listener function is called before print.

But why does await make a difference in this case? StreamController's add() method does not return a Future, so I am not sure why await matters. In fact, in my real-world usage where the stream network gets more complicated, await sometimes is not enough to ensure the order of execution. That's a question for another day.

For now, I just want to see if await is the proper way to do it. Thanks in advance.

Solution 1

The reason the await made a difference here is that any await will wait at least a microtask - regardless of whether the value you are awaiting is a Future or not. If it is a Future it will wait for it to complete or a microtask if it is already complete. It would behave identically if you had await null; following the controller.add(2);.

In general a StreamController doesn't know when all listeners have received the event, especially considering things like async listeners or Stream.asyncMap.

Solution 2

I'm going to add an alternate answer for what I did when waiting for a stream to complete. await for was the key that I was looking for.

var myStream = CustomCacheManager().getFile(_url);
File fetchedFile;
await for (var fileInfo in myStream) {
  fetchedFile = fileInfo.file;
}

// stream is finished now