dart:async

Метод timeout

Создает новый поток с теми же событиями, что и этот поток.

Всякий раз, когда между двумя событиями из этого потока проходит более timeLimit, вызывается функция onTimeout.

Обратный отсчет не начинается до тех пор, пока запущенный поток не будет прослушиваться. Обратный отсчет сбрасывается каждый раз, когда событие пересылается из этого потока или когда поток приостанавливается и возобновляется.

Функция onTimeout вызывается с одним аргументом: EventSink, который позволяет помещать события в возвращаемый поток. Этот EventSink действителен только во время вызова onTimeout.

Если onTimeout опущено, таймаут просто поместит TimeoutException в канал ошибок возвращаемого потока.

Возвращаемый поток является потоком широковещательной передачи, если таковым является этот поток. Если за потоком широковещательной передачи прослушивается более одного подписчика, каждый подписка будет иметь свой индивидуальный таймер, который начинает отсчет при прослушивании, и таймеры подписок могут приостанавливаться индивидуально.

Исходный код

Stream<T> timeout(Duration timeLimit, {void onTimeout(EventSink<T> sink)}) {
  StreamController<T> controller;
  // The following variables are set on listen.
  StreamSubscription<T> subscription;
  Timer timer;
  Zone zone;
  _TimerCallback timeout;

  void onData(T event) {
    timer.cancel();
    controller.add(event);
    timer = zone.createTimer(timeLimit, timeout);
  }

  void onError(error, StackTrace stackTrace) {
    timer.cancel();
    assert(controller is _StreamController ||
        controller is _BroadcastStreamController);
    dynamic eventSink = controller;
    eventSink._addError(error, stackTrace); // Avoid Zone error replacement.
    timer = zone.createTimer(timeLimit, timeout);
  }

  void onDone() {
    timer.cancel();
    controller.close();
  }

  void onListen() {
    // This is the onListen callback for of controller.
    // It runs in the same zone that the subscription was created in.
    // Use that zone for creating timers and running the onTimeout
    // callback.
    zone = Zone.current;
    if (onTimeout == null) {
      timeout = () {
        controller.addError(
            new TimeoutException("No stream event", timeLimit), null);
      };
    } else {
      // TODO(floitsch): the return type should be 'void', and the type
      // should be inferred.
      var registeredOnTimeout =
          zone.registerUnaryCallback<dynamic, EventSink<T>>(onTimeout);
      var wrapper = new _ControllerEventSinkWrapper<T>(null);
      timeout = () {
        wrapper._sink = controller; // Only valid during call.
        zone.runUnaryGuarded(registeredOnTimeout, wrapper);
        wrapper._sink = null;
      };
    }

    subscription = this.listen(onData, onError: onError, onDone: onDone);
    timer = zone.createTimer(timeLimit, timeout);
  }

  Future onCancel() {
    timer.cancel();
    Future result = subscription.cancel();
    subscription = null;
    return result;
  }

  controller = isBroadcast
      ? new _SyncBroadcastStreamController<T>(onListen, onCancel)
      : new _SyncStreamController<T>(onListen, () {
          // Don't null the timer, onCancel may call cancel again.
          timer.cancel();
          subscription.pause();
        }, () {
          subscription.resume();
          timer = zone.createTimer(timeLimit, timeout);
        }, onCancel);
  return controller.stream;
}