Async Generator: async* & yield

Pabrik stream buatan sendiri

Stream.periodic dan Stream.fromIterable hanya mencakup kasus umum. Untuk logika kustom, mis. "baca file baris per baris" atau "polling API sampai kondisi terpenuhi", kamu butuh async generator: fungsi async* yang menghasilkan nilai dengan yield.

Analogi: generator itu seperti keran air. Kamu buka (minta nilai), air mengalir satu gelas (yield satu nilai), keran menunggu sampai kamu minta lagi. Tidak seperti ember (list) yang harus diisi penuh dulu.

Kenapa async* ada?

Karena StreamController manual itu verbose: buat controller, kelola sink, ingat close, tangani error manual. async* merangkum semua itu dalam fungsi biasa yang terbaca sekuensial.

Bentuk dasar

// async* mengembalikan Stream, yield "mengeluarkan" nilai
Stream<int> hitungMundur(int dari) async* {
  for (int i = dari; i > 0; i--) {
    await Future.delayed(Duration(milliseconds: 300));
    yield i; // kirim nilai, jeda di sini sampai diminta lagi
  }
  // selesai: stream otomatis close
}

Future<void> main() async {
  await for (var n in hitungMundur(3)) {
    print('Hitung: $n');
  }
  print('Mulai!');
}

Output: "Hitung: 3", "Hitung: 2", "Hitung: 1" (dengan jeda), "Mulai!". Fungsi menjeda di tiap yield dan lanjut saat konsumen meminta nilai berikutnya. Stream otomatis selesai (close) saat fungsi return.

Contoh kedua, polling API sampai sukses:

Stream<String> pantauStatus() async* {
  for (int i = 1; i <= 5; i++) {
    await Future.delayed(Duration(milliseconds: 200));
    var status = i < 4 ? 'memproses...' : 'selesai!';
    yield 'Cek $i: $status';
    if (i == 4) break;
  }
}

Future<void> main() async {
  await for (var s in pantauStatus()) {
    print(s);
    if (s.contains('selesai')) break; // berhenti lebih awal, generator dibatalkan
  }
}

Saat konsumen break, generator otomatis dibatalkan (tidak ada yield sia-sia). Efisien untuk stream tak berujung.

yield*: delegasi ke stream lain

Stream<int> ganjil() async* {
  yield 1;
  yield 3;
}

Stream<int> genap() async* {
  yield 2;
  yield 4;
}

Stream<int> semua() async* {
  yield* ganjil(); // delegasikan semua nilai ganjil()
  yield* genap();   // lalu semua nilai genap()
  yield 5;
}

Future<void> main() async {
  await for (var n in semua()) {
    print(n); // 1, 3, 2, 4, 5
  }
}

yield* (yield bintang) meneruskan seluruh isi stream lain. Berguna menggabung generator kecil menjadi pipeline besar.

Error di dalam generator

Stream<int> berisiko() async* {
  yield 1;
  yield 2;
  throw Exception('rusak di tengah!');
  // yield 3; // tidak tercapai
}

Future<void> main() async {
  try {
    await for (var n in berisiko()) {
      print(n);
    }
  } catch (e) {
    print('Stream error: $e'); // 1, 2, lalu error
  }
}

Error yang di-throw menjadi error event di stream dan bisa ditangkap try-catch di konsumen. Nilai yang sudah di-yield tetap diterima.

sync*: saudara sinkron

Ada juga sync* untuk Iterable malas (bukan Stream):

// sync* menghasilkan Iterable, tanpa async
Iterable<int> genapSampai(int maks) sync* {
  for (int i = 2; i <= maks; i += 2) {
    yield i;
  }
}

void main() {
  print(genapSampai(10).toList()); // [2, 4, 6, 8, 10]
  // Tidak dihitung sampai diiterasi (lazy)!
}

sync* = generator sinkron untuk Iterable. async* = generator async untuk Stream.

Kesalahan umum

1. return nilai di async*

// SALAH
// Stream<int> s() async* {
//   return 1; // ERROR: async* tidak return nilai
// }

// BENAR
Stream<int> s() async* {
  yield 1;
  yield 2;
}

2. Lupa await di dalam generator

Tanpa await, delay tidak terjadi dan nilai keluar seketika. Itu valid, tapi bukan perilaku "seiring waktu" yang biasanya diinginkan.

3. yield dengan stream yang tidak selesai*

yield* menunggu stream delegasi selesai. Jika stream tak berujung, kode setelahnya tidak pernah jalan.

Kesimpulan

async* + yield = cara deklaratif membuat Stream kustom. yield* mendelegasikan ke stream lain. Error menjadi error event. Untuk versi sinkron malas, ada sync*.

Catatan teknis: Generator bersifat lazy dan single-subscription secara default. Tiap await for baru menjalankan ulang fungsi generator dari awal (kecuali di-broadcast).