Async #

dart:async is Dart’s built-in library that forms the foundation of the entire concurrency model — without it, there’s no Future, Stream, async/await, or StreamController. Although the async and await keywords feel like language features, behind the scenes they work entirely through dart:async. Understanding this library deeply — especially Stream and how to control it — is the difference between developers who merely use async and developers who truly understand Dart’s event loop.

An Overview of dart:async #

flowchart LR
    DA["dart:async"] --> FUT["Future\nA single value in the future"]
    DA --> STR["Stream\nA sequence of future values"]
    DA --> SC["StreamController\nCreate and control streams"]
    DA --> COMP["Completer\nCreate Futures manually"]
    DA --> ZONE["Zone\nAsync execution context"]
    DA --> TIMER["Timer\nSchedule execution"]

Future — a Single Value in the Future #

Future represents a value that may not be available yet — the result of an async operation like an HTTP request, file read, or database query:

import 'dart:async';

// A Future already completed with a value
Future<int> futureLangsung = Future.value(42);
Future<String> futureError = Future.error(Exception('Gagal'));

// A Future with a delay
Future<String> futureDelay = Future.delayed(
  Duration(seconds: 2),
  () => 'Done after 2 seconds',
);

// Create a Future from synchronous computation
Future<int> futureSync = Future(() => 1 + 1);

// async/await — the most idiomatic way
Future<String> ambilData() async {
  final result = await Future.delayed(Duration(seconds: 1), () => 'data');
  return result.toUpperCase();
}

Future.wait — Parallelism #

// Run several Futures in parallel — wait for all to complete
Future<void> contohParalel() async {
  final mulai = DateTime.now();

  // Sequential — ~3 seconds total
  final a = await Future.delayed(Duration(seconds: 1), () => 'A');
  final b = await Future.delayed(Duration(seconds: 1), () => 'B');
  final c = await Future.delayed(Duration(seconds: 1), () => 'C');

  // Parallel — ~1 second total
  final hasil = await Future.wait([
    Future.delayed(Duration(seconds: 1), () => 'A'),
    Future.delayed(Duration(seconds: 1), () => 'B'),
    Future.delayed(Duration(seconds: 1), () => 'C'),
  ]);
  print(hasil); // ['A', 'B', 'C']

  final durasi = DateTime.now().difference(mulai);
  print('Finished in ${durasi.inSeconds} seconds');
}

// Future.wait with error handling
Future<void> waitDenganError() async {
  try {
    final hasil = await Future.wait([
      Future.value(1),
      Future.error(Exception('Gagal')),
      Future.value(3),
    ]);
  } catch (e) {
    print('One of them failed: $e');
    // Future.wait fails immediately when one Future errors
  }

  // eagerError: false — wait for all to finish even if some error
  final hasil = await Future.wait(
    [Future.value(1), Future.error('error'), Future.value(3)],
    eagerError: false,
  );
}

Important Future Methods #

Future<int> nilai = Future.value(42);

// then — transform the value after completion
nilai.then((n) => print('Nilai: $n'));

// then chaining — every then returns a new Future
Future<String> transformed = nilai
    .then((n) => n * 2)       // 84
    .then((n) => 'Hasil: $n'); // 'Hasil: 84'

// catchError — catch errors
nilai
    .then((n) => throw Exception('Gagal'))
    .catchError((e) => print('Error: $e'));

// whenComplete — always executed, like finally
nilai.whenComplete(() => print('Selesai (selalu)'));

// timeout — throw a TimeoutException if it doesn't finish in time
final result = await nilai.timeout(
  Duration(seconds: 5),
  onTimeout: () => -1, // fallback value
);

// Future.any — the value from the first Future that completes
final pertama = await Future.any([
  Future.delayed(Duration(seconds: 3), () => 'lambat'),
  Future.delayed(Duration(seconds: 1), () => 'cepat'),
]);
print(pertama); // 'cepat'

Stream — a Sequence of Future Values #

Stream is an asynchronous sequence of values — like a Future but can emit many values over time:

import 'dart:async';

// A simple stream from an iterable
Stream<int> angka = Stream.fromIterable([1, 2, 3, 4, 5]);

