import 'dart:async';
import 'package:flutter/material.dart';
class StreamDemo extends StatelessWidget {
@override
Widget build(BuildContext context) {
return Scaffold(
appBar: AppBar(
title: Text('StreamDemo'),
),
body: StreamDemoHome(),
);
}
}
class StreamDemoHome extends StatefulWidget {
@override
_StreamDemoHomeState createState() => _StreamDemoHomeState();
}
class _StreamDemoHomeState extends State<StreamDemoHome> {
StreamSubscription _streamSubscription;
@override
void initState() {
super.initState();
print('Create a stream');
Stream<String> _streamDemo = Stream.fromFuture(fetchData());
print('Start listening on a stream');
// 添加订阅
_streamSubscription = _streamDemo.listen(onData, onError: onError, onDone: onDone);
print('Initialize completed.');
}
void onDone() {
print('Done!');
}
void onError(error) {
print('Error: $error');
}
void onData(String data) {
print('$data');
}
// 暂停订阅
void _pauseStream(){
print('Pause subscription');
_streamSubscription.pause();
}
// 恢复订阅
void _resumeStream(){
print('Resume subscription');
_streamSubscription.resume();
}
// 取消订阅
void _cancelStream(){
print('Cancel subscription');
_streamSubscription.cancel();
}
Future<String> fetchData() async {
await Future.delayed(Duration(seconds: 5));
throw 'Something happen';
// return 'hello~';
}
@override
Widget build(BuildContext context) {
return Container(
child: Center(
child: Row(
mainAxisAlignment: MainAxisAlignment.center,
children: <Widget>[
FlatButton(
child: Text('Pause'),
onPressed: _pauseStream,
),
FlatButton(
child: Text('Resume'),
onPressed: _resumeStream,
),
FlatButton(
child: Text('Cancel'),
onPressed: _cancelStream,
),
],
),
),
);
}
}
控制器(controller)
StreamController<String> _streamDemo;
@override
void initState() {
super.initState();
print('Create a stream');
_streamDemo = StreamController<String>();
print('Start listening on a stream');
// 添加订阅
_streamSubscription = _streamDemo.stream.listen(onData, onError: onError, onDone: onDone);
print('Initialize completed.');
}
void _addDataToStream() async {
print('Add data to stream.');
String data = await fetchData();
_streamDemo.add(data);
}
sink
一般来说,对于stream,我们使用stream输出,sink输入
StreamSink _sinkDemo;
@override
void initState() {
super.initState();
print('Create a stream');
_streamDemo = StreamController<String>();
_sinkDemo = _streamDemo.sink;
print('Start listening on a stream');
// 添加订阅
// 当有数据时,通过stream流出
_streamSubscription = _streamDemo.stream.listen(onData, onError: onError, onDone: onDone);
print('Initialize completed.');
}
void _addDataToStream() async {
print('Add data to stream.');
String data = await fetchData();
_sinkDemo.add(data);
}
多次订阅(broadcast)
@override
void initState() {
super.initState();
print('Create a stream');
// Stream<String> _streamDemo = Stream.fromFuture(fetchData());
_streamDemo = StreamController.broadcast();
_sinkDemo = _streamDemo.sink;
print('Start listening on a stream');
// 添加订阅
_streamSubscription = _streamDemo.stream.listen(onData, onError: onError, onDone: onDone);
_streamSubscription = _streamDemo.stream.listen(onDataTwo, onError: onError, onDone: onDone);
print('Initialize completed.');
}
StreamBuilder
根据Stream上的数据渲染组件
StreamBuilder(
stream: _streamDemo.stream,
initialData: '...',
builder: (context, snapshot){
print(snapshot);
return Text('${snapshot.data}');
},
),
网友评论