Множественный доступ к одному потоку

Question

Контекст

Код ниже представляет собой абстракцию, в которой MyClass это какой-то менеджер закачек.

import 'dart:async';

Future<void> main() async {
  MyClass().test().listen((v) => print('t1: $v')).onError(print);
  final commomClass = MyClass();
  commomClass.test().listen((v) => print('t2: $v')).onError(print);
  commomClass.test().listen((v) => print('t3: $v')).onError(print);
}

class MyClass {
  bool _isDownloadInProgress = false;
  int _i = 0;
  StreamController<int> _sc;
  
  Stream<int> test() async* {
    if (_isDownloadInProgress) {
      throw Exception('Download already in progress');
    } else {
      _sc = StreamController<int>();
    }
    
    Timer.periodic(
      const Duration(seconds: 1),
      (t) {
        if (_i == 4) {
          _isDownloadInProgress = false;
          _i = 0;
          _sc.close();
          t.cancel();
        } else {
          _sc.add(_i++);
        }
      },
    );

    yield* _sc.stream;
  }
}

Вопрос

Я ожидал, что после выполнения этого кода он сгенерирует значения t1 и t2, а результат t3 сгенерирует "Загрузка уже выполняется" только один раз. Например:

t1: 0
t2: 0
t3: Download already in progress
t1: 1
t2: 1
t1: 2
t2: 2
t1: 3
t2: 3

Но выводит все четыре t1значения, восемь t3ценности и не "Скачать уже в процессе" сообщение:

t1: 0
t3: 0
t3: 1
t1: 1
t3: 2
t3: 3
t1: 2
t3: 0
t1: 3
t3: 1
t3: 2
t3: 3

Для меня t1 значения будут выводиться правильно, t2 также будет выводиться правильно, а t3 выводит сообщение "Загрузка уже выполняется", потому что, поскольку все выполняется асинхронно, он будет пытаться "загрузить" что-то, что уже загружается (поскольку test() метод был вызван в том же экземпляре MyClass).

Что мне не хватает?

Answer 1

Во-первых, ваш код никогда не устанавливает _isDownloadInProgress значение true, поэтому нет никаких причин, по которым когда-либо появлялось сообщение "Загрузка уже выполняется".

На самом деле это причина второй ошибки. Когда вы звоните в t3listen, поскольку _isDownloadInProgress всегда ложно, это приводит к _sc перезаписывается в дополнение к новому Timer.periodicставится в очередь. Когда срабатывает каждый таймер, он ссылается на_sc, который теперь содержит t3 listen, поэтому вы получаете два таймера, отправляющих события в один и тот же контроллер потока, поэтому вы видите удвоение событий t3.

Просто установка _isDownloadInProgress = true до создания экземпляра таймера достаточно для получения ожидаемых результатов:

class MyClass {
  bool _isDownloadInProgress = false;
  int _i = 0;
  StreamController<int> _sc;
  
  Stream<int> test() async* {
    if (_isDownloadInProgress) {
      throw Exception('Download already in progress');
    } else {
      _sc = StreamController<int>();
    }
    
    _isDownloadInProgress = true; // Add this line
    Timer.periodic(
      const Duration(seconds: 1),
      (t) {
        if (_i == 4) {
          _isDownloadInProgress = false;
          _i = 0;
          _sc.close();
          t.cancel();
        } else {
          _sc.add(_i++);
        }
      },
    );

    yield* _sc.stream;
  }
}

Полученные результаты:

Exception: Download already in progress
t1: 0
t2: 0
t1: 1
t2: 1
t1: 2
t2: 2
t1: 3
t2: 3

Answer 2

Вы забыли установить _isDownloadInProgress = true; внутри функции.

Пытаться:

if (_isDownloadInProgress) {
    throw Exception('Download already in progress');
} else {
  _sc = StreamController<int>();
  _isDownloadInProgress = true;
}

Выход

Exception: Download already in progress
t1: 0
t2: 0
t1: 1
t2: 1
t1: 2
t2: 2
t1: 3
t2: 3

Ты можешь сделать t3 интереснее, отложив его начало:

Future.delayed(const Duration(seconds: 2),
      () => commomClass.test().listen((v) => print('t3: $v')).onError(print));

Выход

t1: 0
t2: 0
Exception: Download already in progress
t1: 1
t2: 1
t1: 2
t2: 2
t1: 3
t2: 3