Dart's future.asStream() quick overview

Guilherme Silva — Sat, 29 Oct 2022 21:47:07 +0000

Dart makes it easy to transform a future into a stream, but what will happen after that? For example, if we cancel the stream, what will happen in the future? And what about if the future throws an exception? Can a try-catch block help?

TLDR

Into

Problem: In dart, you can't cancel (or stop) a future

Possible solution: you can cancel a stream's subscription. So, the idea is to transform a future into a stream, and cancel it when needed.

We need to understand what will happen:

in case it completes normally
- before that, the stream is canceled
- but the stream was already canceled
in case it throws
- before the stream is canceled,
  - onError is defined
    - and cancelOnError is false (default)
    - and cancelOnError is true (default)
  - onError is NOT defined
- after the stream is canceled

Situation 1 - the `future` completes normally, before that the `stream` is canceled

void main(List<String> args) async {
  try {
    final stream = longAPICall().asStream();
    var sub = stream.listen(
      (v) => print('  stream.onData(): future finished with value: $v'),
      onDone: () => print('  stream.onDone(): done'),
      onError: (e) => print(' stream.onError(): ${e.toString()}'),
      // cancelOnError: false, // default - onDone() will be called even when errors happens
      // cancelOnError: true, // onDone() will NOT be called when errors happens
    );
    // await sub.cancel();
    // print('   stream.cancel() was ALREADY cancelled!');
  } catch (e) {
    // this never happens, even if we don't define onError()
    // in this case it'll handled by zones
    print('catched: ${e.toString()}');
  }
}

Future<bool> longAPICall() async {
  print('future has STARTED...');
  await Future.delayed(Duration(seconds: 2));
  print('future is STILL running');
  await Future.delayed(Duration(seconds: 2));
  // throw Exception('BAL');
  print('future has FINISHED');
  return true;
}

output:

future has STARTED...
future is STILL running
future has FINISHED
  stream.onData(): future finished with value: true
  stream.onDone(): done
Exited

conclusions:

both onDone and onData callbacks are called
onData is called with the future's result

Situation 2 - the `future` completes normally, but the `stream` was already canceled

void main(List<String> args) async {
  try {
    final stream = longAPICall().asStream();
    var sub = stream.listen(
      (v) => print('  stream.onData(): future finished with value: $v'),
      onDone: () => print('  stream.onDone(): done'),
      onError: (e) => print('  stream.onError(): ${e.toString()}'),
      // cancelOnError: false, // default - onDone() will be called even when errors happens
      // cancelOnError: true, // onDone() will NOT be called when errors happens
    );
    await sub.cancel();
    print('   stream.cancel() was ALREADY cancelled!');
  } catch (e) {
    // this never happens, even if we don't define onError()
    // in this case it'll handled by zones
    print('catched: ${e.toString()}');
  }
}

Future<bool> longAPICall() async {
  print('future has STARTED...');
  await Future.delayed(Duration(seconds: 2));
  print('future is STILL running');
  await Future.delayed(Duration(seconds: 2));
  // throw Exception('HEHE');
  print('future has FINISHED');
  return true;
}

output:

future has STARTED...
   stream.cancel() was ALREADY cancelled!
future is STILL running
future has FINISHED
Exited

conclusions:

the future continues running, even after we canceled the stream
both onDone and onData callbacks are NOT called (the result is completely ignored)

Situation 3 - `throws` before the `stream` is canceled, `onError` is defined and `cancelOnError` is `false` (default)

void main(List<String> args) async {
  try {
    final stream = longAPICall().asStream();
    var sub = stream.listen(
      (v) => print('  stream.onData(): future finished with value: $v'),
      onDone: () => print('  stream.onDone(): done'),
      onError: (e) => print('  stream.onError(): ${e.toString()}'),
      cancelOnError: false, // default - onDone() will be called even when errors happens
      // cancelOnError: true, // onDone() will NOT be called when errors happens
    );
    // await sub.cancel();
    // print('   stream.cancel() was ALREADY cancelled!');
  } catch (e) {
    // this never happens, even if we don't define onError()
    // in this case it'll handled by zones
    print('catched: ${e.toString()}');
  }
}

Future<bool> longAPICall() async {
  print('future has STARTED...');
  await Future.delayed(Duration(seconds: 2));
  print('future is STILL running');
  await Future.delayed(Duration(seconds: 2));

  throw Exception('HEHE');

  print('future has FINISHED');
  return true;
}

output:

future has STARTED...
future is STILL running
  stream.onError(): Exception: HEHE
  stream.onDone(): done
Exited

conclusions:

both onDone and onError callbacks are called

Situation 4 - `throws` before the `stream` is canceled, `onError` is defined, `cancelOnError` is `true`

