Flutter中Stream的Sink添加数据（三）

Stream使用Sink添加数据，可以实现多次订阅，

添加数据不使用StreamController也是单次订阅的

StreamSubscription _streamSubScriptConller;
@override
void initState() {
  // TODO: implement initState
  super.initState();
  //添加数据
  Stream<String> _strema = Stream.fromFuture(getData());
  //监听数据
  _streamSubScriptConller=_strema.listen(onData,onDone: onDone,onError: onError);
}

单次订阅使用的是控制器直接添加的数据

StreamController<String> _strDemo;
@override
void initState() {
  // TODO: implement initState
  super.initState();
  _strDemo=StreamController<String>();

  _strDemo.stream.listen(onData,onDone: onDone,onError: onError);
}

//添加数据
void _addData() async {
  String data=await getData();
  _strDemo.add(data);
}

这里的_strDemo.add添加的数据都是单次订阅的

多次订阅Sink

StreamController<String> _strDemo; //控制器变量
StreamSink _sinkDemo;//多次订阅变量
@override
void initState() {
  // TODO: implement initState
  super.initState();
  创建一个可以多次订阅Stream的Contorller
  _strDemo=StreamController.broadcast();
  _sinkDemo=_strDemo.sink;
  //多次监听，只要有数据或者数据传入完成，或者出现错误都会执行，可以添加多个订阅
_strDemo.stream.listen(onData,onDone: onDone,onError: onError);
_strDemo.stream.listen(onDataTwo,onDone: onDoneTwo,onError: onErrorTwo);
}

多次订阅使用的使用StreamController.broadcast()方法，这样添加的数据才能被多次监听

import 'package:flutter/material.dart';
import 'dart:async';

class StremaDemoSink extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('Strema案例'),
      ),
      body: StrameDemoHome(),
    );
  }
}

class StrameDemoHome extends StatefulWidget {
  @override
  _StrameDemoHomeState createState() => _StrameDemoHomeState();
}

class _StrameDemoHomeState extends State<StrameDemoHome> {

  StreamController<String> _strDemo;
  StreamSink _sinkDemo;
  @override
  void initState() {
    // TODO: implement initState
    super.initState();
    _strDemo=StreamController.broadcast();
    _sinkDemo=_strDemo.sink;
    _strDemo.stream.listen(onData,onDone: onDone,onError: onError);
    _strDemo.stream.listen(onDataTwo,onDone: onDoneTwo,onError: onErrorTwo);
  }
  @override

  void dispose() {
    // TODO: implement dispose
    //关掉不需要的Stream;
    _strDemo.close();
    super.dispose();
  }
  //监听数据
  void onDataTwo(data){
    print("two $data");
  }
  //捕获错误
  void onErrorTwo(error){
    print("error is $error Two");
  }
  //数据传递完成
  void onDoneTwo(){
    print("数据传递完成 Two");
  }

  //监听数据
  void onData(data){
    print(data);
  }
  //捕获错误
  void onError(error){
    print("error is $error");
  }
  //数据传递完成
  void onDone(){
    print("数据传递完成");
  }

  Future<String> getData() async {
    await Future.delayed(Duration(seconds: 3));
    return 'hello';
  }
  //添加数据
  void _addData() async {
    String data=await getData();
    _sinkDemo.add(data);
  }
  @override
  Widget build(BuildContext context) {
    return Container(
      child: Row(
        children: <Widget>[
          FlatButton(
            child: Text('添加数据'),
            onPressed: _addData,
          ),
        ],
      ),
    );
  }
}