Irbisa · Cheatsheet

Streams & Rx

Streams, StreamController, StreamBuilder, broadcast streams

Middle Developer · 23 questions

  1. 01

    What is the difference between a Future and a Stream, and when do you use each?

    A Future delivers one result or one error, once; a Stream delivers any number of values and errors over time and may then finish.

  2. 02

    What is a Stream in Dart, and how do single-subscription and broadcast streams differ?

    A Stream<T> is an asynchronous sequence of events: data, errors and a final done.

    // Single-subscription: exactly one listenerfinal single = StreamController<int>();single.stream.listen((v) => print('A got $v'));// single.stream.listen(...); // Bad state: Stream has already been listened to // Broadcast: many listeners, no replay…
  3. 03

    What does this code print, and what changes if you remove .broadcast()?

    import 'dart:async'; Future<void> main() async {  final controller = StreamController<String>.broadcast();  controller.add('hello');  controller.stream.listen(print);  controller.add('world');  await controller.close();}

    It prints only world: a broadcast controller with no listeners discards hello on the spot.

  4. 04

    What is the difference between FutureBuilder and StreamBuilder?

    FutureBuilder waits for one Future and rebuilds when it completes; StreamBuilder subscribes to a Stream and rebuilds on every event until it closes.

    class _ProfileState extends State<Profile> {  late final Future<User> _user = api.fetchUser(widget.id);     // once  late final Stream<int> _unread = chat.watchUnread(widget.id); // over time   @override  Widget build(BuildContext context) {…
  5. 05

    What does a StreamController give you, and how do you use it to turn a callback API (a sensor, a socket) into a Stream?

    A StreamController pairs a sink you push into (add, addError, close) with a stream you hand out.

    Stream<int> watch(ValueListenable<int> source) {  late final StreamController<int> controller;  void push() => controller.add(source.value);   controller = StreamController<int>.broadcast(    onListen: () => source.addListener(push),      // first listener: start…
  6. 06

    What is a StreamSubscription, and how do you manage its lifecycle in a widget?

    StreamSubscription is the handle listen returns: through it you pause, resume and cancel delivery.

    class _TickerState extends State<Ticker> {  StreamSubscription<int>? _sub;  int _last = 0;   @override  void initState() {…
  7. 07

    When do you use await for instead of listen on a stream?

    await for consumes events one at a time inside an async function and does not take the next event until the loop body finishes; listen registers a callback, returns at once and gives you a StreamSubscription to cancel later.

    // Sequential: the next order is not delivered until this one is savedFuture<void> saveAll(Stream<Order> orders) async {  await for (final order in orders) {    await db.insert(order);    if (order.isLast) break;   // cancels the subscription  }…
  8. 08

    What does this code print, and in what order?

    Stream<int> numbers() async* {  print('start');  yield 1;  yield 2;} void main() {  final stream = numbers();  print('created');  stream.listen(print);  print('listened');}

    It prints created, listened, start, 1, 2.

  9. 09

    What does StreamBuilder do for you, and what do connectionState and the snapshot tell you?

    StreamBuilder subscribes to a stream for you, rebuilds its builder on every event and cancels the subscription when the widget leaves the tree.

    class _PricesState extends State<Prices> {  late final Stream<Quote> _quotes = repo.watchQuotes(widget.symbol);   @override  Widget build(BuildContext context) {    return StreamBuilder<Quote>(…
  10. 10

    Two widgets listen to the same stream and you get "Bad state: Stream has already been listened to". What is happening?

    The stream is single-subscription: it allows exactly one listener for its whole life, and cancelling that listener does not free the slot.

    // wrong: two listeners on a single-subscription streamfinal controller = StreamController<int>();controller.stream.listen(print);controller.stream.listen(print);   // Bad state: already listened to// right: broadcast when you own the controllerfinal broadcast = StreamController<int>.broadcast();…
  11. 11

    How do errors travel through a Stream, and what happens to an error nobody handles?

    Errors are events: addError, or a throw inside an async* body, sends them down the same pipeline as data to the listener's onError.

    // wrong: data handled, errors escape to the zone and the crash reporter_sub = repo.messages.listen((m) => setState(() => _messages.add(m))); // right: handle both channels_sub = repo.messages.listen(  (m) => setState(() => _messages.add(m)),…
  12. 12

    Find the bug: the progress bar reaches 100%, but the "Upload complete" message never appears.

    Stream<double> upload(Uploader client, File file) {  final progress = StreamController<double>();  client      .send(file, onProgress: (sent, total) => progress.add(sent / total))      .catchError((Object e, StackTrace s) => progress.addError(e, s));  return progress.stream;} Future<void> finish(Uploader client, File file) async {  await for (final p in upload(client, file)) {    setProgress(p);  }  showDone();}

    The StreamController is never closed, so the stream never sends done: after 100% the await for keeps waiting and showDone() never runs, nor does anything after await finish(...), such as hiding a spinner or a finally block.

  13. 13

    What is a StreamTransformer, and how do you write your own stream operator without rxdart?

    A StreamTransformer<S, T> turns one stream into another and is applied with stream.transform(...); utf8.decoder and LineSplitter are transformers.

    // Per event: trim lines and drop blank onesfinal trimNonEmpty = StreamTransformer<String, String>.fromHandlers(  handleData: (line, sink) {    final t = line.trim();    if (t.isNotEmpty) sink.add(t);  },…
  14. 14

    How do you combine several streams: merge, zip, combineLatest?

    merge forwards every event from every stream as it arrives; zip pairs the first event of each stream, then the second, and so on; combineLatest emits the latest value of every stream whenever any one of them emits.

    import 'package:rxdart/rxdart.dart'; // merge: every event from every source, in arrival orderfinal allMessages = Rx.merge<Message>([channelA.messages, channelB.messages]); // zip: the i-th question with the i-th answer…
  15. 15

    What does rxdart add on top of dart:async (BehaviorSubject, debounceTime, switchMap)?

    rxdart adds BehaviorSubject (a stream with a current value), debounceTime (wait for a pause) and switchMap (cancel the previous inner stream); all of them still return a normal Stream.

    import 'package:rxdart/rxdart.dart'; class SearchViewModel {  SearchViewModel(this._api);  final SearchApi _api;…
  16. 16

    Your repository re-emits an identical list on every poll and the page rebuilds each time; you added distinct() and nothing changed: why?

    A List (and any model without ==) is compared by identity, so every freshly decoded poll result counts as a new value and distinct() lets it through.

  17. 17

    You call await feed.quotes.first to read the current price and the call never returns: what did you assume that a Stream does not provide?

    You assumed a stream has a current value.

  18. 18

    Sign-in success shows its SnackBar three times because it is fired inside the builder of a StreamBuilder: what rule is being broken?

    The rule is that builder only draws: Flutter may call it any number of times, so a side effect like a SnackBar or navigation belongs in a subscription callback, not in the build.

    // wrong: runs on every rebuild, not on every eventbuilder: (context, snap) {  if (snap.data is SignedIn) {    ScaffoldMessenger.of(context).showSnackBar(welcome);  }  return const HomeBody();…
  19. 19

    Find the bug: after switching between chats, messages from the chat you left keep appearing in the open one.

    class ChatController {  ChatController(this._repo);  final ChatRepository _repo;  final messages = ValueNotifier<List<Message>>(const []);  StreamSubscription<Message>? _sub;  void open(String chatId) {    messages.value = const [];    _sub = _repo        .watch(chatId)        .listen((m) => messages.value = [...messages.value, m]);  }  void dispose() {    _sub?.cancel();    messages.dispose();  }}

    open overwrites _sub without cancelling the previous subscription, so every chat opened so far keeps listening and its messages land in the same list.

  20. 20

    How are streams used for state management in Flutter, and why do apps rarely keep state in a raw StreamController?

    Streams model state as a sequence of values: the source emits a new immutable state and widgets rebuild through StreamBuilder.

  21. 21

    What is backpressure, and how do single-subscription and broadcast streams handle it?

    Backpressure is a slow consumer making a faster producer wait.

    // A producer that respects backpressure: suspends at yield while pausedStream<List<int>> readChunks(File file) async* {  final raf = await file.open();  try {    while (true) {      final chunk = await raf.read(64 * 1024);…
  22. 22

    A WebSocket price feed goes quiet on a flaky mobile network, the stream never errors, and the UI shows a stale price forever: what do you add?

    Add stream.timeout so that silence becomes an event: if no price arrives within the window, the pipeline gets a TimeoutException and the UI can show "stale" and reconnect.

    class PriceFeed {  final _out = StreamController<Quote>.broadcast();  IOWebSocketChannel? _channel;  StreamSubscription<Quote>? _sub;  Timer? _retry;  int _attempt = 0;…
  23. 23

    A test adds an event to a StreamController, asserts on the listener immediately, and nothing has run yet: is sync: true the fix?

    No. Asynchronous delivery is the contract: add schedules the event and the listener runs in a later microtask, so the code after add always finishes first.

    // wrong: the assertion runs before the listener doestest('stores the value', () {  final controller = StreamController<int>();  var seen = 0;  controller.stream.listen((v) => seen = v);  controller.add(7);…