void main(List<String> args) async {
  try {
    final stream = longAPICall().asStream();
    var sub = stream.listen(
      (v) => print('  stream.onData(): future finished with value: $v'),
      onDone: () => print('  stream.onDone(): done'),
      onError: (e) => print('  stream.onError(): ${e.toString()}'),
      // cancelOnError: false, // default - onDone() will be called even when errors happens
      cancelOnError: true, // onDone() will NOT be called when errors happens
    );
    // await sub.cancel();
    // print('   stream.cancel() was ALREADY cancelled!');
  } catch (e) {
    // this never happens, even if we don't define onError()
    // in this case it'll handled by zones
    print('catched: ${e.toString()}');
  }
}

Future<bool> longAPICall() async {
  print('future has STARTED...');
  await Future.delayed(Duration(seconds: 2));
  print('future is STILL running');
  await Future.delayed(Duration(seconds: 2));
  throw Exception('HEHE');
  print('future has FINISHED');
  return true;
}

output:

future has STARTED...
future is STILL running
  stream.onError(): Exception: HEHE
Exited

conclusions:

onError callbacks is called
onDone is NOT called
try-catch is NOT called

Situation 5 - `throws` before the `stream` is canceled, `onError` is NOT defined

void main(List<String> args) async {
  try {
    final stream = longAPICall().asStream();
    var sub = stream.listen(
      (v) => print('  stream.onData(): future finished with value: $v'),
      onDone: () => print('  stream.onDone(): done'),
      // onError: (e) => print('  stream.onError(): ${e.toString()}'),
      // cancelOnError: false, // default - onDone() will be called even when errors happens
      // cancelOnError: true, // onDone() will NOT be called when errors happens
    );
    // await sub.cancel();
    // print('   stream.cancel() was ALREADY cancelled!');
  } catch (e) {
    // this never happens, even if we don't define onError()
    // in this case it'll handled by zones
    print('catched: ${e.toString()}');
  }
}

Future<bool> longAPICall() async {
  print('future has STARTED...');
  await Future.delayed(Duration(seconds: 2));
  print('future is STILL running');
  await Future.delayed(Duration(seconds: 2));
  throw Exception('HEHE');
  print('future has FINISHED');
  return true;
}

output:

future has STARTED...
future is STILL running
  stream.onDone(): done
Unhandled exception:
Exception: BAL
#0      longAPICall
bin/future_playground.dart:29
<asynchronous suspension>
#1      new Stream.fromFuture.<anonymous closure> (dart:async/stream.dart:247:17)
<asynchronous suspension>

Exited (255)

conclusions:

both onDone is still called (because cancelOnError's default is false)
try-catch is ignored (aahá! 👀)
exception is caught by Zone's error handling

Situation 6 - the `future` `throws` after the `stream` is canceled

void main(List<String> args) async {
  try {
    final stream = longAPICall().asStream();
    var sub = stream.listen(
      (v) => print('  stream.onData(): future finished with value: $v'),
      onDone: () => print('  stream.onDone(): done'),
      // onError: (e) => print('  stream.onError(): ${e.toString()}'),
      // cancelOnError: false, // default - onDone() will be called even when errors happens
      // cancelOnError: true, // onDone() will NOT be called when errors happens
    );
    await sub.cancel();
    print('   stream was ALREADY cancelled!');
  } catch (e) {
    // this never happens, even if we don't define onError()
    // in this case it'll handled by zones
    print('catched: ${e.toString()}');
  }
}

Future<bool> longAPICall() async {
  print('future has STARTED...');
  await Future.delayed(Duration(seconds: 2));
  print('future is STILL running');
  await Future.delayed(Duration(seconds: 2));
  throw Exception('HEHE');
  print('future has FINISHED');
  return true;
}

output:

future has STARTED...
   stream was ALREADY cancelled!
future is STILL running
Exited

conclusions:

the future continues running even after we canceled the stream
try-catch is ignored
onError was not declared, and it didn't make a difference (the error is simply ignored, and wasn't sent to zone 's handler) (aahá(2)! 👀)

DEV Community: Guilherme Silva

Dart's future.asStream() quick overview

TLDR

Into

Situation 1 - the future completes normally, before that the stream is canceled

Situation 2 - the future completes normally, but the stream was already canceled

Situation 3 - throws before the stream is canceled, onError is defined and cancelOnError is false (default)

Situation 4 - throws before the stream is canceled, onError is defined, cancelOnError is true

Situation 5 - throws before the stream is canceled, onError is NOT defined

Situation 6 - the future throws after the stream is canceled

Situation 1 - the `future` completes normally, before that the `stream` is canceled

Situation 2 - the `future` completes normally, but the `stream` was already canceled

Situation 3 - `throws` before the `stream` is canceled, `onError` is defined and `cancelOnError` is `false` (default)

Situation 4 - `throws` before the `stream` is canceled, `onError` is defined, `cancelOnError` is `true`

Situation 5 - `throws` before the `stream` is canceled, `onError` is NOT defined

Situation 6 - the `future` `throws` after the `stream` is canceled