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

注意:

streamSubscription 的 cancle 方法不能调用，否则 onData 接收不到Stream事件。

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 --  对不起我错了