How to restart async function in bloc while it's working?

There is a simple bloc, which uses rxdart controller. It does nothing but streaming srings in general. There is a logic in it to start reading again.

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();
  }
}

and a view which simply receive and display those strings from bloc logic

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(),
        );
      },
    );
  }
}

before we go on Reader route we execute the following, i.e. nothing special but getting our bloc, start reading async function and finally go to the Reader route

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

The wrong part — there is a refresh button in Reader view to start reading again from the beginning, but it does not work as expected, i.e. it does not stop current reading and does not start new reading. Wherein I get no errors.

What I missed with this is that it doesn't work as expected?

these changes solved the problem

in view:

onPressed: () {
    readerBloc.started = true;
},

in bloc:

if (started) {
    started = !started;
    startReading();
    break;
}