--- title: "Streams" description: "Stream types, StreamController, transformers, StreamBuilder, and rate limiting with throttle and debounce." tags: [flutter, dart, streams, async] --- ## What a stream is A stream is a sequence of asynchronous events. It's like an asynchronous Iterable: instead of getting the next event when you ask for it, the stream tells you when an event is ready. Streams are particularly useful for: * Real-time data (like location updates) * UI events (button clicks, scrolling) * WebSocket connections * Long-running operations that produce multiple values A `Future` delivers one value and completes. A stream delivers many values over time and only ends when it is closed. Futures and the event loop that drives both live in [async-and-event-loop](https://brain.narayann.dev/notes/engineering/flutter/async-and-event-loop). *** ## Single subscription and broadcast streams There are two main types of streams in Flutter/Dart: ### Single subscription streams ```dart Stream countStream() async* { for (int i = 1; i <= 10; i++) { await Future.delayed(Duration(seconds: 1)); yield i; } } ``` **Characteristics:** * Can have only one listener during its lifetime * If you try to listen again after canceling, it will throw an exception * Events are delivered in sequence * Used for reading files, network requests * Preserves the order of events ### Broadcast streams ```dart // Creating a broadcast stream from a single subscription stream Stream singleStream = countStream(); Stream broadcastStream = singleStream.asBroadcastStream(); ``` **Characteristics:** * Can have multiple listeners at the same time * Can be listened to, canceled, and listened to again * Used for UI events, notifications * Great for cases where multiple parts of your application need to respond to the same events *** ## StreamController A StreamController is a class that creates and manages a Stream. It gives you control over adding events to the stream. ```dart import 'dart:async'; class CounterBloc { // Create a StreamController final _counterController = StreamController(); // Expose the stream from the controller Stream get counterStream => _counterController.stream; // Expose the sink for adding events Sink get counterSink => _counterController.sink; // Add values to the stream void increment(int current) { counterSink.add(current + 1); } // Always close controllers when done void dispose() { _counterController.close(); } } ``` ### The two kinds of controller 1. **Regular StreamController** ```dart final controller = StreamController(); ``` Creates a single-subscription stream. 2. **Broadcast StreamController** ```dart final controller = StreamController.broadcast(); ``` Creates a broadcast stream that allows multiple listeners. ### One listener, or many A plain `StreamController` gives you a single subscription stream, so a second `listen` throws: ```text Bad state: Stream has already been listened to ``` A broadcast controller accepts as many listeners as you want: ```dart var controller = StreamController.broadcast(); controller.stream.listen((data) { print("Listener1: $data"); }); controller.stream.listen((data) { print("Listener2: $data"); }); controller.add(1); ``` ```text Listener1: 1 Listener2: 1 ``` | Feature | `StreamController` | `StreamController.broadcast` | | --------- | ----------------------------- | ---------------------------- | | Listeners | One | Many | | Use case | File stream, HTTP response | Events, UI signals | | Buffering | Buffers until someone listens | No buffering | A broadcast stream behaves like an event bus: values sent while nobody is listening are gone. *** ## Working with streams Three things you do with any stream: create one, listen to it, and reshape what comes out. ### Creating a stream ```dart // Using async* and yield Stream countStream() async* { for (int i = 1; i <= 10; i++) { await Future.delayed(Duration(seconds: 1)); yield i; } } // Using StreamController final controller = StreamController(); controller.add(1); // Add values to the stream controller.addError('Error occurred'); // Add errors controller.close(); // Close the stream when done // From Futures Stream.fromFuture(Future.delayed(Duration(seconds: 2), () => 42)); // Periodic stream Stream.periodic(Duration(seconds: 1), (count) => count).take(10); ``` ### Listening to a stream ```dart final subscription = myStream.listen( (data) => print('Data: $data'), // onData onError: (error) => print('Error: $error'), // onError onDone: () => print('Stream is done'), // onDone cancelOnError: false // Continue listening after errors ); // Cancel subscription when done subscription.cancel(); ``` ### Transforming a stream ```dart // Map values myStream.map((value) => value * 2); // Filter values myStream.where((value) => value % 2 == 0); // Combine with another stream myStream.merge(otherStream); // Process events in order myStream.asyncMap((value) async { await Future.delayed(Duration(milliseconds: 500)); return value * 2; }); ``` *** ## Stream transformers A transformer sits between a stream and its listener, reshaping events on the way through: filtering them, mapping them to another type, or merging them with another stream. ### What a transformer is A stream transformer takes a stream as input, processes its events, and outputs a new stream of potentially different types. The `StreamTransformer` class is the base class for all transformers, where: * `S` is the input type * `T` is the output type ### Kinds of transformer #### Built in transformers Dart provides several built-in stream transformation methods: ##### map Converts each element in a stream. ```dart Stream numberStream = Stream.fromIterable([1, 2, 3, 4, 5]); Stream stringStream = numberStream.map((number) => 'Number: $number'); ``` ##### where Filters stream elements based on a condition. ```dart Stream numberStream = Stream.fromIterable([1, 2, 3, 4, 5]); Stream evenNumbersStream = numberStream.where((number) => number % 2 == 0); ``` ##### expand Expands each element into multiple elements. ```dart Stream numberStream = Stream.fromIterable([1, 2, 3]); Stream expandedStream = numberStream.expand((number) => [number, number * 2]); // Output: 1, 2, 2, 4, 3, 6 ``` ##### take and skip Limit the number of events processed. ```dart Stream numberStream = Stream.fromIterable([1, 2, 3, 4, 5]); Stream takenStream = numberStream.take(3); // First 3 elements Stream skippedStream = numberStream.skip(2); // Skip first 2 elements ``` ##### distinct Removes consecutive duplicates. ```dart Stream numberStream = Stream.fromIterable([1, 1, 2, 2, 3, 3, 3, 4]); Stream distinctStream = numberStream.distinct(); // Output: 1, 2, 3, 4 ``` #### Asynchronous transformers ##### asyncMap For asynchronous mapping operations. ```dart Stream numberStream = Stream.fromIterable([1, 2, 3]); Stream asyncMappedStream = numberStream.asyncMap((number) async { await Future.delayed(Duration(seconds: 1)); return 'Processed: $number'; }); ``` ##### asyncExpand Asynchronously expands elements into multiple elements. ```dart Stream numberStream = Stream.fromIterable([1, 2, 3]); Stream asyncExpandedStream = numberStream.asyncExpand((number) async* { await Future.delayed(Duration(milliseconds: 500)); yield 'First: $number'; await Future.delayed(Duration(milliseconds: 500)); yield 'Second: $number'; }); ``` #### Error handling transformers ##### handleError Catches errors and provides recovery options. ```dart Stream errorStream = Stream.fromFuture(Future.error('Oops!')); Stream handledStream = errorStream.handleError( (error) => print('Caught error: $error'), test: (error) => error is String ); ``` #### Custom transformers ##### StreamTransformer.fromHandlers ```dart StreamTransformer customTransformer = StreamTransformer.fromHandlers( handleData: (data, sink) { if (data > 0) { sink.add('Positive: $data'); } else { sink.add('Non-positive: $data'); } }, handleError: (error, stackTrace, sink) { sink.add('Error occurred: $error'); }, handleDone: (sink) { sink.add('Stream completed'); sink.close(); }, ); Stream numberStream = Stream.fromIterable([-1, 0, 1, 2]); Stream transformedStream = numberStream.transform(customTransformer); ``` ##### A custom transformer class ```dart class DoubleTransformer extends StreamTransformerBase { @override Stream bind(Stream stream) { return stream.map((value) => value * 2); } } Stream numberStream = Stream.fromIterable([1, 2, 3]); Stream doubledStream = numberStream.transform(DoubleTransformer()); // Output: 2, 4, 6 ``` #### Combining transformers You can chain multiple transformers together for complex transformations: ```dart Stream numberStream = Stream.fromIterable([1, 2, 3, 4, 5, 6]); Stream processedStream = numberStream .where((number) => number % 2 == 0) // Filter even numbers .map((number) => number * 10) // Multiply by 10 .map((number) => 'Processed: $number'); // Convert to string // Output: "Processed: 20", "Processed: 40", "Processed: 60" ``` #### RxDart transformers The RxDart package extends Dart's Stream API with additional transformers: ##### debounceTime Emits an item after a specific duration has passed without another item being emitted. ```dart import 'package:rxdart/rxdart.dart'; Stream numberStream = Stream.periodic(Duration(milliseconds: 300), (i) => i).take(10); Stream debouncedStream = numberStream.debounceTime(Duration(milliseconds: 500)); ``` ##### bufferCount Collects items from the source stream and emits them as a list. ```dart Stream numberStream = Stream.periodic(Duration(milliseconds: 100), (i) => i).take(10); Stream> bufferedStream = numberStream.bufferCount(3); // Output: [0,1,2], [3,4,5], [6,7,8], [9] ``` ##### switchMap Switches to a new stream based on each item from the source stream. ```dart Stream triggerStream = Stream.periodic(Duration(seconds: 2), (i) => i).take(3); Stream switchedStream = triggerStream.switchMap((i) => Stream.periodic(Duration(milliseconds: 500), (j) => 'Source $i: Item $j').take(3) ); ``` *** ## StreamBuilder Flutter provides the StreamBuilder widget for building UI based on stream values: ```dart StreamBuilder( stream: counterBloc.counterStream, initialData: 0, // Optional initial data builder: (context, snapshot) { if (snapshot.hasError) { return Text('Error: ${snapshot.error}'); } if (snapshot.connectionState == ConnectionState.waiting) { return CircularProgressIndicator(); } return Text('Count: ${snapshot.data}'); }, ) ``` *** ## Throttle and debounce Both throttle and debounce are techniques used to control how frequently a function executes, especially when handling rapidly firing events like scrolling, resizing, or typing. While they seem similar, they serve different purposes and behave differently. ### Throttle **Definition**: Throttle limits the execution of a function to once per specified time interval, regardless of how many times the function is triggered during that interval. **Behavior**: If a throttled function is called multiple times within the specified interval, it will execute once at the beginning of the interval, then ignore subsequent calls until the interval has passed. **Visual Representation**: ```text Calls: | | | | | | | | | | | | | | Throttled: | | | | ↑ ↑ ↑ ↑ Time → 0s 1s 2s 3s ``` **Use Cases**: * Scroll event handling * Mousemove events * Window resize events * Game input controls * Continuous button presses **Example in Dart**: ```dart DateTime _lastExecution = DateTime.fromMillisecondsSinceEpoch(0); final int _throttleTimeMs = 1000; // 1 second void throttledFunction() { final now = DateTime.now(); if (now.difference(_lastExecution).inMilliseconds > _throttleTimeMs) { _lastExecution = now; // Execute your actual function here print('Function executed at ${now.toString()}'); } } ``` ### Debounce **Definition**: Debounce delays the execution of a function until after a specified wait period has elapsed since the last time the function was triggered. **Behavior**: If a debounced function is called multiple times within the wait period, all calls are ignored except the last one, which executes after the wait period. **Visual Representation**: ```text Calls: | | | | | | | Debounced: | | | ↑ ↑ ↑ Time → 0s 1s 2s 3s ``` **Use Cases**: * Search input (execute search after user stops typing) * Form validation * API requests triggered by user input * Saving drafts automatically * Auto-complete suggestions **Example in Dart**: ```dart Timer? _debounceTimer; final int _debounceTimeMs = 1000; // 1 second void debouncedFunction() { if (_debounceTimer?.isActive ?? false) { _debounceTimer!.cancel(); } _debounceTimer = Timer(Duration(milliseconds: _debounceTimeMs), () { // Execute your actual function here print('Function executed at ${DateTime.now().toString()}'); }); } ``` ### Side by side | Aspect | Throttle | Debounce | | ------------------------- | ------------------------------------------ | ------------------------------------------------ | | Execution timing | At the start of the interval | After the quiet period | | Behavior with rapid calls | Executes periodically | Waits until calls stop | | Number of executions | Multiple, at fixed intervals | At most once, after waiting | | Best for | Regular updates during continuous activity | Processing final state after activity ends | | Metaphor | "Execute at most once every X seconds" | "Wait until X seconds of quiet before executing" | ### Picking one **Use Throttle When**: * You need regular updates during continuous events * You want to ensure a minimum time between function executions * The intermediate states are important **Use Debounce When**: * You only care about the final state after a series of events * You want to wait for user input to stabilize before processing * You need to reduce API calls or expensive operations Throttle when you want steady updates during a continuous action. Debounce when you only care about the final value. *** ## Best practices 1. **Always close StreamControllers** when they're no longer needed to prevent memory leaks 2. **Cancel stream subscriptions** when they're no longer needed 3. **Handle errors** appropriately in your streams 4. **Use broadcast streams** when multiple listeners are required 5. **Consider using StreamBuilder** for UI updates based on stream data 6. **Use RxDart** for more advanced stream operations (like combining streams) Most stream bugs come down to two things: a controller that was never closed, and a listener that outlived the widget that created it. *** For transformers specifically: 1. **Choose the right transformer** for your use case, don't overengineer 2. **Consider performance implications** for high-frequency streams 3. **Handle errors properly** throughout the transformation chain 4. **Dispose of subscriptions** to prevent memory leaks 5. **Use RxDart** for complex stream operations when needed 6. **Test transformers** with various inputs including edge cases 7. **Chain transformers thoughtfully**, since excessive chaining can impact readability Stream transformers are essential tools for reactive programming in Flutter, making data processing more declarative and maintainable. ***