dart:async

метод timeout

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

Когда кто-либо прослушивает возвращённый поток и проходит более timeLimit без выдачи этим потоком каких-либо событий, вызывается функция onTimeout, которая затем может выдавать дополнительные события в возвращённый поток.

Обратный отсчёт начинается, когда начинают прослушивать возвращённый поток, и запускается заново при выдаче события этим потоком, а также при приостановке и возобновлении прослушивания возвращённого потока. Обратный отсчёт останавливается, когда прослушивание возвращённого потока приостанавливается или отменяется. Когда обратный отсчёт завершается и вызывается функция onTimeout, новый отсчёт не начинается, даже если выдаются события. Если промежуток между событиями этого потока кратен timeLimit, между событиями произойдёт не более одного тайм-аута.

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

Если onTimeout не указан, тайм-аут помещает TimeoutException в канал ошибок возвращённого потока. Если вызов onTimeout выбрасывает исключение, оно передаётся как ошибка в возвращённый поток.

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

Пример:

Future<String> waitTask() async {
  return await Future.delayed(
      const Duration(seconds: 4), () => 'Complete');
}
final stream = Stream<String>.fromFuture(waitTask())
    .timeout(const Duration(seconds: 2), onTimeout: (controller) {
  print('TimeOut occurred');
  controller.close();
});

stream.listen(print, onDone: () => print('Done'));

// Outputs:
// TimeOut occurred
// Done

Реализация

Stream<T> timeout(Duration timeLimit, {void onTimeout(EventSink<T> sink)?}) {
  _StreamControllerBase<T> controller;
  if (isBroadcast) {
    controller = new _SyncBroadcastStreamController<T>(null, null);
  } else {
    controller = new _SyncStreamController<T>(null, null, null, null);
  }

  Zone zone = Zone.current;
  // Register callback immediately.
  _TimerCallback timeoutCallback;
  if (onTimeout == null) {
    timeoutCallback = () {
      controller.addError(
          new TimeoutException("No stream event", timeLimit), null);
    };
  } else {
    var registeredOnTimeout =
        zone.registerUnaryCallback<void, EventSink<T>>(onTimeout);
    var wrapper = new _ControllerEventSinkWrapper<T>(null);
    timeoutCallback = () {
      wrapper._sink = controller; // Only valid during call.
      zone.runUnaryGuarded(registeredOnTimeout, wrapper);
      wrapper._sink = null;
    };
  }

  // All further setup happens inside `onListen`.
  controller.onListen = () {
    Timer timer = zone.createTimer(timeLimit, timeoutCallback);
    var subscription = this.listen(null);
    // Set up event forwarding. Each data or error event resets the timer
    subscription
      ..onData((T event) {
        timer.cancel();
        timer = zone.createTimer(timeLimit, timeoutCallback);
        // Controller is synchronous, and the call might close the stream
        // and cancel the timer,
        // so create the Timer before calling into add();
        // issue: https://github.com/dart-lang/sdk/issues/37565
        controller.add(event);
      })
      ..onError((Object error, StackTrace stackTrace) {
        timer.cancel();
        timer = zone.createTimer(timeLimit, timeoutCallback);
        controller._addError(
            error, stackTrace); // Avoid Zone error replacement.
      })
      ..onDone(() {
        timer.cancel();
        controller.close();
      });
    // Set up further controller callbacks.
    controller.onCancel = () {
      timer.cancel();
      return subscription.cancel();
    };
    if (!isBroadcast) {
      controller
        ..onPause = () {
          timer.cancel();
          subscription.pause();
        }
        ..onResume = () {
          subscription.resume();
          timer = zone.createTimer(timeLimit, timeoutCallback);
        };
    }
  };

  return controller.stream;
}