Stream ဆိုတာဘာလဲ?

Original Post: https://www.didierboelens.com/2018/08/reactive-programming-streams-bloc/

Stream ဆိုတာ အစွန်း၂ဖက်ပါတဲ့ ပိုက်လိုပဲ၊ တဖက်ကနေ ထည့်ပေးပြီး တဖက်က ပြန်ထွက်လာတာမျိုးပေါ့။ ပိုက်ကို ထိန်းချုပ်ပေးတာကကျတော့ StreamController လို့ခေါ်တယ်။ ဝင်ပေါက်ကို sink လို့သတ်မှတ်ပြီး ထွက်ပေါက်ကိုတော့ stream (S အသေး) နဲ့သတ်မှတ်တယ်။

Stream ထဲကနေ ကိုယ်ကြိုက်တာကို လက်ဆင့်ကမ်းပေးလို့ရတယ်။ value, event, object, collection, map, error ဒါမှမဟုတ် နောက်ထပ် Stream တခုကိုပါ လက်ဆင့်ကမ်းလို့ရတယ်။

Stream ထဲထည့်ပေးလိုက်တာကို သိချင်တယ်ဆို StreamController ကို listen လုပ်ပေးထားဖို့တော့လိုတယ်။

listener သတ်မှတ်လိုက်တဲ့အခါ StreamSubscription ဆိုတဲ့ object ကိုရလာတယ်။ ဒီ StreamSubscription object ကနေပဲ Stream ထဲမှာ တခုခုထည့်ပေးလိုက်တဲ့အခါ သိရမှာဖြစ်တယ်။

နားထောင်နေတဲ့ active listener တခုရှိပြီဆိုတာနဲ့ Stream ကနေပြီး event တွေစပြီး လွှင့်ပေးတယ်။ ဘယ်အချိန်တွေမှာ event ထုတ်လွှင့်သလဲဆို
၁ - stream ကနေပြီး data ထွက်ဖို့ရှိတဲ့အခါ
၂ - stream ကို error လှမ်းပို့လိုက်တဲ့အခါ
၃ - stream ပိတ်သွားတဲ့အခါ
အဲ့လိုအချိန်တွေမှာ event တွေကို ထုတ်လွှင့်ပေးတယ်။

StreamSubscription object ကိုသုံးပြီးတော့ Stream ကို နားထောင်တာ လုံးဝရပ်ဖို့၊ ခဏရပ်ဖို့၊ ဆက်နားထောင်ဖို့ ဒါတွေကို သတ်မှတ်ပေးလို့ရတယ်။

Stream ကို ပိုက်လိုမျိုးဆိုပြီး ပြောလို့ တဖက်ကဝင်လာတာ တစ်ဖက်က ထွက်သွားရုံပဲတော့ မဟုတ်ဘူး။ ဝင်လာတာတွေကို ပုံစံပြောင်းတာတွေလဲ လုပ်ပေးလို့ရတယ်။ ဒီလိုလုပ်ဖို့ဆိုရင်တော့ StreamTransformer ဆိုတာလိုအပ်လာပြီ။ StreamTransformer ရဲ့အလုပ်လုပ်ပုံက ရိုးရိုး function လိုမျိုးပဲ ဒါပေမယ့်သူက Stream ထဲဝင်လာတဲ့ data တွေကို ရယူပြီး အလုပ်လုပ်သွားတာဖြစ်တယ်။ သူထဲမှာ လိုအပ်တာတွေ ပြုလုပ်ပြီးရင် Stream အနေနဲ့ ပြန်ထွက်လာတယ်။ ဒါကြောင့် StreamTransformer တွေအများကြီးကို တခုပြီးတခု ဆင့်သုံးသွားလို့လဲ​ ရတယ်။ StreamTransformer တွေကို filtering, regrouping, modification, injection, buffering, processing တွေအပြင် အသုံးပြုလို့ရတာတွေ အများကြီး ရှိတယ်။

Stream အမျိုးအစား ၂ခုရှိတယ်။

Single-subscription Streams - ဒီအမျိုးအစားကတော့ Stream စတာကနေ ဆုံးတဲ့အချိန်ထိ listener တခုပဲ နားထောင်လို့ရတယ်။ ပထမနားထောင်နေတဲ့ listener ကိုရပ်ပြီး ပြန်နားထောင်မယ်ဆိုရင်တောင် မရတော့ဘူး။

Broadcast Streams - ဒီအမျိုးအစားကတော့ listener ကြိုက်သလောက် နားထောင်လို့ရတယ်။ listener စပြီးနားထောင်လိုက်တဲ့အချိန်ကစပြီး Stream ထဲမှာဖြစ်လာသမျှ event တွေကို လက်ခံရရှိမှာဖြစ်တယ်။

// Single-Subscription Stream
import 'dart:async';

void main() {
  final StreamController ctrl = new StreamController();
  final StreamSubscription subscription = ctrl.stream.listen((data) {
    print(data);
  });

  ctrl.sink.add('my name');
  ctrl.sink.add(1234);
  ctrl.sink.add({'a': 'Element A', 'b': 'Element B'});
  ctrl.sink.add(3.142);

  ctrl.close();
}

// Broadcast Stream with StreamTransformer
import 'dart:async';

void main() {
  final StreamController<int> ctrl = new StreamController<int>.broadcast();
  final StreamSubscription subscription = ctrl.stream
      .where((value) => value % 2 == 0)
      .listen((value) => print(value));

  for (var i = 0; i < 11; i++) {
    ctrl.sink.add(i);
  }

  ctrl.close();
}