Streams & Rx
Irbisa · cheatsheetSeptember 13, 2026

Streams & Rx

Streams, StreamController, StreamBuilder, broadcast streams

Middle Developer20 itemscompressed for a skim
  1. 01

    What are Streams in Dart? Explain single-subscription vs broadcast streams.

    Medium

    A Stream<T> delivers a sequence of async events over time.

    // ── async* generator ──────────────────────────────
Stream<int> countDown(int from) async* {
  for (var i = from; i >= 0; i--) {
    await Future.delayed(const Duration(seconds: 1));
    yield i;
  }
…
  2. 02

    What does a StreamController give you, and which transformations would you apply to a search-input stream?

    Medium

    StreamController gives you a stream plus a sink you push events into, so any callback-based source can become a Stream.

    // ── StreamController ────────────────────────────────
class SearchBloc {
  final _queryCtrl = StreamController<String>();
  Sink<String> get querySink => _queryCtrl.sink;

  // Debounce: wait 300ms after last keystroke
…
  3. 03

    What is StreamSubscription and how do you manage its lifecycle correctly?

    Medium

    A call to listen hands back a StreamSubscription — your only handle to pause, resume or cancel delivery.

    class TickerWidget extends StatefulWidget {
  const TickerWidget({super.key, required this.events});
  final Stream<int> events;
  @override State<TickerWidget> createState() => _TickerWidgetState();
}
…
  4. 04

    When do you use await for vs listen on a stream?

    Medium

    Two ways to consume a stream — both delegate to the same subscription mechanism, but they shape your code very differently.

    // ── await for — sequential, easy to read ──
Future<void> processOrders(Stream<Order> incoming) async {
  await for (final order in incoming) {
    try {
      // ✅ The next event won't be delivered until this finishes — backpressure
      await orders.persist(order);
…
  5. 05

    How do you handle errors in a Stream pipeline?

    Medium

    Errors on a stream are first-class events delivered through the error channel — they don't throw at the listen call site.

    // ── listen with explicit onError ──
stream.listen(
  (data) => use(data),
  onError: (Object e, StackTrace s) {
    log('failed: $e');
    crashlytics.recordError(e, s);
…
  6. 06

    What does rxdart bring on top of dart:async — BehaviorSubject, switchMap, debounce?

    Hard

    rxdart wraps dart:async with the Rx-style operators that are standard in other reactive libraries (RxJava, RxJS).

    import 'package:rxdart/rxdart.dart';

// ── BehaviorSubject — last value replayed to new listeners ──
class AuthService {
  final _user = BehaviorSubject<User?>.seeded(null);
  Stream<User?> get user$ => _user.stream;
…
  7. 07

    How do you combine multiple streams — merge, zip, combineLatest?

    Medium

    Three combinators cover most multi-stream patterns.

    import 'package:rxdart/rxdart.dart';

// ── merge — every event from every source ──
final allMessages = Rx.merge<Message>([
  channelA.messages,    // stream from channel A
  channelB.messages,    // stream from channel B
…
  8. 08

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

    Hard

    Backpressure is what happens when a producer emits faster than its consumer can process.

    // ── Backpressure with a single-subscription generator ──
Stream<Block> readChunks(File file) async* {
  final raf = await file.open();
  try {
    var offset = 0;
    while (true) {
…
  9. 09

    How do you decide between Stream and Future for an API?

    Medium

    Future = ONE event at SOME point.

    // ── Future — single result ──
class UserService {
  Future<User> getById(String id) async => api.fetchUser(id);
  Future<void> save(User u) async => api.saveUser(u);
}
…
  10. 10

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

    Medium

    StreamBuilder subscribes to the stream, rebuilds on every event, and cancels the subscription when it 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>(
…
  11. 11

    How do you expose a callback-based API — a sensor, a socket, a plugin — as a Stream?

    Medium

    A StreamController turns any push-based callback into a Stream, and its onListen and onCancel hooks are where the underlying resource starts and stops.

    class BatteryService {
  static const _channel = EventChannel('app/battery');
  StreamSubscription<dynamic>? _native;

  late final StreamController<int> _controller = StreamController<int>.broadcast(
    onListen: _start,     // first subscriber: turn the hardware on
…
  12. 12

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

    Medium

    A single-subscription stream accepts exactly one listener for its whole life, and cancelling that listener does not free the slot.

    // ❌ two listeners on a single-subscription stream
final controller = StreamController<int>();
controller.stream.listen(print);
controller.stream.listen(print);   // Bad state: already listened to
// ✅ broadcast when you own the controller
final broadcast = StreamController<int>.broadcast();
…
  13. 13

    Sign-in success shows its SnackBar three times because it is fired from inside a StreamBuilder's builder — what rule is being broken?

    Medium

    The builder is a pure render function that Flutter may call at any time and any number of times, so anything with a side effect belongs on a subscription instead.

    // auth.state is a broadcast stream of the current session

// ❌ a side effect inside builder: runs on every rebuild, not on every event
StreamBuilder<AuthState>(
  stream: auth.state,
  builder: (context, snapshot) {
…
  14. 14

    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?

    Medium

    Silence is not an event, so nothing in the pipeline can notice it — you turn the absence of events into an event with stream.timeout.

    class PriceFeed {
  final _out = StreamController<Quote>.broadcast();
  StreamSubscription<Quote>? _sub;
  int _attempt = 0;

  Stream<Quote> get quotes => _out.stream;   // survives every reconnect
…
  15. 15

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

    Medium

    distinct compares each event with the previously emitted one using ==, and a freshly decoded list of freshly constructed models is never == to the last batch.

    // ❌ as first written: no ==, so identity comparison, so distinct() is a no-op
// class Order { Order(this.id, this.status); final String id, status; }
// repo.pollOrders().distinct().listen(render);   // still fires on every poll

// ✅ value equality: two structurally identical payloads now compare equal
class Order {
…
  16. 16

    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?

    Medium

    A Stream has no current value: first subscribes, waits for the next event after that moment, then cancels — so on a stream that has already emitted, you are waiting for the one after.

    // ❌ hangs: nothing has happened *since* this line started running
final Quote now = await feed.quotes.first;

// ❌ fails differently: a closed, empty stream throws instead of hanging
await const Stream<int>.empty().first;      // StateError: No element
…
  17. 17

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

    Hard

    No — asynchronous delivery is the contract, not a bug, and sync: true "fixes" the symptom by handing arbitrary listener code control in the middle of your own add call.

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

    Shutting a service down hangs on await _controller.close(), and nothing ever listened to that controller — what is close waiting for?

    Hard

    close() returns the controller's done future, and on a single-subscription controller that future completes only once a listener has received the done event — with no listener, the buffered events and the done event just sit there.

    // ❌ never completes: nothing ever listened to _events
class Uploader {
  final _events = StreamController<Progress>();
  Stream<Progress> get events => _events.stream;

  Future<void> shutdown() async {
…
  19. 19

    A flaky socket's addError lands in Crashlytics as an unhandled async error and no widget reacts to it — who was supposed to catch that?

    Hard

    An error event with no onError handler on the subscription is neither swallowed nor thrown at the add site — it goes to Zone.current.handleUncaughtError for the zone in which listen was called.

    void main() {
  // The net: uncaught async errors from any stream land here
  PlatformDispatcher.instance.onError = (error, stack) {
    Crashlytics.instance.recordError(error, stack, fatal: false);
    return true;                        // handled — do not re-report
  };
…
  20. 20

    A socket sends a handshake line before its data frames — how do you read that one line and hand the rest of the single-subscription stream to a parser?

    Hard

    Wrap it in a StreamQueue from package:async: it takes the one subscription for you and turns the stream into a pull API you can read a piece at a time, then gives the remainder back as a stream.

    import 'dart:convert';
import 'package:async/async.dart';

Future<Session> open(Socket socket) async {
  final lines = socket
      .cast<List<int>>()
…