Как перезапустить асинхронную функцию в блоке, пока она работает?

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

class ReaderBloc {
  bool started = false;
  final _publishStream = PublishSubject<String>();
  Stream<String> get publishStream => _publishStream.stream;

  startReading() async {
    List<String> lines = ['a','b','c','d','e','f','g','h'];

    for (String l in lines) {
      if (started) {
        started = false;
        _publishStream.done;
        break;
      }
      _publishStream.add(l);
      await Future.delayed(const Duration(milliseconds: 3000));
    }
  }

  dispose() {
    _publishStream.close();
  }
}

и представление, которое просто получает и отображает эти строки из блочной логики

class Reader extends StatefulWidget {
  const Reader({super.key});
  @override
  State<Reader> createState() => _ReaderState();
}

class _ReaderState extends State<Reader> {
  late ReaderBloc readerBloc;

  @override
  void dispose() {
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    readerBloc = Provider.of<ReaderBloc>(context);

    return DefaultTabController(
      length: 1,
      child: Scaffold(
        appBar: AppBar(
          actions: <Widget>[
            IconButton(
              icon: const Icon(
                Icons.refresh,
                color: Colors.white,
              ),
              onPressed: () {
                // seems this part doesn't work
                readerBloc.started = true;
                readerBloc.startReading();
              },
            )
          ],
        ),
        body: TabBarView(
          children: [
            Scaffold(body: tabBody(readerBloc.publishStream)),
          ],
        ),
      ),
    );
  }

  tabBody(Stream<String> stream) {
    return StreamBuilder<String>(
      stream: stream,
      builder: (context, snapshot) {
        if (!snapshot.hasData) {
          return const Center(child: CircularProgressIndicator());
        }
        return Text(
          snapshot.data.toString(),
        );
      },
    );
  }
}

прежде чем мы перейдем к маршруту Reader, мы выполняем следующее, то есть ничего особенного, но получаем наш блок, начинаем читать асинхронную функцию и, наконец, переходим к маршруту Reader

Provider.of<ReaderBloc>(context, listen: false).startReading();
Navigator.of(context).push(MaterialPageRoute(builder: (context) => const Reader()));

Неправильная часть — в режиме чтения есть кнопка обновления, чтобы снова начать чтение с начала, но она не работает должным образом, т. е. не останавливает текущее чтение и не начинает новое чтение. При этом я не получаю ошибок.

Что я пропустил с этим, так это то, что он не работает должным образом?