Async, Future & Stream

1. Concurrency model Dart

Dart dùng single-threaded event loop — giống JavaScript. Mỗi Dart program chạy trên một isolate (= 1 thread riêng + 1 heap riêng). Không share memory giữa isolate.

• Microtask queue — task ưu tiên cao. Future.value(), .then() resolved schedule vào đây.
• Event queue — task ưu tiên thấp hơn. Timer callback, I/O completion, gesture event.
• Mỗi tick: drain hết microtask queue, lấy 1 event từ event queue chạy, lặp.
• Code đồng bộ chạy trên main isolate — không yield event loop trong khi chạy.

2. Future<T> — value xuất hiện trong tương lai

// Tạo Future
final a = Future.value(5);                  // Future<int> resolved 5
final b = Future.delayed(Duration(seconds: 1), () => 'hi');
final c = Future.error(Exception('boom'));

.then() — callback chain style

fetchUser()
  .then((user) {
    print(user.name);
    return fetchPosts(user.id);
  })
  .then((posts) => print('${posts.length} posts'))
  .catchError((e) => print('Error: $e'))
  .whenComplete(() => print('done'));

async/await — sugar đẹp hơn

Future<void> demo() async {
  try {
    final user = await fetchUser();
    print(user.name);
    final posts = await fetchPosts(user.id);
    print('${posts.length} posts');
  } catch (e) {
    print('Error: $e');
  } finally {
    print('done');
  }
}

async function — không chạy đồng bộ

Future<int> compute() async {
  print('B');
  return 42;
}

void main() {
  print('A');
  compute().then(print);
  print('C');
}
// Output: A, B, C, 42 (không phải A, B, 42, C)

Lý do: compute chạy đến end body và return Future. Future resolve schedule microtask. Microtask chạy sau khi print('C') xong.

3. Error handling Async

Future<String> fetch() async {
  throw Exception('oops');
}

Future<void> main() async {
  // Cách 1: try/catch quanh await (idiomatic)
  try {
    final r = await fetch();
  } catch (e) {
    print('caught: $e');
  }

  // Cách 2: .catchError (callback)
  fetch().catchError((e) => print('caught2: $e'));
}

ElevatedButton(
  onPressed: () async {
    await riskyOperation();  // nếu throw → unhandled exception
  },
  ...
)

// Pattern an toàn
ElevatedButton(
  onPressed: () async {
    try {
      await riskyOperation();
    } catch (e) {
      _showError(e);
    }
  },
  ...
)

Future.wait — parallel

Future<String> slowApi(String id) async {
  await Future.delayed(Duration(seconds: 1));
  return 'data-$id';
}

Future<void> main() async {
  final sw = Stopwatch()..start();

  // Tuần tự — ~3 giây
  final a = await slowApi('1');
  final b = await slowApi('2');
  final c = await slowApi('3');
  print('Sequential: ${sw.elapsedMilliseconds}ms');

  // Parallel — ~1 giây (3 task chạy cùng lúc)
  sw.reset();
  final [a2, b2, c2] = await Future.wait([
    slowApi('1'),
    slowApi('2'),
    slowApi('3'),
  ]);
  print('Parallel: ${sw.elapsedMilliseconds}ms');
}

Future.any — timeout pattern

Future<String> withTimeout(Future<String> task, Duration timeout) {
  return Future.any([
    task,
    Future.delayed(timeout, () => throw TimeoutException('timed out')),
  ]);
}

// Hoặc dùng built-in
Future.delayed(Duration(seconds: 5)).timeout(Duration(seconds: 2));
// Future throws TimeoutException sau 2s

4. Stream<T> — Iterable async

Stream = "Iterable mỗi value đến tại thời điểm khác nhau". So sánh:

Single-subscription vs Broadcast

// Single-subscription — chỉ 1 listener
final single = Stream.fromIterable([1, 2, 3]);
single.listen((v) => print(v));
// single.listen((v) => print(v));  // ❌ runtime error

// Broadcast — nhiều listener
final broadcast = single.asBroadcastStream();
broadcast.listen((v) => print('A: $v'));
broadcast.listen((v) => print('B: $v'));

5. Tạo Stream

StreamController — manual push

final controller = StreamController<int>();

// Subscribe
controller.stream.listen(
  (v) => print('value: $v'),
  onError: (e) => print('error: $e'),
  onDone: () => print('done'),
);

// Push events
controller.add(1);
controller.add(2);
controller.addError(Exception('oops'));
controller.add(3);
controller.close();   // trigger onDone

async*/yield — generator

Stream<int> countdown(int from) async* {
  for (var i = from; i >= 0; i--) {
    await Future.delayed(Duration(seconds: 1));
    yield i;
  }
}

// Consume
await for (final v in countdown(3)) {
  print(v);
}
// 3 (sau 1s), 2, 1, 0

Built-in factories

Stream.fromIterable([1, 2, 3]);          // emit 3 value rồi done
Stream.value(42);                       // emit 1 value rồi done
Stream.error(Exception('oops'));         // emit 1 error rồi done
Stream.periodic(Duration(seconds: 1), (i) => i);  // emit mỗi giây, vô hạn

6. Tiêu thụ Stream

.listen() — callback style

final subscription = stream.listen(
  (v) => print('data: $v'),
  onError: (e) => print('error: $e'),
  onDone: () => print('done'),
  cancelOnError: false,
);

// Sau khi không cần, cancel để tránh leak
subscription.cancel();