// A stream with delays between values
Stream<int> hitungMundur = Stream.periodic(
  Duration(seconds: 1),
  (i) => 5 - i,
).take(5); // 5, 4, 3, 2, 1

// async* generator — the most flexible way to create streams
Stream<int> fibonacci() async* {
  int a = 0, b = 1;
  while (true) {
    yield a;
    final temp = a + b;
    a = b;
    b = temp;
  }
}

// Read a stream
await for (final n in fibonacci().take(10)) {
  print(n); // 0, 1, 1, 2, 3, 5, 8, 13, 21, 34
}

Stream Types: Single-Subscription vs Broadcast #

// Single-subscription stream (default)
// Can only be listened to by ONE listener at a time
final single = Stream.fromIterable([1, 2, 3]);
single.listen(print); // ✓
// single.listen(print); // ✗ StateError: Stream has already been listened to

// Broadcast stream — can be listened to by many listeners at once
final broadcast = Stream.fromIterable([1, 2, 3]).asBroadcastStream();
broadcast.listen((n) => print('Listener 1: $n'));
broadcast.listen((n) => print('Listener 2: $n'));
// Both receive all values

// Check the stream type
print(single.isBroadcast);    // false
print(broadcast.isBroadcast); // true

Listening to Streams #

Stream<int> stream = Stream.periodic(Duration(milliseconds: 500), (i) => i).take(5);

// Way 1: await for (most idiomatic)
await for (final nilai in stream) {
  print(nilai);
}

// Way 2: listen (more control)
final subscription = stream.listen(
  (nilai) => print('Data: $nilai'),        // onData
  onError: (e) => print('Error: $e'),       // onError
  onDone: () => print('Stream done'),    // onDone
  cancelOnError: false,                     // continue even with errors
);

// Subscription control
await subscription.pause(Future.delayed(Duration(seconds: 1))); // pause 1 second
subscription.resume();
await subscription.cancel(); // stop listening

Stream Transformations #

Streams support all the functional methods similar to Iterable, but working asynchronously:

