KAL
Menu

Transformasi stream

Transformasi asinkron stream melalui pemetaan, penyaringan, dan penggabungan.

Daftar isi

Stream mentah dari sebuah sumber jarang berbentuk yang persis dibutuhkan konsumennya, sehingga event perlu disaring, diubah, dipotong, atau dipadukan sebelum digunakan. dart:async menyediakan metode transformasi yang mengembalikan stream baru tanpa mengubah sumbernya, dan semuanya bekerja per event sebagaimana event itu mengalir. Topik ini membahas map dan where untuk pemetaan serta penyaringan, asyncMap untuk transformasi yang menunggu pekerjaan asinkron, take dan skip untuk memotong rangkaian, fold dan reduce untuk merangkum, serta penggabungan dua stream dengan controller siaran dan addStream. Semua contoh menggunakan dart:async bawaan dan dapat dijalankan di terminal.

Pemetaan dan penyaringan#

Metode map mengubah setiap event dengan fungsi tertentu, sedangkan where menyaring event dengan sebuah syarat, dan keduanya mengembalikan stream baru yang menampung hasil transformasinya. Rantai metode ini ditulis berurutan seperti pada Iterable, tetapi eksekusinya per event, yaitu setiap event yang tiba mengalir melewati seluruh rantai sebelum event berikutnya ditarik dari sumber. Urutan rantai juga menentukan kerja yang dilakukan, karena menyaring lebih dahulu mengurangi jumlah pemetaan yang berjalan. Pemanggilan metode transformasi tidak menarik satu event pun di awal, sehingga penggunaan memori tetap rendah untuk stream yang panjang. Contoh berikut menyaring pembacaan sensor suhu, lalu memetakan angka yang lolos menjadi pesan siap cetak.

dart saring-tampung.dart 
Stream<double> bacaSensor() {
  return Stream.fromIterable([21.5, 22.8, 19.4, 24.1, 23.6]);
}

Future<void> main() async {
  final panas = bacaSensor()
      .where((suhu) => suhu > 22.0)
      .map((suhu) => "suhu tinggi: $suhu");
  await for (final pesan in panas) {
    print(pesan);
  }
}
bash 
# menjalankan program dari folder topik
dart run saring-tampung.dart
text Keluaran 
suhu tinggi: 22.8
suhu tinggi: 24.1
suhu tinggi: 23.6

Dari lima pembacaan, hanya tiga di antaranya lolos syarat suhu di atas 22.0, dan ketiganya tampil dalam bentuk pesan hasil pemetaan. Variabel panas hanyalah deskripsi alur, yaitu sumber yang disaring lalu dipetakan, sehingga tidak ada event yang diproses sebelum await for memintanya. Penyusunan alur lebih dahulu dan konsumsi kemudian memisahkan dua tanggung jawab yang berbeda, yaitu mendefinisikan transformasi dan menjalankannya.

Transformasi asinkron dengan asyncMap#

Apabila fungsi pengubah setiap event itu sendiri bersifat asinkron dan mengembalikan Future, metode map tidak lagi cukup karena metode tersebut menuntut fungsi sinkron. Metode asyncMap menggantikan perannya, yaitu memanggil fungsi asinkron untuk setiap event dan menunggu Future-nya selesai sebelum meneruskan hasilnya ke stream baru. Hasilnya tetap sebuah stream bertipe keluaran fungsi pengubahnya, sehingga rantai transformasi lain dapat dilanjutkan setelahnya. Pemrosesan tetap berurutan, sehingga hasil event pertama selalu tiba sebelum event kedua mulai dilayani. Pola ini cocok untuk setiap event yang perlu diolah lebih dahulu, misalnya nomor antrean yang dilayani satu per satu di sebuah loket.

dart async-map.dart 
Stream<int> nomorAntrean() {
  return Stream.fromIterable([1, 2, 3]);
}

Future<String> layani(int nomor) async {
  await Future.delayed(const Duration(milliseconds: 20));
  return "nomor $nomor dilayani";
}

Future<void> main() async {
  await for (final pesan in nomorAntrean().asyncMap(layani)) {
    print(pesan);
  }
  print("loket tutup");
}
bash 
# menjalankan program dari folder topik
dart run async-map.dart
text Keluaran 
nomor 1 dilayani
nomor 2 dilayani
nomor 3 dilayani
loket tutup

Fungsi layani menyimulasikan pekerjaan 20 milidetik per nomor, dan asyncMap(layani) menunggu setiap pekerjaan sebelum meneruskan pesannya. Keluaran tetap berurutan sesuai nomor antrean meskipun pekerjaannya lambat, karena asyncMap tidak melayani event berikutnya sebelum event sebelumnya selesai. Baris loket tutup baru muncul setelah seluruh antrean habis, sebagaimana baris sesudah await for menandai akhir rangkaian pada contoh sebelumnya.

Mengambil sebagian dengan take dan skip#

Metode take mengambil event pertama sebanyak yang diminta, sedangkan skip melewati event awal sebanyak itu dan menyerahkan sisanya. Keduanya berguna ketika rangkaian hanya perlu dipotong di ujungnya, misalnya pratinjau beberapa baris pertama dari log atau pengabaian baris judul pada berkas data. Padanan yang menguji syarat setiap event juga tersedia dalam bentuk takeWhile dan skipWhile, sehingga pemotongan tidak harus berbasis jumlah. Metode-metode ini dapat dirangkai dengan map dan where tanpa batasan, karena semua metode transformasi mengembalikan stream baru dengan antarmuka yang sama.