await for — đợi tuần tự

Future<void> main() async {
  await for (final v in stream) {
    print(v);
    if (v > 10) break;  // tự cancel subscription
  }
}

7. Transform Stream

Stream<int> numbers() async* {
  for (var i = 1; i <= 10; i++) yield i;
}

numbers()
  .where((x) => x % 2 == 0)            // filter
  .map((x) => x * 10)                  // transform
  .take(3)                              // chỉ lấy 3 đầu
  .listen(print);
// 20, 40, 60

// asyncMap — transform async
numbers().asyncMap((x) async {
  await Future.delayed(Duration(milliseconds: 100));
  return x * 2;
});

// distinct — bỏ duplicate liên tiếp
Stream.fromIterable([1, 1, 2, 2, 3, 3]).distinct().listen(print);
// 1, 2, 3

8. StreamSubscription lifecycle

final sub = stream.listen(print);

sub.pause();              // tạm dừng — value buffer
sub.resume();             // tiếp tục — flush buffer
await sub.cancel();       // dừng vĩnh viễn, free resource

class _MyWidgetState extends State<MyWidget> {
  late StreamSubscription _sub;

  @override
  void initState() {
    super.initState();
    _sub = service.events.listen(_handle);
  }

  @override
  void dispose() {
    _sub.cancel();           // CRITICAL — không cancel = memory leak
    super.dispose();
  }
}

9. Isolate intro

Khi cần CPU-bound work (parse JSON 10MB, image process), main isolate sẽ block UI. Cần isolate riêng.

compute() — Flutter shortcut

// Function phải top-level hoặc static
List<User> parseUsers(String json) {
  return (jsonDecode(json) as List)
      .map((j) => User.fromJson(j))
      .toList();
}

// Chạy ở isolate khác main
final users = await compute(parseUsers, largeJsonString);
// Main isolate KHÔNG block trong khi parse

Isolate.spawn — raw

void heavyWork(SendPort port) {
  final result = ...; // compute heavy
  port.send(result);
}

final receivePort = ReceivePort();
await Isolate.spawn(heavyWork, receivePort.sendPort);
final result = await receivePort.first;

10. Bài tập

Future<String> fetchUser(int id) async {
  await Future.delayed(Duration(seconds: 1));
  return 'user_$id';
}

Future<void> main() async {
  final sw = Stopwatch()..start();
  await fetchUser(1);
  await fetchUser(2);
  await fetchUser(3);
  print('Sequential: ${sw.elapsedMilliseconds}ms');  // ~3000ms

  sw.reset();
  await Future.wait([fetchUser(1), fetchUser(2), fetchUser(3)]);
  print('Parallel: ${sw.elapsedMilliseconds}ms');    // ~1000ms
}

Future<void> main() async {
  final countdown = Stream.periodic(Duration(seconds: 1), (i) => 10 - i)
      .take(11);
  await for (final v in countdown) {
    print(v);
  }
}

Future<T> retry<T>(
  Future<T> Function() task, {
  int times = 3,
  Duration delay = const Duration(seconds: 1),
}) async {
  for (var i = 0; i < times; i++) {
    try {
      return await task();
    } catch (e) {
      if (i == times - 1) rethrow;
      await Future.delayed(delay);
    }
  }
  throw StateError('unreachable');
}

// Test
var attempts = 0;
Future<int> flaky() async {
  attempts++;
  if (attempts < 3) throw Exception('fail');
  return 42;
}

void main() async {
  print(await retry(flaky));  // 42 after 3 attempts
}

class EventBus {
  final _ctrl = StreamController<String>.broadcast();

  Stream<String> get events => _ctrl.stream;

  void emit(String event) {
    _ctrl.add(event);
  }

  void close() {
    _ctrl.close();
  }
}

void main() async {
  final bus = EventBus();
  bus.events.listen((e) => print('A: $e'));
  bus.events.listen((e) => print('B: $e'));

  bus.emit('login');
  bus.emit('purchase');
  // A: login, B: login, A: purchase, B: purchase
}

fetchUser(1)
  .then((user) => fetchPosts(user.id))
  .then((posts) => processPosts(posts))
  .catchError((e) => print('Error: $e'))
  .whenComplete(() => print('done'));

Future<void> main() async {
  try {
    final user = await fetchUser(1);
    final posts = await fetchPosts(user.id);
    final result = await processPosts(posts);
  } catch (e) {
    print('Error: $e');
  } finally {
    print('done');
  }
}

11. Quiz

12. Tổng kết

• ✅ Single-threaded event loop + isolate model.
• ✅ Future<T> ≈ JS Promise. async/await sugar.
• ✅ Error: try/catch quanh await idiomatic.
• ✅ Future.wait parallel, Future.any timeout.
• ✅ Stream<T> — Iterable async. Single vs broadcast.
• ✅ Tạo: StreamController, async*/yield, factory.
• ✅ Tiêu thụ: .listen, await for.
• ✅ Transform: .map, .where, .asyncMap, .distinct.
• ✅ StreamSubscription lifecycle — pause/resume/cancel.
• ✅ Isolate intro: compute + Isolate.spawn.

13. Kết nối

• Chương 8: Generic Future<T> / Stream<T>.
• Flutter chương 2 — Widget: FutureBuilder, StreamBuilder.
• Flutter chương 7 — Bloc: dựa trên Stream.
• Flutter chương 10 — Network: HTTP call async.