Stream<int> angka = Stream.fromIterable([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);

// map — transform every value
Stream<String> label = angka.map((n) => 'item-$n');

// where — filter values
Stream<int> genap = angka.where((n) => n.isEven); // 2, 4, 6, 8, 10

// take and skip
Stream<int> tiga = angka.take(3);      // 1, 2, 3
Stream<int> sisanya = angka.skip(7);   // 8, 9, 10

// takeWhile and skipWhile
Stream<int> kurangDari5 = angka.takeWhile((n) => n < 5); // 1, 2, 3, 4
Stream<int> setelah5 = angka.skipWhile((n) => n <= 5);   // 6, 7, 8, 9, 10

// expand — one value becomes many values
Stream<int> diperluas = angka.expand((n) => [n, n * 10]);
// 1, 10, 2, 20, 3, 30, ...

// asyncMap — map with an async operation
Stream<String> dariDB = angka.asyncMap(
  (id) async => await ambilNamaDariDatabase(id),
);

// asyncExpand — expand with an async operation
Stream<String> diperluasAsync = angka.asyncExpand(
  (id) => ambilTagDariDatabase(id), // returns a Stream
);

// distinct — remove consecutive duplicate values
Stream<int> unik = Stream.fromIterable([1, 1, 2, 2, 3, 1, 1]).distinct();
// 1, 2, 3, 1

// handleError — catch errors inside the pipeline
Stream<int> aman = angka
    .map((n) { if (n == 5) throw Exception('Lima!'); return n; })
    .handleError(
      (e) => print('Error caught: $e'),
      test: (e) => e is Exception,
    );

// fold — aggregate all values into one
int jumlah = await angka.fold(0, (acc, n) => acc + n); // 55

// reduce — fold without an initial value
int maks = await angka.reduce((a, b) => a > b ? a : b); // 10

// toList — collect all values into a List
List<int> semuaNilai = await angka.toList();

// first, last, single
int pertama = await angka.first;
int terakhir = await angka.last;
bool ada = await angka.any((n) => n > 5);     // true
bool semua = await angka.every((n) => n > 0);  // true
int jumlahEl = await angka.length;             // 10

StreamController — Creating and Controlling Streams #

StreamController lets you create your own streams and add values to them manually:

import 'dart:async';

// A basic StreamController
final controller = StreamController<int>();

// The stream that can be listened to
final stream = controller.stream;

// Add values
controller.sink.add(1);
controller.sink.add(2);
controller.sink.add(3);

// Add an error
controller.sink.addError(Exception('An error occurred'));

// Close the stream (triggers onDone)
await controller.sink.close();

// Always check whether there's a listener before adding
if (!controller.isClosed) {
  controller.add(nilai);
}

Broadcast StreamControllers #

// For streams that need many listeners
final broadcastController = StreamController<String>.broadcast();

// Can subscribe repeatedly
broadcastController.stream.listen((s) => print('A: $s'));
broadcastController.stream.listen((s) => print('B: $s'));

broadcastController.add('pesan 1'); // received by A and B
broadcastController.add('pesan 2'); // received by A and B

// onListen and onCancel — callbacks when listeners join/leave
final terkontrol = StreamController<int>.broadcast(
  onListen: () => print('New listener!'),
  onCancel: () => print('The last listener left'),
);

The Repository Pattern with Streams #

// A common pattern: StreamController as simple state management
class ProdukRepository {
  final _controller = StreamController<List<Produk>>.broadcast();
  final List<Produk> _cache = [];

  Stream<List<Produk>> get produkStream => _controller.stream;

  Future<void> muat() async {
    final data = await _api.ambilProduk();
    _cache
      ..clear()
      ..addAll(data);
    _controller.add(List.unmodifiable(_cache));
  }

  Future<void> tambah(Produk produk) async {
    await _api.buatProduk(produk);
    _cache.add(produk);
    _controller.add(List.unmodifiable(_cache));
  }

  void dispose() {
    _controller.close(); // required! prevents memory leaks
  }
}

// Consume the stream in the UI
final repo = ProdukRepository();
repo.produkStream.listen((produk) {
  print('Update: ${produk.length} products');
});

await repo.muat();

StreamTransformer — Custom Transformations #

For transformations that can’t be expressed with Stream’s built-in methods:

import 'dart:async';

// Create a custom transformer
StreamTransformer<int, String> intKeString = StreamTransformer.fromHandlers(
  handleData: (data, sink) {
    sink.add('Nilai: $data');
  },
  handleError: (error, stackTrace, sink) {
    sink.addError('Error: $error', stackTrace);
  },
  handleDone: (sink) {
    sink.add('Selesai');
    sink.close();
  },
);

// Use the transformer
final transformed = Stream.fromIterable([1, 2, 3]).transform(intKeString);
await for (final s in transformed) {
  print(s); // 'Nilai: 1', 'Nilai: 2', 'Nilai: 3', 'Selesai'
}

// A debounce transformer — only emits the last value after a pause
StreamTransformer<T, T> debounce<T>(Duration duration) {
  Timer? timer;
  return StreamTransformer.fromHandlers(
    handleData: (data, sink) {
      timer?.cancel();
      timer = Timer(duration, () => sink.add(data));
    },
    handleDone: (sink) {
      timer?.cancel();
      sink.close();
    },
  );
}

// Debounced search — only searches after the user stops typing for 300ms
final searchStream = searchController.stream
    .transform(debounce(Duration(milliseconds: 300)));

Completer — Creating Futures Manually #

Completer lets you create a Future that can be completed from outside:

import 'dart:async';

// A simple example
final completer = Completer<String>();

// Somewhere: complete the future
Future.delayed(Duration(seconds: 2), () {
  completer.complete('Selesai!');
  // or: completer.completeError(Exception('Gagal'));
});

// Elsewhere: wait for the future
final hasil = await completer.future;
print(hasil); // 'Selesai!'

// Check the status
print(completer.isCompleted); // true after complete() is called

A Real Use Case — Callbacks to Futures #

// Convert a callback-based API into a Future-based one
Future<String> callbackKeFuture() {
  final completer = Completer<String>();

  // An old API that uses callbacks
  legacyAPI.fetch(
    onSuccess: (data) => completer.complete(data),
    onError: (error) => completer.completeError(error),
  );

  return completer.future;
}

// Usage
final data = await callbackKeFuture(); // can be used with async/await!

// An async lock using Completer
class AsyncLock {
  Completer<void>? _completer;

  Future<void> ambil() async {
    while (_completer != null) {
      await _completer!.future; // wait until the lock is released
    }
    _completer = Completer<void>();
  }

  void lepas() {
    _completer?.complete();
    _completer = null;
  }
}

Timer — Scheduling Execution #

import 'dart:async';

// A one-shot timer
final timer = Timer(Duration(seconds: 5), () {
  print('Executed after 5 seconds');
});

// Cancel before it executes
timer.cancel();
print(timer.isActive); // false after cancel

// A periodic timer — executed repeatedly
int hitungan = 0;
final periodic = Timer.periodic(Duration(seconds: 1), (timer) {
  hitungan++;
  print('Tick: $hitungan');
  if (hitungan >= 5) timer.cancel(); // stop after 5 times
});

// Timer.run — execute on the next event queue (after microtasks)
Timer.run(() => print('On the next event queue'));

Async Error Handling #

// try/catch for Futures
Future<void> denganError() async {
  try {
    await Future.error(Exception('Gagal'));
  } on Exception catch (e) {
    print('Caught: $e');
  } finally {
    print('Always executed');
  }
}

// Errors in Streams
final stream = Stream<int>.error(Exception('Stream error'));
stream.listen(
  (nilai) => print(nilai),
  onError: (e) => print('Stream error: $e'),
  cancelOnError: true, // stop the stream on error
);

// runZonedGuarded — catch uncaught errors
runZonedGuarded(() {
  // Code that might throw uncaught errors
  Timer(Duration.zero, () => throw Exception('Error in a timer!'));
}, (error, stackTrace) {
  print('Uncaught error: $error');
  // log to Sentry, Crashlytics, etc.
});

dart:async Anti-Patterns #

Not Closing StreamControllers #

// ANTI-PATTERN: StreamController never closed — memory leak!
class Repository {
  final _controller = StreamController<List<Item>>();
  Stream<List<Item>> get stream => _controller.stream;
  // ✗ no dispose/close
}

// CORRECT: always close the controller when it's no longer used
class Repository {
  final _controller = StreamController<List<Item>>();
  Stream<List<Item>> get stream => _controller.stream;

  void dispose() {
    _controller.close(); // ✓ must be called when the repository is no longer used
  }
}

Forgetting to await Futures #

// ANTI-PATTERN: a Future that isn't awaited — errors vanish silently
void simpanData(Data data) {
  repository.simpan(data); // ✗ if this fails, the error disappears without a trace
}

// CORRECT: await or handle the Future explicitly
Future<void> simpanData(Data data) async {
  await repository.simpan(data); // ✓ errors will propagate to the caller
}

// Or if you really want fire-and-forget, catch the error
void simpanData(Data data) {
  repository.simpan(data).catchError((e) {
    log.error('Failed to save: $e'); // ✓ the error doesn't disappear
  });
}

Summary #

  • Future for single async values, Stream for sequences of async values — choose based on how many values the operation produces.
  • Future.wait() to run several Futures in parallel — far faster than sequential awaits. Use eagerError: false if you want to wait for all to finish even when some error.
  • Single-subscription vs broadcast — default streams can only have one listener. Use asBroadcastStream() or StreamController.broadcast() for multiple listeners.
  • await for is the most idiomatic way to read a stream — cleaner than listen() and automatically closes the subscription.
  • StreamController for creating custom streams — always close it with controller.close() in dispose() to prevent memory leaks.
  • StreamTransformer for transformations not available as built-in methods — debounce, throttle, buffer, window, and more.
  • Completer for converting callback-based APIs into Future-based ones — the only idiomatic way to create a Future completed from outside.
  • Timer.periodic for repeated execution — always keep its reference and call cancel() when it’s no longer needed to prevent memory leaks.
  • runZonedGuarded for catching uncaught async errors — important for logging in production apps.
  • Always await or catch Futures — ignoring a Future that might error is one of the hardest-to-trace bugs because the error disappears without a trace.

← Previous: Math   Next: Convert →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact