Manipulasi Stream

Mengolah aliran data

Stream mentah jarang langsung dipakai: kamu ingin memfilter, mengubah, menggabung. Kabar baiknya, Stream punya method transformasi mirip List (map, where), plus yang khas async: asyncMap, debounce (via package), dan generator async*.

Analogi: stream mentah itu seperti air sungai. Method transformasi adalah bendungan dan filter: menyaring sampah (where), mengubah jadi listrik (map), menggabung anak sungai (merge).

Kenapa transformasi stream penting?

Karena data realtime jarang rapi: sensor mengirim noise, chat mengirim pesan kosong, API mengirim duplikat. Membersihkan di level stream menjaga UI tetap sederhana: UI hanya menerima data yang sudah layak tampil.

map, where, dan kawan-kawan

void main() async {
  var angka = Stream.fromIterable([1, 2, 3, 4, 5, 6]);

  var hasil = angka
      .where((n) => n % 2 == 0) // saring: hanya genap
      .map((n) => n * 10) // ubah: kali 10
      .take(2); // ambil 2 pertama saja

  await for (var n in hasil) {
    print(n); // 20, 40
  }
}

Rantai ini lazy: tidak ada yang diproses sampai ada pendengar, dan take(2) menghentikan stream induk setelah 2 nilai. Efisien untuk stream tak berujung.

Contoh kedua, distinct dan skip:

void main() async {
  var sensor = Stream.fromIterable([1, 1, 2, 1, 3, 3, 2]);

  await for (var n in sensor.distinct()) {
    print(n); // 1, 2, 1, 3, 2 (duplikat BERURUTAN dibuang)
  }

  var data = Stream.fromIterable([10, 20, 30, 40]);
  await for (var n in data.skip(2)) {
    print(n); // 30, 40
  }
}

distinct() hanya membuang duplikat yang berurutan (1,1 → 1), bukan semua duplikat global.

asyncMap vs map: transformasi async

Future<String> ambilNama(int id) async {
  await Future.delayed(Duration(milliseconds: 100));
  return 'User-$id';
}

void main() async {
  var ids = Stream.fromIterable([1, 2, 3]);

  // map: menghasilkan Stream<Future<String>> (salah!)
  // var salah = ids.map((id) => ambilNama(id));

  // asyncMap: menunggu tiap Future, menghasilkan Stream<String>
  await for (var nama in ids.asyncMap((id) => ambilNama(id))) {
    print(nama); // User-1, User-2, User-3
  }
}

Aturan: transformasi yang mengembalikan Future WAJIB pakai asyncMap (atau asyncExpand), bukan map. Salah pakai = stream berisi Future mentah.

async*: generator stream

Fungsi async* membuat stream dengan yield, mirip generator:

// Menghasilkan angka 1..3 dengan jeda
Stream<int> hitung() async* {
  for (int i = 1; i <= 3; i++) {
    await Future.delayed(Duration(milliseconds: 200));
    yield i; // "keluarkan" nilai, lanjutkan nanti
  }
}

Future<void> main() async {
  await for (var n in hitung()) {
    print('Dapat: $n');
  }
  print('Selesai');
}

yield seperti return yang bisa dipakai berkali-kali: tiap yield mengirim satu nilai lalu menjeda fungsi sampai nilai berikutnya diminta. Ada juga yield* untuk mendelegasikan ke stream lain.

Kesalahan umum

1. map dengan fungsi async

// SALAH: Stream<Future<String>>
// var s = Stream.fromIterable([1]).map((id) => ambilNama(id));

// BENAR
void main() async {
  var s = Stream.fromIterable([1]).asyncMap((id) => ambilNama(id));
  await for (var n in s) {
    print(n);
  }
}

2. Lupa stream itu lazy

Transformasi tidak jalan tanpa pendengar. stream.map(...) tanpa listen/await for = tidak terjadi apa-apa.

3. distinct() dianggap menghapus semua duplikat

void main() async {
  var s = Stream.fromIterable([1, 2, 1]);
  await for (var n in s.distinct()) {
    print(n); // 1, 2, 1 (duplikat tak berurutan TETAP ada!)
  }
}

Untuk deduplikasi global, tampung di Set manual atau pakai package rxdart.

Kesimpulan

Transformasi stream: where/map/take/skip/distinct untuk olah sinkron, asyncMap untuk olah async, async*+yield untuk membuat stream sendiri. Rantai yang lazy dan composable membuat pipeline data realtime bersih.

Catatan teknis: Untuk operator canggih (debounce, throttle, combineLatest, switchMap), pakai package rxdart. Itu standar industri Flutter untuk stream kompleks.