Streams & Rx

A Future delivers one result or one error, once; a Stream delivers any number of values and errors over time and may then finish. Use a Future for one-shot work and a Stream when values keep arriving. **Future**: fetch a user, submit a form, read a whole file. The caller writes await, and an error simply throws **Stream: websocket messages, sensor readings, download progress, auth state changes. The caller reacts to each value and owns a subscription it has to cancel Converting**: stream.first and stream.toList() give a Future; future.asStream() gives a one-event stream Smell: a stream that always emits exactly one event and closes is a Future in disguise

A Stream<T> is an asynchronous sequence of events: data, errors and a final done. A single-subscription stream allows one listener for its whole life and buffers events until it arrives; a broadcast stream allows many listeners but drops events nobody is listening to. Single-subscription (the default): one listener, ever; the listener can pause the producer. File reads, HTTP bodies, sockets, async* generators Broadcast: any number of listeners joining and leaving; a late listener sees only events from that moment on. UI events, an event bus, StreamController.broadcast() Checking and adapting: stream.isBroadcast tells you which one you have; asBroadcastStream() lets several listeners share a single-subscription stream Listening: listen returns a StreamSubscription to cancel later; await for loops over the events inside an async function

It prints only world: a broadcast controller with no listeners discards hello on the spot. Without .broadcast() it prints hello and then world, because a single-subscription controller buffers events until its one listener arrives. Why: a broadcast stream has no memory; a listener sees only events added after it subscribed When a late listener needs the latest value: keep it in a field, use rxdart's BehaviorSubject, or hold state in Bloc or Riverpod Delivery is still asynchronous: world is printed in a later microtask, after add has returned

FutureBuilder waits for one Future and rebuilds when it completes; StreamBuilder subscribes to a Stream and rebuilds on every event until it closes. Both hand the builder an AsyncSnapshot with connectionState, data and error. States: a future goes waiting → done; a stream goes waiting → active (one rebuild per event) → done when it closes Subscription: StreamBuilder cancels its subscription when it leaves the tree or receives a different stream instance Same pitfall: create the future or stream in initState, a field or a provider, never inline in build, or every rebuild restarts the request or re-subscribes Pick by source: a one-off request → FutureBuilder; Firestore snapshots(), auth state, a websocket → StreamBuilder

A StreamController pairs a sink you push into (add, addError, close) with a stream you hand out. To wrap a callback API, register the callback in onListen, forward each call with add, and unregister it in onCancel. Start lazily: registering the callback in onListen means nothing runs, and no battery is spent, until someone listens Stop with the last listener: on a broadcast controller onListen and onCancel fire for the first and the last listener, so the source can start and stop many times Report properly: failures go through addError, not throw; call close() when the source is finished for good; check isClosed before add, because adding after close throws StateError Pick the kind: the default controller buffers and supports pause for one owner such as a repository; .broadcast() when several widgets listen, accepting that events sent while nobody listens are lost Hide the controller: return Stream<T>, so callers cannot push into it or close it When not to: if the source is already a loop over futures, an async* generator is shorter and cleans up in finally

StreamSubscription is the handle listen returns: through it you pause, resume and cancel delivery. In a State you keep it in a field, create it in initState and cancel it in dispose, or the callback keeps firing after the widget is gone. **cancel()**: stops delivery; on a single-subscription stream it also stops the producer (an async* generator runs its finally, a file read closes the file). On a broadcast stream it only removes you **cancel() returns a Future: await it when cleanup must finish first, such as closing a file before deleting it; calling it twice is safe pause() / resume(): events buffer while paused; on a single-subscription stream the producer pauses too The widget gets a new stream**: in didUpdateWidget cancel the old subscription and listen to the new one What a forgotten cancel costs: setState() called after dispose() errors, and the closure keeps the whole State and its context alive, which is a leak

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. **await for**: sequential, with backpressure: while the body awaits, a single-subscription source is paused. Errors arrive as exceptions you catch with try/catch, and the first one ends the loop **listen**: the tool for widgets and services: you get onError, onDone and a handle to cancel in dispose. An async callback is not awaited, so the next event can start while the previous handler still runs Ending early: break or return inside await for cancels the subscription, but nothing outside the function can stop the loop Common bug: await for over a stream that never closes (a socket, auth state) means the code after the loop never runs

