Flutter 的 StreamSubscription 探究

一、简介

StreamSubscription 是订阅Stream中的事件。当对Stream使用listen监听时，则返回一个StreamSubscription对象。StreamSubscription是对当前Stream的监听产生的状态的管理对象，它能获取订阅事件的状态(是否被暂停)、订阅事件的取消、订阅事件的恢复以及保留用于处理事件的回调(onData,onDone,onError)。

二、构建

对Stream使用listen监听，即返回一个StreamSubscription对象，代码如下:

// 构建
void init() {
  var stream = Stream.fromIterable([1, 2, 3, 4]);
  StreamSubscription<int> streamSubscription = stream.listen((event) {});
  print(streamSubscription);
}

三、属性

• isPaused -> bool

  isPaused 是判断Stream订阅事件是否暂停。实例代码:

  // isPaused
  void isPaused() {
    var stream = Stream.fromIterable([1, 2, 3, 4]);
    StreamSubscription<int> streamSubscription = stream.listen((event) {});
    print(streamSubscription.isPaused); // false
    streamSubscription.pause();
    print(streamSubscription.isPaused); // true
    streamSubscription.cancel();
  }
  

四、方法

• onData(void handleData(T data)?)

  onData 是StreamSubscription的回调方法，在Stream发出事件时回调。实例代码:

  void onData() {
    var stream = Stream.fromIterable([1, 2, 3, 4]);
    StreamSubscription<int> streamSubscription = stream.listen((event) {
      print("Stream -- listen -- $event");
    });
    streamSubscription.onData((data) {
      print("StreamSubscription -- onData -- $data");
    });
    // streamSubscription.cancel();
  }
  

  日志输出：

  flutter: StreamSubscription -- onData -- 1
  flutter: StreamSubscription -- onData -- 2
  flutter: StreamSubscription -- onData -- 3
  flutter: StreamSubscription -- onData -- 4
  

  注意:

  1. streamSubscription 的 cancle 方法不能调用，否则 onData 接收不到Stream事件。
  2. streamSubscription 的 onData 方法实现，则Stream 的 listen 方法的 onData 不会调用,可传 null。

• onDone(void handleDone()?)

  onDone 是StreamSubscription的回调方法，在Stream 事件发送完时回调该方法。实例代码:

  void onDone() {
    var stream = Stream.fromIterable([1, 2, 3, 4]);
    StreamSubscription<int> streamSubscription = stream.listen(null);
    streamSubscription.onData((data) {
      print("StreamSubscription -- onData -- $data");
    });
    streamSubscription.onDone(() {
      print("StreamSubscription -- onDone");
    });
  }
  

  日志输出:

  flutter: StreamSubscription -- onData -- 1
  flutter: StreamSubscription -- onData -- 2
  flutter: StreamSubscription -- onData -- 3
  flutter: StreamSubscription -- onData -- 4
  flutter: StreamSubscription -- onDone
  

• onError(Function? handleError)

  onError 是StreamSubscription的回调方法，在 Stream 发出错误事件时回调。实例代码:

  void onError() {
    var stream = Stream.error("对不起,出错了");
    StreamSubscription streamSubscription = stream.listen(null);
    streamSubscription.onData((data) {
      print("StreamSubscription -- onData -- $data");
    });
    streamSubscription.onDone(() {
      print("StreamSubscription -- onDone");
    });
    streamSubscription.onError((e) {
      print("StreamSubscription -- onError -- $e");
    });
  }
  

  日志输出:

  flutter: StreamSubscription -- onError -- 对不起,出错了
  flutter: StreamSubscription -- onDone
  

  注意: onDone 方法也被调用，切还在onError之后。

• cancel() -> Future<void>

  cancel是取消对Stream的监听。实例代码:

  void cancel() {
    var stream = Stream.fromIterable([1, 2, 3, 4]);
    StreamSubscription<int> streamSubscription = stream.listen(null);
    streamSubscription.onData((data) {
      print("StreamSubscription -- onData -- $data");
    });
    streamSubscription.cancel();
  }
  

  上面代码没有输出。

• pause([Future<void>? resumeSignal]) -> void & resume() -> void

  pause是对Stream的订阅者进行暂停；resume是对暂停的订阅者进行恢复。实例代码:

  void pauseAndresume() {
    var stream = Stream.fromIterable([1, 2, 3, 4]);
    StreamSubscription<int> streamSubscription = stream.listen(null);
    streamSubscription.onData((data) {
      print("StreamSubscription -- onData -- $data");
    });
    streamSubscription.pause();
    print(streamSubscription.isPaused); // true
    streamSubscription.resume();
    print(streamSubscription.isPaused); // false
  }
  

  注意:

  如果实现 pause 的 Future 方法，则StreamSubcription 将撤销暂停，在 Future 完成时在恢复StreamSubcription。代码如下:

  void pauseAndresume1() {
    var stream = Stream.fromIterable([1, 2, 3, 4]);
    StreamSubscription<int> streamSubscription = stream.listen(null);
    streamSubscription.onData((data) {
      print("StreamSubscription -- onData -- $data");
    });
    streamSubscription.pause(Future(() {
      print("pause -- future - 完成");
    }));
  }
  

  日志输出:

  pause -- future - 完成
  flutter: StreamSubscription -- onData -- 1
  flutter: StreamSubscription -- onData -- 2
  flutter: StreamSubscription -- onData -- 3
  flutter: StreamSubscription -- onData -- 4
  

  上面的代码同下:

  void pauseAndresume1() {
    var stream = Stream.fromIterable([1, 2, 3, 4]);
    StreamSubscription<int> streamSubscription = stream.listen(null);
    streamSubscription.onData((data) {
      print("StreamSubscription -- onData -- $data");
    });
    streamSubscription.pause();
    Future(() {
      print("pause -- future - 完成");
    }).whenComplete(() => streamSubscription.resume());
  }
  

• asFuture<E>([E? futureValue]) -> Future<E>

  asFuture 是将StreamSubscription的onDone和onError 的回调以Future的形式输出, onDone 方法不再调用，而被Future的 .then 取代;onError 方法也不调用，而被Future的 catchError 取代。实例代码:

  onDone 取代

  void asFuture() {
    var stream = Stream.fromIterable([1, 2, 3, 4]);
    StreamSubscription streamSubscription = stream.listen(null);
    streamSubscription.onData((data) {
      print("StreamSubscription -- onData -- $data");
    });
    streamSubscription.onDone(() {
      print("--onDone--");
    });
    streamSubscription.asFuture(Future(() {
      return "onDone & onError 合集";
    })).then((value) {
      print("asFuture -- $value");
    }).catchError((e) {
      print("asFuture --catchError --  $e");
    });
  }
  

  日志输出:

  flutter: StreamSubscription -- onData -- 1
  flutter: StreamSubscription -- onData -- 2
  flutter: StreamSubscription -- onData -- 3
  flutter: StreamSubscription -- onData -- 4
  flutter: asFuture -- Instance of 'Future<String>'
  

  onError 取代

  void asFuture() {
    var stream = Stream.error("对不起我错了");
    StreamSubscription streamSubscription = stream.listen(null);
    streamSubscription.onData((data) {
      print("StreamSubscription -- onData -- $data");
    });
    streamSubscription.onError((e) {
      print("--onError-- $e");
    });
    streamSubscription.asFuture(Future(() {
      return "onDone & onError 合集";
    })).then((value) {
      print("asFuture -- $value");
    }).catchError((e) {
      print("asFuture --catchError --  $e");
    });
  }
  

  日志输出:

  flutter: asFuture --catchError --  对不起我错了