dart potong-lompat.dart 
Stream<String> kataMasuk() {
  return Stream.fromIterable(["ali", "budi", "citra", "dewi", "edi"]);
}

Future<void> main() async {
  print("tiga kata pertama:");
  await for (final kata in kataMasuk().take(3)) {
    print(kata);
  }
  print("dua kata terakhir:");
  await for (final kata in kataMasuk().skip(3)) {
    print(kata);
  }
}
bash 
# menjalankan program dari folder topik
dart run potong-lompat.dart
text Keluaran 
tiga kata pertama:
ali
budi
citra
dua kata terakhir:
dewi
edi

take(3) menghentikan konsumsi setelah tiga event, sehingga sisa rangkaian tidak pernah ditambahkan ke stream barunya, sedangkan skip(3) membuang tiga event pertama dan meneruskan sisanya. Angka pada keduanya berupa jumlah event, bukan satuan waktu, sehingga jeda antarevent tidak memengaruhi hasil pemotongannya. Kedua pemotongan dilakukan pada stream terpisah yang masing-masing dijalankan dari awal, karena kataMasuk() dipanggil dua kali dan keduanya memperoleh rangkaian yang sama. Pada sumber yang benar-benar mengalir seperti pembacaan sensor langsung, pemotongan semacam ini tetap berlaku pada event yang diterima sejak langganan dimulai.

Merangkum dengan fold dan reduce#

Metode fold menggabungkan seluruh event menjadi satu nilai dengan akumulator yang nilai awalnya ditentukan, sedangkan reduce melakukan hal serupa tanpa nilai awal dan menggunakan event pertama sebagai titik mulai. Keduanya mengembalikan Future yang lengkap setelah sinyal selesai diterima, sehingga hasilnya diambil dengan await. reduce menuntut stream yang berisi minimal satu event dan tipe hasilnya sama dengan tipe eventnya, sedangkan fold bebas mengubah tipe karena akumulatornya ditentukan sendiri. Contoh berikut merangkum lima pembacaan suhu menjadi totalnya dan nilai tertingginya.

dart ringkasan-angka.dart 
Stream<double> bacaSensor() {
  return Stream.fromIterable([21.5, 22.8, 19.4, 24.1, 23.6]);
}

Future<void> main() async {
  final total = await bacaSensor()
      .fold<double>(0.0, (jumlah, suhu) => jumlah + suhu);
  final tertinggi = await bacaSensor().reduce((a, b) => a > b ? a : b);
  print("total: $total");
  print("tertinggi: $tertinggi");
}
bash 
# menjalankan program dari folder topik
dart run ringkasan-angka.dart
text Keluaran 
total: 111.4
tertinggi: 24.1

fold menjumlahkan kelima event dengan akumulator yang dimulai dari 0.0, dan hasilnya 111.4. reduce membandingkan dua event berurutan dan menyimpan yang lebih tinggi, sehingga event terakhir yang tersisa adalah 24.1. Stream sumber diputar dua kali karena dua perhitungan berbeda membutuhkan alur yang sama, dan pemutaran ulang tersebut mungkin berkat sumber fromIterable yang menjalankan ulang koleksinya untuk setiap konsumen.

Menggabungkan dua stream#

Penggabungan sederhana dapat dibangun tanpa package tambahan dengan controller siaran dan metode addStream. Metode addStream menyalin seluruh event dari sebuah stream sumber ke stream tujuan, dan pemanggilannya mengembalikan Future yang selesai setelah sumber habis. Menunggu pemanggilan pertama sebelum memanggil yang kedua membuat urutan gabungan menjadi pasti, yaitu seluruh event papan pertama lebih dahulu, disusul seluruh event papan kedua. Pola ini cukup untuk antrean yang digabung berurutan, sedangkan pengurutan yang lebih kompleks membutuhkan penyatuan berbasis waktu di luar cakupan topik ini.

dart gabung.dart 
import 'dart:async';

Future<void> main() async {
  final gabungan = StreamController<String>.broadcast();
  final hasil = <String>[];
  final langganan = gabungan.stream.listen(hasil.add);
  final papanA = Stream.fromIterable(["A1", "A2"]);
  final papanB = Stream.fromIterable(["B1", "B2"]);
  await gabungan.addStream(papanA);
  await gabungan.addStream(papanB);
  await gabungan.close();
  await langganan.cancel();
  print(hasil.join(" "));
}
bash 
# menjalankan program dari folder topik
dart run gabung.dart
text Keluaran 
A1 A2 B1 B2

Pendengar pertama didaftarkan sebelum penggabungan dimulai, sehingga tidak ada event yang terlewat pada saat addStream bekerja. Keluaran A1 A2 B1 B2 memperlihatkan urutan yang pasti, yaitu seluruh isi papan A dahulu karena addStream pertama ditunggu sampai selesai. Pemanggilan close mengakhiri stream gabungan setelah kedua papan habis, dan langganan dibatalkan untuk merilis sumber dayanya sebelum hasil dicetak.

Ringkasan#

  • map dan where mengubah serta menyaring event per kejadian, dan keduanya mengembalikan stream baru tanpa menyentuh sumbernya.
  • asyncMap menunggu Future dari fungsi pengubah setiap event, sehingga hasil transformasi asinkronnya tetap berurutan.
  • take dan skip memotong rangkaian di ujungnya, sedangkan fold dan reduce merangkum seluruh event menjadi satu nilai melalui await.
  • Penggabungan berurutan dapat dibangun dengan controller siaran dan addStream yang ditunggu satu per satu, sehingga urutan eventnya pasti.

Referensi#