It prints created, listened, start, 1, 2. Calling an async* function only creates the stream; the body starts after someone listens, and even then in a later microtask, so the synchronous code after listen runs first. Lazy, or "cold": with no listener, start is never printed; the work begins on listen, not on the call Asynchronous delivery: even the first yield reaches the listener in a microtask, never inside the listen call **Paused at yield**: while the listener is paused, the generator waits at its yield; cancelling makes that yield return and runs the finally blocks Single-subscription: a second stream.listen throws; call numbers() again to get a fresh run

StreamBuilder subscribes to a stream for you, rebuilds its builder on every event and cancels the subscription when the widget leaves the tree. Everything it knows arrives in an AsyncSnapshot. **connectionState**: waiting before the first event, active while events flow, done when the stream closed. done does not mean success: an empty closed stream never had data **data and error**: the latest event; the snapshot never holds both, so an error event resets data to null. Keep the last good value yourself if it must survive an error **initialData: fills the first frame so the UI does not flash a spinner Common mistake**: creating the stream inline in the stream: argument gives a new instance on every rebuild, so StreamBuilder re-subscribes, restarts the query and drops back to waiting; if the new stream wraps a single-subscription source, the second listen throws. Keep the stream in a field or a provider

The stream is single-subscription: it allows exactly one listener for its whole life, and cancelling that listener does not free the slot. The second widget's listen is the one that throws. More things count as listening than people expect: listen, await for, first, toList, forEach and every StreamBuilder each take the one subscription, so even one widget hits this when it subscribes again, for example a listen inside build, or a StreamBuilder handed a new stream object that wraps the same single-subscription source. Where they come from: a plain StreamController(), an async* generator, file and socket reads, HTTP bodies Best fix: subscribe once, feed a store: one listener updates a ValueNotifier, Bloc or provider, and every widget watches that; late widgets then also see the current value **StreamController.broadcast(): when you own the controller and truly need several independent listeners asBroadcastStream()**: to adapt a stream you do not own; broadcast is not free, though: late subscribers miss what already happened and there is no backpressure

Errors are events: addError, or a throw inside an async* body, sends them down the same pipeline as data to the listener's onError. If the listener has no onError, the error goes to the zone's uncaught-error handler (in Flutter, PlatformDispatcher.instance.onError). **onError in listen**: receives the error and its stack trace; listen(onData) alone handles data and lets every error escape **handleError**: catches errors mid-pipeline; its test callback lets you swallow only the types you expect **await for**: the error is thrown at the loop, so wrap it in try/catch. await for, first and forEach stop at the first error, while listen keeps going unless you pass cancelOnError: true **A throw in async* does not crash the caller; it becomes an error event and the generator ends Errors as state**: at the repository or Bloc boundary map transport failures to a state like Failed(e), so the UI renders a retry instead of a dead stream **Not FlutterError.onError**: that handler gets framework errors from build, layout and gestures, not stream errors

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. Why: a stream ends only when its producer says so: close() on the controller, or the end of an async* body. The upload Future completing tells the stream nothing, and await for cannot tell "no events yet" from "no events ever" How the fix works: whenComplete(progress.close) closes the controller after success and after failure alike, so the loop either ends normally or throws the upload error Where else it bites: toList(), last, length, drain() and a StreamBuilder waiting for ConnectionState.done all wait for done as well Catch it next time: enable the close_sinks lint, which flags a StreamController created in a function that never calls close; in a test, expect(stream, emitsInOrder([..., emitsDone])) hangs until the timeout when done never comes

