KAL
Menu

Stream

Stream sebagai rangkaian event yang dikonsumsi dengan await for.

Daftar isi

Apabila Future mewakili satu nilai yang datang kemudian, Stream mewakili rangkaian nilai yang datang satu per satu dari waktu ke waktu, misalnya pembacaan sensor, baris log, atau pesan yang masuk ke sebuah antrean. Topik ini membahas cara mengonsumsi stream dengan await for, cara membuat stream sendiri dengan StreamController, serta perbedaan langganan tunggal (single subscription) dan siaran (broadcast) yang menentukan jumlah pendengar yang diizinkan. Semua contoh menggunakan dart:async bawaan dan dapat dijalankan sepenuhnya di terminal, sehingga perilaku yang dijelaskan dapat diulang persis.

Mengonsumsi stream dengan await for#

Pernyataan await for mengulang sebuah stream dengan bentuk yang mirip for biasa, yaitu tubuh perulangan dijalankan satu kali untuk setiap event yang tiba. Perulangan ini hanya boleh ditulis di dalam fungsi async, dan eksekusi fungsi menunggu di setiap event sampai event berikutnya tersedia. Perulangan berakhir secara normal ketika stream mengirim sinyal selesai (done), yaitu saat tidak ada event lagi yang akan dikirim. Sumber stream paling sederhana untuk latihan adalah Stream.fromIterable, yang menghasilkan satu event untuk setiap anggota koleksinya.

dart pembacaan.dart 
Stream<double> bacaSensor() {
  return Stream.fromIterable([21.5, 21.8, 22.4, 21.9]);
}

Future<void> main() async {
  print("mulai membaca sensor");
  await for (final suhu in bacaSensor()) {
    print("suhu tercatat: $suhu");
  }
  print("semua pembacaan selesai");
}
bash 
# menjalankan program dari folder topik
dart run pembacaan.dart
text Keluaran 
mulai membaca sensor
suhu tercatat: 21.5
suhu tercatat: 21.8
suhu tercatat: 22.4
suhu tercatat: 21.9
semua pembacaan selesai

Setiap anggota list menjadi satu event, dan tubuh perulangan mencetaknya satu per satu sesuai urutan sumbernya. Baris terakhir baru dicetak setelah stream mengirim sinyal selesai, sehingga kode sesudah await for dapat menganggap seluruh rangkaian sudah habis. Stream.fromIterable memang menjalankan sumbernya untuk setiap pendengar, sehingga apabila stream yang sama didengar dua kali, keduanya memperoleh seluruh elemen dari awal. Perilaku untuk pendengar ganda inilah yang dibahas lebih rinci pada bagian langganan tunggal dan siaran.

Membuat stream dengan StreamController#

StreamController adalah alat membuat stream yang eventnya diisi manual oleh penulis kode. Metode add mengirim satu event ke stream, sedangkan metode close mengirim sinyal selesai dan mengembalikan Future yang lengkap setelah sinyal itu diproses. Metode addError mengirim error sebagai event di sela data, sehingga kegagalan per kejadian dapat dilaporkan tanpa menghentikan stream. Pemanggilan add sebelum ada pendengar tidak menghilangkan event, karena controller langganan tunggal menyimpan event pada antrean dan menyerahkannya saat ada yang mendengarkan. Pola ini mencerminkan kegunaan nyata stream, yaitu pihak pengirim dan pihak penerima tidak berjalan pada saat yang bersamaan.

dart kontroler.dart 
import 'dart:async';

Future<void> main() async {
  final kontroler = StreamController<String>();
  kontroler.stream.listen((pesan) {
    print("diterima: $pesan");
  });
  kontroler.add("antrean A1");
  kontroler.add("antrean A2");
  await kontroler.close();
  print("kontroler ditutup");
}
bash 
# menjalankan program dari folder topik
dart run kontroler.dart
text Keluaran 
diterima: antrean A1
diterima: antrean A2
kontroler ditutup

Contoh antrean loket di atas memperlihatkan pengiriman dan penerimaan yang urutannya rapat. Pesan kontroler ditutup hanya dicetak setelah await kontroler.close() selesai, yang berarti seluruh event sudah diterima pendengar beserta sinyal selesainya. Antrean internal controller juga menjelaskan mengapa event tidak hilang ketika pengirimannya lebih cepat daripada penerimaannya. Pada program yang lebih besar, isi add biasanya berada di bagian lain, misalnya pemroses tugas yang mengumumkan hasilnya, sedangkan pendengarnya berada di bagian antarmuka atau pencatat log.

Langganan tunggal dan aturan mendengar sekali#

