Brain
EngineeringFlutter
#flutter#dart#streams#async

Streams

Stream types, StreamController, transformers, StreamBuilder, and rate limiting with throttle and debounce.

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.

Single subscription and broadcast streams

There are two main types of streams in Flutter/Dart:

Single subscription streams

Stream<int> 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

// Creating a broadcast stream from a single subscription stream
Stream<int> singleStream = countStream();
Stream<int> 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.

import 'dart:async';

class CounterBloc {
  // Create a StreamController
  final _counterController = StreamController<int>();

  // Expose the stream from the controller
  Stream<int> get counterStream => _counterController.stream;

  // Expose the sink for adding events
  Sink<int> 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

    final controller = StreamController<int>();

    Creates a single-subscription stream.

  2. Broadcast StreamController

    final controller = StreamController<int>.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:

Bad state: Stream has already been listened to

A broadcast controller accepts as many listeners as you want:

var controller = StreamController<int>.broadcast();

controller.stream.listen((data) {
  print("Listener1: $data");
});

controller.stream.listen((data) {
  print("Listener2: $data");
});

controller.add(1);
Listener1: 1
Listener2: 1
FeatureStreamControllerStreamController.broadcast
ListenersOneMany
Use caseFile stream, HTTP responseEvents, UI signals
BufferingBuffers until someone listensNo 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

// Using async* and yield
Stream<int> countStream() async* {
  for (int i = 1; i <= 10; i++) {
    await Future.delayed(Duration(seconds: 1));
    yield i;
  }
}

// Using StreamController
final controller = StreamController<int>();
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

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

// 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<S, T> 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.

Stream<int> numberStream = Stream.fromIterable([1, 2, 3, 4, 5]);
Stream<String> stringStream = numberStream.map((number) => 'Number: $number');
where

Filters stream elements based on a condition.

Stream<int> numberStream = Stream.fromIterable([1, 2, 3, 4, 5]);
Stream<int> evenNumbersStream = numberStream.where((number) => number % 2 == 0);
expand

Expands each element into multiple elements.

Stream<int> numberStream = Stream.fromIterable([1, 2, 3]);
Stream<int> expandedStream = numberStream.expand((number) => [number, number * 2]);
// Output: 1, 2, 2, 4, 3, 6
take and skip

Limit the number of events processed.

Stream<int> numberStream = Stream.fromIterable([1, 2, 3, 4, 5]);
Stream<int> takenStream = numberStream.take(3); // First 3 elements
Stream<int> skippedStream = numberStream.skip(2); // Skip first 2 elements
distinct

Removes consecutive duplicates.

Stream<int> numberStream = Stream.fromIterable([1, 1, 2, 2, 3, 3, 3, 4]);
Stream<int> distinctStream = numberStream.distinct();
// Output: 1, 2, 3, 4

Asynchronous transformers

asyncMap

For asynchronous mapping operations.

Stream<int> numberStream = Stream.fromIterable([1, 2, 3]);
Stream<String> asyncMappedStream = numberStream.asyncMap((number) async {
  await Future.delayed(Duration(seconds: 1));
  return 'Processed: $number';
});
asyncExpand

Asynchronously expands elements into multiple elements.

Stream<int> numberStream = Stream.fromIterable([1, 2, 3]);
Stream<String> 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.

Stream<int> errorStream = Stream.fromFuture(Future.error('Oops!'));
Stream<int> handledStream = errorStream.handleError(
  (error) => print('Caught error: $error'),
  test: (error) => error is String
);

Custom transformers

StreamTransformer.fromHandlers
StreamTransformer<int, String> 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<int> numberStream = Stream.fromIterable([-1, 0, 1, 2]);
Stream<String> transformedStream = numberStream.transform(customTransformer);
A custom transformer class
class DoubleTransformer extends StreamTransformerBase<int, int> {
  @override
  Stream<int> bind(Stream<int> stream) {
    return stream.map((value) => value * 2);
  }
}

Stream<int> numberStream = Stream.fromIterable([1, 2, 3]);
Stream<int> doubledStream = numberStream.transform(DoubleTransformer());
// Output: 2, 4, 6

Combining transformers

You can chain multiple transformers together for complex transformations:

Stream<int> numberStream = Stream.fromIterable([1, 2, 3, 4, 5, 6]);
Stream<String> 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.

import 'package:rxdart/rxdart.dart';

Stream<int> numberStream = Stream.periodic(Duration(milliseconds: 300), (i) => i).take(10);
Stream<int> debouncedStream = numberStream.debounceTime(Duration(milliseconds: 500));
bufferCount

Collects items from the source stream and emits them as a list.

Stream<int> numberStream = Stream.periodic(Duration(milliseconds: 100), (i) => i).take(10);
Stream<List<int>> 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.

Stream<int> triggerStream = Stream.periodic(Duration(seconds: 2), (i) => i).take(3);
Stream<String> 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:

StreamBuilder<int>(
  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:

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:

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:

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:

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

AspectThrottleDebounce
Execution timingAt the start of the intervalAfter the quiet period
Behavior with rapid callsExecutes periodicallyWaits until calls stop
Number of executionsMultiple, at fixed intervalsAt most once, after waiting
Best forRegular updates during continuous activityProcessing 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.

Async and the Event Loop

Futures, async and await, the microtask and event queues, and error handling in async code.

Widgets, State, and Lifecycle

State and setState, keys, BuildContext, the widget and app lifecycles, InheritedWidget, and const constructors.

On this page

What a stream isSingle subscription and broadcast streamsSingle subscription streamsBroadcast streamsStreamControllerThe two kinds of controllerOne listener, or manyWorking with streamsCreating a streamListening to a streamTransforming a streamStream transformersWhat a transformer isKinds of transformerBuilt in transformersmapwhereexpandtake and skipdistinctAsynchronous transformersasyncMapasyncExpandError handling transformershandleErrorCustom transformersStreamTransformer.fromHandlersA custom transformer classCombining transformersRxDart transformersdebounceTimebufferCountswitchMapStreamBuilderThrottle and debounceThrottleDebounceSide by sidePicking oneBest practices