Giới thiệu Stream
Lập trình bất đồng bộ là một thuật ngữ phổ biến trong lập trình. Với ngôn ngữ lập trình Dart, chúng ta đã quen với Future class cung cấp một tính toán không được hoàn thành ngay lập tức và sẽ thông báo kết quả khi sẵn sàng, việc này chỉ thực hiện đúng 1 lần. Thay vào đó Stream là một luồng các sự kiện bất đồng bộ và cho phép chúng ta lắng nghe các sự kiện này khi chúng được bắn ra từ luồng.
A stream is a sequence of asynchronous events.
Phân loại Stream
-
Có 2 loại Stream
- Single subscription streams
- Broadcast streams
Hiểu cơ bản là Single subscription streams chỉ có thể lắng nghe 1 lần duy nhất. (Nếu có sự kiện lắng nghe từ một nơi nào khác hoặc lần thứ 2 trở lên thì sẽ báo lỗi). Trái lại Broadcast streams có thể lắng nghe bất kỳ đâu.
//Sử dụng controller//**Single subscription streamsStreamController streamController =StreamController();Stream stream = streamController.stream;//Broadcast streams-StreamController broadcastStreamController =StreamController.broadcast();Stream stream = broadcastStreamController .stream;//Sử dụng async*Stream<int>countStream(int to)async*{for(int i =1; i <= to; i++){yield i;}}//yield -> bắn ra một event//yield* -> bắn ra 1 stream**
Các phương thức
- Sức mạnh của Stream không chỉ giúp việc lập trình bất đồng bộ dễ dàng mà nó còn có các bộ method hổ trợ rất mạnh mẽ như filter, transform…
- Xêm thêm tại https://dart.dev/tutorials/language/streams
- Thư viện support: extension cho stream Rxdart
- Tham khảo https://reactivex.io/ để tìm hiểu về Reactivex/Functional reactive programming để tìm hiểu về cơ chế, cách hoạt động, cách sử dụng (support rất nhiều ngôn ngữ)
<aside>
⚠️ Có một câu hỏi nhỏ: Vậy điểm khác biệt giữa 2 loại Stream này là gì?
</aside>
Tại sao nên sử dụng Stream
- Stream là một thư viện/api core của Dart.
- → Dễ dàng tạo ra các Plugin (Tách nhỏ repo)
- → Nắm vững công nghệ
- Stream phù hợp với phong cách *làm mới giao diện (declarative) của Flutter. (***so với cách Imperative ở Native) và mô hình MVVM.
- Giảm phụ thuộc vào các thư viện của các bên thứ 3.
- → Tránh lỗi khi thay đổi version
- → Giảm app size
- Một số thư viện như bloc, getx có cơ chế phụ thuộc vào Stream.
Xây dựng giao diện bằng StreamBuilder
StreamBuilder lắng nghe sự thay đổi của Stream và làm mới lại giao diện.
Để sử dụng StreamBuilder cần gọi
constStreamBuilder({Key? key,Stream<T>? stream,T? initialData,
required AsyncWidgetBuilder<T> builder,})Ý nghĩa Parameters:
T? initialData: giá trị mặc định, nếu không truyền vào thì coi như chưa nhận được dữ liệu từ Stream
Stream<T>? stream: truyền Stream cần lắng nghe và được xử lý ở hàm builder
required AsyncWidgetBuilder<T> builder: Xây dựng giao diện được thiết lập ở đây
builder:(BuildContext context,AsyncSnapshot<int> snapshot,){//return Widget here;}classAsyncSnapshot<T>{/// Creates an [AsyncSnapshot] with the specified [connectionState],/// and optionally either [data] or [error] with an optional [stackTrace]/// (but not both data and error).constAsyncSnapshot._(this.connectionState,this.data,this.error,this.stackTrace):assert(connectionState !=null),assert(!(data !=null&& error !=null)),assert(stackTrace ==null|| error !=null);Khi xây dựng UI cần chú ý tới thành phần snapshot trong builder
- Kiểm tra ConnectionState xem tình trạng kết nối với Stream
- snapshot.hasError và snapshot.error: Kiểm tra có lỗi và lấy lỗi. (Xử dụng
addErrorđể bắn ra sự kiện lỗi) - snapshot.hasData và snapshot.data: Kiểm tra có dữ liệu và lấy dữ liệu. (Xử dụng
addđể bắn ra dữ liệu)
Các trạng thái của ConnectionState
/// The state of connection to an asynchronous computation.////// The usual flow of state is as follows:////// 1. [none], maybe with some initial data./// 2. [waiting], indicating that the asynchronous operation has begun,/// typically with the data being null./// 3. [active], with data being non-null, and possible changing over time./// 4. [done], with data being non-null.////// See also:////// * [AsyncSnapshot], which augments a connection state with information/// received from the asynchronous computation.enumConnectionState{/// Not currently connected to any asynchronous computation.////// For example, a [FutureBuilder] whose [FutureBuilder.future] is null.
none,/// Connected to an asynchronous computation and awaiting interaction.
waiting,/// Connected to an active asynchronous computation.////// For example, a [Stream] that has returned at least one value, but is not/// yet done.
active,/// Connected to a terminated asynchronous computation.
done,}- Khi không truyền Stream vào StreamBuilder (có nghĩa Stream là null) → none
- Khi truyền Stream (initialData có thể bằng null hoặc không) và chưa add sự kiện vào → waiting
- Khi truyền Stream và add sự kiện → active
- Khi truyền Stream và close() → done
Ví dụ về xử lý giao diện:
StreamBuilder<int>(
stream: stream,
builder:(BuildContext context,AsyncSnapshot<int> snapshot,){if(snapshot.connectionState ==ConnectionState.waiting){returnCircularProgressIndicator();}elseif(snapshot.connectionState ==ConnectionState.active
|| snapshot.connectionState ==ConnectionState.done){if(snapshot.hasError){**//<-- Kiểm tra có lỗi**returnconstText('Error');}elseif(snapshot.hasData){**//<-- Kiểm tra có data**returnText(
snapshot.data.toString(),**//<-- Lấy data**
style:constTextStyle(color:Colors.red, fontSize:40));}else{returnconstText('Empty data');}}else{returnText('State: ${snapshot.connectionState}');}},),[Demo] Xây dựng ứng dụng đếm ngược thời gian bằng Stream
<aside>
📌 Demo bên dưới không xử lý các trạng thái lỗi
</aside>
Các tính năng cơ bản:
Bắt đầu – Tạm dừng – Tiếp tục – Làm mới
UI Bắt đầu
UI khi đã Bắt đầu
Các bước thực hiện
<aside>
✍🏻 Các khai báo và các hàm được viết trong State của StatefulWidget
</aside>
classCountDownCustomCubitPageextendsStatefulWidget{constCountDownCustomCubitPage({Key? key, required this.seconds}):super(key: key);final int seconds;State<CountDownCustomCubitPage>createState()=>CountDownCustomCubitPageState();}classCountDownCustomCubitPageStateextendsState<CountDownCustomCubitPage>{/*
Code here
*/}- Tạo StreamController để quản lý luồng dữ liệu
finalStreamController<int> _timeStreamController =StreamController();Stream<int>get _timeStream => _timeStreamController.stream;- Tạo StreamSubscription để quản lý việc đếm ngược
_timeSubscription?.pause();//tạm dừng
_timeSubscription?.resume();//tiếp tục
_timeSubscription?.cancel();//huỷ bỏStreamSubscription? _timeSubscription;void_onStart(){
_timeSubscription =Stream.periodic(constDuration(seconds:1),(computationCount)=> _start - computationCount).listen((event){
_timeStreamController.add(event);if(event ==0){_onFinish();}},);}//nhớ _timeSubscription?.dispose(); ở dispose()void_onResume(){if(_timeSubscription?.isPaused ??false){
_timeSubscription?.resume();}}void_onPause(){if(!(_timeSubscription?.isPaused ??true)){
_timeSubscription?.pause();}}void_onFinish(){
_timeSubscription?.cancel();
_timeSubscription =null;}void_onReset(){
_timeSubscription?.cancel();
_timeSubscription =null;
_timeStreamController.add(_start);}
Stream.periodic(**const** Duration(seconds: 1), (computationCount) => **_start** - computationCount)tạo một Stream trả về giá trị định kỳ sau 1 giây.
**_timeStreamController**.add(event); add thời gian mới vào Stream
- Hiển thị dữ liệu lên giao diện
StreamBuilder<int>(
initialData: _start,
stream: _timeStream,
builder:(context, snapshot){if(snapshot.hasData){final int time = snapshot.data!;var separateWidget =Padding(
padding:constEdgeInsets.symmetric(horizontal:4),
child:Text(':',
style:Theme.of(context).textTheme.headline2?.copyWith(
fontFamily:'BlackOpsOne',),
textAlign:TextAlign.center,),);returnRow(
mainAxisSize:MainAxisSize.min,
crossAxisAlignment:CrossAxisAlignment.center,
children:[_TextWidget(
number: time.hour.tens,),_TextWidget(
number: time.hour.ones,),
separateWidget,_TextWidget(
number: time.minute.tens,),_TextWidget(
number: time.minute.ones,),
separateWidget,_TextWidget(
number: time.second.tens,),_TextWidget(
number: time.second.ones,),],),;}returnconstSizedBox();}),- Extension lấy thông tin thời gian từ kiểu int
extensionIntToTimeon int {///lấy thông tin giờ
int get hour =>_getHour();
int _getHour(){return(this/3600).floor();// return Duration(seconds: this).inHours;}///lấy thông tin giờ
int get minute =>_getMinute();
int _getMinute(){return(this/60).floor()%60;}///lấy thông tin giây
int get second =>_getSecond();
int _getSecond(){returnthis%60;}///format hiển thị số ở hàng chục
int get tens =>_getTens();
int _getTens(){if(this>=10){return((this-(this%10))/10).round();}return0;}///format hiển thị số ở hàng đơn vị
int get ones =>_getOnes();
int _getOnes(){returnthis%10;}}- Nhằm
giúp cho các công việc không thực hiện lại công việc nó đang thực hiệnthực hiện gọi các function thông qua streamController
voidinitState(){super.initState();setTime();///việc quản lý các sự kiện bằng stream ở đây///giúp cho các công việc không thực hiện lại công việc nó đang thực hiện///bằng hàm distinct()
_functionSubscription = _functionController.stream.distinct().listen((event){switch(event){caseCountDownEvent.start:_onStart();break;caseCountDownEvent.pause:_onPause();break;caseCountDownEvent.resume:_onResume();break;caseCountDownEvent.reset:_onReset();break;}});}//ví dụButton(
onTap:(){
_functionController.add(CountDownEvent.resume);},
title:'Resume',),<aside>
📖 Full code →
</aside>
[Demo] Tự tạo flutter_bloc theo style Cubit bằng Stream
- Nhằm tăng tính đọc hiểu dữ liệu và phân chia mã nguồn giữa giao diện và logic chúng ta cần tách biệt xử lý ở các mà
Bước 1: Tạo abstract class để định nghĩa các biến và hàm cơ bản của bloc
abstractclassCustomCubit<T>{CustomCubit(T initValue){
_streamCtrl =StreamController.broadcast()..add(initValue);//nếu dùng thêm thư viện rxdart thì xài seed để init value}
late StreamController<T> _streamCtrl;Stream<T>get stream => _streamCtrl.stream;voidemit(T state){
_streamCtrl.add(state);}voidclose(){
_streamCtrl.close();}}<aside>
⚠️ **`_streamCtrl** = StreamController.broadcast()..add(initValue);`
stream không thể nhận được event này bới stream.listen được gọi sau khi hàm hàm constructor chạy.
Có thể tự initValue ở StreamBuilder hoặc dùng thư viện RxDart có hổ trợ chức năng này
</aside>
Bước 2:
- Tạo TimerCubit
- Mang hết khai báo và hàm từ vào trong TimerCubit
classTimerCubitextendsCustomCubit<int>{TimerCubit(int initValue):super(initValue){
startTime = initValue;}StreamSubscription? _subscription;
late StreamSubscription _controlSubscription;finalStreamController<CountDownEvent> _timerController =StreamController.broadcast();Stream<CountDownEvent>get timerControllerStream => _timerController.stream;
int startTime =0;voidclose(){
_subscription?.cancel();
_controlSubscription.cancel();
_timerController.close();super.close();}voidinit(){
_controlSubscription = _timerController.stream.distinct().listen((event){switch(event){caseCountDownEvent.start:_onStart();break;caseCountDownEvent.pause:_onPause();break;caseCountDownEvent.resume:_onResume();break;caseCountDownEvent.reset:_onReset();break;}});}void_setTime(int time){emit(time);}void_onStart(){if(_subscription !=null){_onReset();}
_subscription =Stream.periodic(constDuration(seconds:1),(computationCount)=> startTime - computationCount).listen((time){_setTime(time);if(time ==0){_onFinish();}},);}void_onResume(){if(_subscription?.isPaused ??false){
_subscription?.resume();}}void_onPause(){if(!(_subscription?.isPaused ??true)){
_subscription?.pause();}}void_onFinish(){
_subscription?.cancel();
_subscription =null;}void_onReset(){
_subscription?.cancel();
_subscription =null;}voidtimerController(CountDownEvent event){
_timerController.add(event);}}Bước 3: Tạo BlocProvider
- Việc tạo BlocProvider để quản lý instance của bloc bằng context nếu không muốn làm theo cách này có thể tạo biến toàn cục để quán lý riêng.
- BlocProvider được ứng dùng từ InheritedWidget
// final timerCubit = CustomBlocProvider.of<TimerCubit>(context);classCustomBlocProvider<TextendsCustomCubit>extendsInheritedWidget{constCustomBlocProvider({super.key,
required this.bloc,
required super.child,});finalT bloc;staticCustomBlocProvider? maybeOf<TextendsCustomCubit>(BuildContext context){return context.dependOnInheritedWidgetOfExactType<CustomBlocProvider<T>>();}staticT of<TextendsCustomCubit>(BuildContext context){finalCustomBlocProvider? result = maybeOf<T>(context);assert(result !=null,'No BlocProvider found in context');return result!.bloc asT;}
bool updateShouldNotify(CustomBlocProvider oldWidget)=> bloc != oldWidget.bloc;}final timerCubit = context.read<TimerCubit>();extensionReadCustomBlocProviderOfContextonBuildContext{T read<TextendsCustomCubit>(){returnCustomBlocProvider.of<T>(this);}}Bước 4: Khai báo
classCountDownCustomCubitPageextendsStatefulWidget{constCountDownCustomCubitPage({Key? key, required this.seconds}):super(key: key);final int seconds;State<CountDownCustomCubitPage>createState()=>CountDownCustomCubitPageState();}classCountDownCustomCubitPageStateextendsState<CountDownCustomCubitPage>{
late TimerCubit _timerCubit;voidinitState(){super.initState();
_timerCubit =TimerCubit(widget.seconds)..init();}voiddispose(){
_timerCubit.close();super.dispose();}Widgetbuild(BuildContext context){returnCustomBlocProvider<TimerCubit>(
bloc: _timerCubit,
child:Scaffold(
appBar:AppBar(title:Text("Timer test")),
body:Center(
child:SingleChildScrollView(
child:Column(
children:<Widget>[_Content(),/*Some Code*/],),),),),);}}class _Content extendsStatelessWidget{const_Content({Key? key}):super(key: key);Widgetbuild(BuildContext context){final timerCubit = context.read<TimerCubit>();// final timerCubit = CustomBlocProvider.of<TimerCubit>(context);returnStreamBuilder<int>(
initialData: timerCubit.startTime,
stream: timerCubit.stream,
builder:(context, snapshot){if(snapshot.hasData){final int time = snapshot.data!;var separateWidget =Padding(
padding:constEdgeInsets.symmetric(horizontal:4),
child:Text(':',
style:Theme.of(context).textTheme.headline2?.copyWith(
fontFamily:'BlackOpsOne',),
textAlign:TextAlign.center,),);returnFittedBox(
child:InkWell(
onTap:(){},
child:Row(
mainAxisSize:MainAxisSize.min,
crossAxisAlignment:CrossAxisAlignment.center,
children:[_TextWidget(
number: time.hour.tens,),_TextWidget(
number: time.hour.ones,),
separateWidget,_TextWidget(
number: time.minute.tens,),_TextWidget(
number: time.minute.ones,),
separateWidget,_TextWidget(
number: time.second.tens,),_TextWidget(
number: time.second.ones,),],),),);}returnconstSizedBox();});}}<aside>
📖 Full code →
</aside>
Các tính năng có thể bổ xung
- Có thể custom _Content thành BlocBuilder
- Thêm các tính năng như buildWhen, listenner
- MultiBlocProvider
Tham khảo
https://dart.dev/tutorials/language/streams
https://medium.flutterdevs.com/exploring-streambuilder-in-flutter-5958381bca67
Nguồn: viblo.asia