Flutter: Stream 实现定时轮询功能

Stream 是 dart 的核心库之一。Future 用于表示单个运算的结果，而 Stream 则表示多个结果的序列。

今天我们来分享一段代码，通过 Stream 来实现定时轮询功能：

typedef Future<T> FutureGenerator<T>();

class StreamTool {

  /// interval 轮询时间间隔
  /// maxCount 最大轮询数
  Stream<T> timedPolling<T>(Duration interval, FutureGenerator<T> future,
      [int maxCount]) {
    StreamController<T> controller;
    int counter = 0;
    bool polling = true;

    void stopTimer() {
      polling = false;
    }

    void tick() async {
      counter++;
      T result = await future();
      if (!controller.isClosed) {
        controller.add(result);
      }
      if (counter == maxCount) {
        stopTimer();
        controller.close();
      } else if (polling) {
        Future.delayed(interval, tick);
      }
    }

    void startTimer() {
      polling = true;
      tick();
    }
    
    //StreamSubscription调用pause，cancel时，stream里面的轮询也能响应暂停或取消
    controller = StreamController<T>(
      onListen: startTimer,
      onPause: stopTimer,
      onResume: startTimer,
      onCancel: stopTimer,
    );

    return controller.stream;
  }
}

使用方式，如下test方法示例：

//配合test，模拟future任务
Future<String> testFuture() async {
    //模拟耗时
    await Future.delayed(Duration(seconds: 1));
    String randomStr = Random().nextInt(10).toString();
    return Future.value(randomStr);
  }

void test() {
    var pollingStream = timedPolling(const Duration(seconds: 1), testFuture, 15);
    StreamSubscription<String> subscription;
    int counter = 0;
    //进行流内容监听
    subscription = pollingStream.listen((result) {
      counter++;
      print("stream result is $result");
      if (counter == 5) {
        // 在第5次打印时，通知subscription暂停，5秒后恢复，暂停期间，stream内的轮询也会暂停
        subscription?.pause(Future.delayed(const Duration(seconds: 5)));
      }
    });
  }

知识点# Stream有两种类型：

第一种： Single-Subscription 类型的 Stream，只能设置一次监听，重复设置会丢失原来的事件。上面的代码实现属于这种类型，好处是我们能在生命周期进行关联方法的调用：

controller = StreamController<T>(
      onListen: startTimer, //监听成立时，开启轮询任务
      onPause: stopTimer, //暂停时，关闭轮询
      onResume: startTimer, //暂停恢复后，重新开始轮询
      onCancel: stopTimer, //取消时，停止轮询
    )

第二种：Broadcast 类型的 Stream，可以在同一时间设置多个不同的监听器同时监听。我们常用这种类型的 Stream 实现广播通知，有兴趣的同学可以看看之前写的文章：Flutter使用Stream进行消息通知
关键代码如下：

StreamController<T>.broadcast().stream