Stream dari StreamController tanpa argumen tambahan bersifat langganan tunggal, artinya stream itu hanya boleh didengarkan satu pendengar sepanjang umurnya. Ketentuan ini menjaga kepastian pengiriman, karena setiap event hanya dimiliki satu penerima dan tidak ada yang tertinggal di antara dua pihak. Apabila sebuah stream langganan tunggal didengarkan untuk kedua kalinya, pemanggilan listen memunculkan StateError dengan pesan yang menyatakan stream sudah didengarkan. Kesalahan semacam ini lazim terjadi ketika stream disimpan di variabel dan diteruskan ke lebih dari satu konsumen tanpa disadari.

dart ganda-listen.dart 
import 'dart:async';

Future<void> main() async {
  final kontroler = StreamController<int>();
  kontroler.add(1);
  kontroler.add(2);
  final stream = kontroler.stream;
  final langganan = stream.listen((nilai) {
    print("pendengar pertama: $nilai");
  });
  await kontroler.close();
  await langganan.cancel();
  print("langganan pertama selesai");
  try {
    stream.listen((nilai) {
      print("pendengar kedua: $nilai");
    });
  } on StateError catch (e) {
    print("gagal mendengar ulang: ${e.message}");
  }
}
bash 
# menjalankan program dari folder topik
dart run ganda-listen.dart
text Keluaran 
pendengar pertama: 1
pendengar pertama: 2
langganan pertama selesai
gagal mendengar ulang: Stream has already been listened to.

Pendengar pertama menerima kedua event beserta sinyal selesainya, lalu langganannya dibatalkan dengan cancel agar sumber daya dirilis. Meskipun pendengar pertama sudah tuntas sejak lama, percobaan mendengar ulang tetap ditolak dengan StateError, sebagaimana terlihat pada baris terakhir keluaran. Penolakan itu berlaku seumur stream, sehingga solusinya bukan mendengar ulang, melainkan menggunakan stream siaran yang memang dirancang untuk banyak pendengar. Sifat satu pendengar ini cocok untuk aliran yang berdiri sendiri, misalnya pembaca berkas atau proses yang mengerjakan satu tugas dari awal sampai akhir.

Broadcast untuk banyak pendengar#

StreamController.broadcast() membuat stream siaran yang setiap eventnya dikirim kepada semua pendengar yang aktif pada saat event ditambahkan. Pendengar dapat didaftarkan kapan saja, masing-masing memperoleh rangkaian event yang dimulai sejak pendaftarannya, dan langganannya dapat dibatalkan dengan cancel tanpa memengaruhi pendengar lain. Bentuk ini tepat untuk kejadian yang dipantau banyak pihak sekaligus, misalnya catatan suhu yang dibaca oleh pencatat log dan papan tampilan secara bersamaan. Konsekuensinya, event yang dikirim sebelum sebuah pihak mendaftar tidak sampai kepada pihak tersebut.

dart broadcast.dart 
import 'dart:async';

Future<void> main() async {
  final kontroler = StreamController<int>.broadcast();
  final langganan1 = kontroler.stream.listen((nilai) {
    print("pendengar A: $nilai");
  });
  final langganan2 = kontroler.stream.listen((nilai) {
    print("pendengar B: $nilai");
  });
  kontroler.add(10);
  kontroler.add(24);
  await kontroler.close();
  await langganan1.cancel();
  await langganan2.cancel();
  print("siaran selesai");
}
bash 
# menjalankan program dari folder topik
dart run broadcast.dart
text Keluaran 
pendengar A: 10
pendengar B: 10
pendengar A: 24
pendengar B: 24
siaran selesai

Keluaran menunjukkan setiap event diputar kepada kedua pendengar, dengan urutan penyerahan per pendengar yang tetap sesuai urutan pengirimannya. Kedua pendengar mendaftar sebelum event pertama ditambahkan, sehingga tidak ada event yang terlewat. Apabila salah satu langganan dibatalkan di tengah siaran, pendengar sisanya terus menerima event tanpa gangguan, dan inilah perbedaan perilaku yang menonjol dibandingkan stream langganan tunggal.

Ringkasan#

  • Stream mewakili rangkaian event yang dikonsumsi dengan await for di dalam fungsi async, dan perulangannya berakhir pada sinyal selesai.
  • StreamController mengirim event dengan add, mengakhiri stream dengan close, dan menyimpan event pada antrean sebelum ada pendengar.
  • Stream langganan tunggal hanya boleh didengarkan sekali, dan mendengar ulang memunculkan StateError berupa Stream has already been listened to.
  • Stream siaran dari StreamController.broadcast() mengirim setiap event kepada semua pendengar yang terdaftar, dan langganan dapat dibatalkan satu per satu.

Referensi#