A StreamTransformer<S, T> turns one stream into another and is applied with stream.transform(...); utf8.decoder and LineSplitter are transformers. For your own operator, StreamTransformer.fromHandlers covers per-event logic, and an extension method with async* covers stateful logic. **fromHandlers**: handleData and handleError get the event and an EventSink, handleDone only the sink, so one input can produce zero, one or many outputs **Extension with async*** reads best for stateful operators like scan; the result is single-subscription, gets backpressure for free, and ends at the first source error **StreamTransformer.fromBind or StreamTransformerBase: when you need full control: timers, keeping broadcast-ness, custom cancellation Keep the contract: forward errors you do not handle, cancel the source when the output is cancelled, close the output when the input is done Built-ins first**: map, where, expand, asyncMap, asyncExpand, distinct, take, skip and timeout cover most needs

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. **merge**: one output in arrival order for several sources of the same type, like two chat channels; it ends when all inputs end. Rx.merge in rxdart or StreamGroup.merge from package:async **zip: pairs by index and ends when the shortest input ends, so the slowest input sets the pace combineLatest**: derived state, such as "the form is valid when email and password are valid"; it emits nothing until every input has emitted once, so seed inputs with startWith or BehaviorSubject.seeded Common mistake: zip for state: if one input emits twice as often, the pairs drift out of sync and the unmatched backlog grows in memory

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. **BehaviorSubject**: a broadcast controller that remembers the last value and replays it to every new listener; .value reads it synchronously. This is what makes a stream usable as state: a widget that subscribes late still sees the current value **PublishSubject and ReplaySubject: the first is a plain broadcast controller; the second replays the last N or all events, useful for logs and caches debounceTime: emits only after a pause in events; the search-box operator switchMap: maps each event to an inner stream and cancels the previous one, so a stale result for "flu" never lands after "flut" More**: combineLatest, withLatestFrom, startWith, throttleTime, bufferTime When not to add it: with Riverpod or Bloc most of this is covered (Bloc has restartable() and custom transformers); for one small pipeline an async* operator is enough Common mistake: a subject is a controller you own, so closing it in dispose is your job

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. Compare by content with your own equals or give the elements value equality. Lists are compared by reference: pass distinct(const ListEquality<Order>().equals) from package:collection (it needs == on Order), or a cheap key: distinct((a, b) => a.etag == b.etag) beats deep-comparing a thousand rows Give the model value equality: override == and hashCode, use a record, or generate it with Equatable or freezed Narrow first: map to exactly the slice the widget draws, then distinct, so a badge watches an int rather than a list Per subscription: distinct remembers the previous event per listener, so a late subscriber's first event always passes

You assumed a stream has a current value. It does not: first subscribes and waits for the next event after that moment, so on a broadcast stream that already emitted the price you are waiting for the next one, which may never come. **What first does: listens, takes the first event that arrives, cancels and completes the future; on a broadcast stream everything before that instant is gone Other traps**: if the stream closes without an event, first throws StateError: No element; on a single-subscription stream it uses up the only subscription Keep the value somewhere readable: a field in the repository, a ValueNotifier, a BehaviorSubject with .value, or Bloc/Riverpod state Design the API as a pair: Future<T> current() for "what is it now" and Stream<T> changes() for "tell me when it moves" Mental model: a stream is a sequence of things that happen, not a variable you can read.

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. Why three times: the builder runs on every rebuild, not only on stream events: a parent rebuild, a theme change, a route animation frame or a hot reload re-runs it with the same snapshot, and the snapshot keeps the "signed in" value as long as the stream is open **Why addPostFrameCallback does not help: it dodges the "called during build" exception but fires just as often Where it goes**: subscribe once in initState, keep the latest state in a field and show the SnackBar in the listen callback; in bloc that is BlocListener, in Riverpod ref.listen **Check mounted**: only if the callback awaits something before touching context Rule of thumb: the builder answers "what does this state look like"; the listener answers "what happens now that this occurred".

