Streams und reaktive Daten

Beyond a Single Future

Listening to a Stream

stream.listen(
  (data) => print('Got: ' + data.toString()),
  onError: (e) => print('Error: ' + e.toString()),
  onDone: () => print('Done'),
);

Creating Streams with async*

Stream<int> countTo(int n) async* {
  for (var i = 1; i <= n; i++) {
    await Future.delayed(const Duration(seconds: 1));
    yield i;
  }
}

await for

Future<void> run() async {
  await for (final value in countTo(3)) {
    print(value);
  }
}

Single vs Broadcast

final broadcast = controller.stream.asBroadcastStream();
broadcast.listen((v) => print('A: ' + v.toString()));
broadcast.listen((v) => print('B: ' + v.toString()));

StreamController

final controller = StreamController<String>();
controller.add('hello');
controller.add('world');
controller.close();

Transforming Streams

stream
  .where((n) => n.isEven)
  .map((n) => n * 10)
  .take(5)
  .listen(print);

StreamBuilder Widget

StreamBuilder<int>(
  stream: countTo(5),
  builder: (context, snapshot) {
    if (!snapshot.hasData) return const CircularProgressIndicator();
    return Text('Value: ' + snapshot.data.toString());
  },
);

Reading the Snapshot

if (snapshot.hasError) return Text('Error: ' + snapshot.error.toString());

Cancelling Subscriptions

late StreamSubscription sub;

@override
void dispose() {
  sub.cancel();
  super.dispose();
}

Backpressure & Errors

stream
  .handleError((e) => print('Skipped: ' + e.toString()))
  .listen(print);

Quick Check

Recap