open overwrites _sub without cancelling the previous subscription, so every chat opened so far keeps listening and its messages land in the same list. dispose cancels only the last one; the others leak for as long as their streams live. Why: assigning a new value to a field cancels nothing: the old StreamSubscription stays active, and its closure keeps the controller and the ValueNotifier reachable How the fix works: _sub?.cancel() at the top of open stops the previous chat before the next one starts; cancelling twice, or cancelling null, is harmless Same bug, other places: listen inside refresh(), in didUpdateWidget without cancelling the old subscription, or on every button tap: each call adds a listener, so events arrive two or three times Let a stream do the switching: model the open chat as a stream of ids and use switchMap from rxdart, which cancels the previous inner stream for you Catch it next time: the cancel_subscriptions lint stays silent here, because _sub is cancelled in dispose; a test that opens chat A, then B, pushes into A and expects B's list unchanged fails at once

Streams model state as a sequence of values: the source emits a new immutable state and widgets rebuild through StreamBuilder. Bloc is built on this, but a raw stream has no current value, no replay for late widgets and no owner, so apps use Bloc, Riverpod or a BehaviorSubject. The reactive idea: the UI is a function of state; widgets do not ask "what is it now", they are told when it changes What a raw stream lacks: a synchronous current value for the first frame, replay for widgets that subscribe later, skipping equal states, and someone responsible for closing the controller Bloc: bloc.stream is a broadcast Stream<State> plus a synchronous state getter; BlocBuilder adds buildWhen and disposal Riverpod: StreamProvider turns a stream into an AsyncValue with loading and error states; most state lives in a Notifier or AsyncNotifier Where raw streams still fit: data sources that are streams by nature: Firestore snapshots(), Drift watch queries, websockets, connectivity. Adapt them once into the state layer

Backpressure is a slow consumer making a faster producer wait. In Dart it works only for single-subscription streams whose producer actually waits: an async* generator suspends at yield while the listener is paused; a broadcast stream cannot pause its source, so a slow listener just buffers. How a pause travels: subscription.pause(), or await for while its body awaits, pauses the controller; onPause fires and a generator stops at its next yield. HTTP bodies and File.openRead stop reading the same way **controller.add never blocks**: if your code keeps calling add on a paused controller, events queue in memory; real backpressure needs a producer that checks isPaused or reacts to onPause / onResume **An async callback in listen gives none**: the stream does not await it, so the next event is delivered at once; use await for or asyncMap, which pauses the source while each future runs Broadcast: pausing one listener buffers events for that listener only; the source keeps going. Fine for taps, wrong for bulk data When the source cannot slow down: sensors, scroll, a price feed: shed load instead with throttleTime, sampleTime or bufferTime from rxdart, or keep only the latest value Production symptom: memory climbs steadily during a large import: an unbounded buffer between a fast producer and a consumer that never pushed back

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. Nothing else in the pipeline can notice that events stopped. **What timeout does**: the countdown starts on listen and resets on every event; on expiry it emits an error, or whatever your onTimeout callback adds to the sink It does not touch the source: the socket stays open and a late event still flows; if you want the connection dropped, cancel it yourself Heartbeat: a dead mobile connection can stay silently open for minutes, so send protocol pings (pingInterval on IOWebSocketChannel) and require replies Reconnect means a new stream: a closed channel never reopens; keep the UI on a stream you own and swap the socket underneath it with exponential backoff Show stale as a state: a price that may be wrong, with no marker, is the bug the user reports

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. sync: true hides the symptom by running arbitrary listener code in the middle of your own add call. Why delivery is async: you can add before anyone listens (the controller buffers), listener code never runs in the middle of the producer's own method, and every listener sees events in emission order **What sync: true really does**: the listener runs on the add call stack, but only when a subscription exists and is not paused; otherwise the event buffers as usual. "Sometimes synchronous" is the worst contract Re-entrancy: with sync: true the listener runs on the producer's stack and can call back into it (setState, a bloc event) while the producer is halfway through updating its own state; on a broadcast sync controller a nested add throws Bad state: Cannot fire new event. A listener exception still does not reach add, it goes to the zone's uncaught-error handler The real fix in tests: let the queue turn with await pumpEventQueue(); better, assert on the stream with expectLater and stream matchers, and use fakeAsync when timers like debounce are involved