Flutter DataStream not closing and re-building properly. [Bad state: Stream has already been listened to.]

Question

Okay, so I'm aware that a stream can be manufactured to listen to a stream more than once, using a broadcast system, but that's specifically NOT what I'm trying to do here.

I'm also editing this as the one answer I have received isn't currently able to resolve my issue, so any assistance would be greatly appreciated.

Effectively for some reason the code I have is not deleting the stream in it's entirety, and if re-used, it is trying to re-listen to the same stream that has already been listened-to and closed, none of which works (Obviously). Instead of trying to listen to that same stream again, I'm trying to create a NEW stream to listen to. (Deleting and cleaning away all information from the original first stream).

Original post continues below:

I'm using a DataStream template for streaming data to and/or from various parts of my program, and I'm not entirely certain how to rectify this. I'm certain it's a silly newb error, but I haven't used DataStreams enough to understand why this is happening.

Now don't get me wrong, going through a single cycle of my program works perfectly fine, no issues at all. However, once I've completed a single cycle through the program, if I try to go through a second time, I get the error:

Bad state: Stream has already been listened to.

So from this I know my program is not creating a new stream, and instead trying to re-use the original stream, and I'm not 100% certain how to stop this functionality, (Or even if I should). (Honestly the number of times I would expect multiple cycles to be completed are slim to null, but I want to resolve these kinds of errors before they become problems.)

Edit: Minimal Reproducible Example to Follow

File 1 (main.dart)

import 'package:flutter/cupertino.dart';
import 'dart:async';
import './page2.dart';
import './stream.dart';

void main() => runApp(MyApp());

DataStream stream = DataStream();


class MyApp extends StatelessWidget {
  // This widget is the root of your application.
  @override
  Widget build(BuildContext context) {
    return CupertinoApp(
      title: 'Splash Test',
      theme: CupertinoThemeData(
        primaryColor: Color.fromARGB(255, 0, 0, 255),
      ),
      home: MyHomePage(title: 'Splash Test Home Page'),
    );
  }
}

class MyHomePage extends StatefulWidget {
  MyHomePage({Key key, this.title}) : super(key: key);

  final String title;

  @override
  _MyHomePageState createState() => _MyHomePageState();
}

class _MyHomePageState extends State<MyHomePage> {
  bool textBool = false;
  int counter = 0;

  void changeTest(context) async {
    int counter = 0;
    Timer.periodic(Duration (seconds: 2), (Timer t) {
      counter++;
      stream.dataSink.add(true);
      if (counter >= 3) {
        t.cancel();
        stream.dispose();
        Navigator.pop(context);
      } 
    },);
    Navigator.push(context, CupertinoPageRoute(builder: (context) => Page2(stream: stream)));
  }



  @override
  Widget build(BuildContext context) {

    return CupertinoPageScaffold(
      child: Center(
        child: CupertinoButton(
          child: Text('To Splash'),
          onPressed: () => changeTest(context),
        ),
      ), 
    );
  }
}

File 2 (stream.dart)

import 'dart:async';

class DataStream {
  StreamController _streamController;

    StreamSink<bool> get dataSink =>
      _streamController.sink;

  Stream<bool> get dataStream =>
      _streamController.stream;

  DataStream() {
    _streamController = StreamController<bool>();
  }

  dispose() {
    _streamController?.close();
  }

}

File 3 (page2.dart)

import 'package:flutter/material.dart';
import 'package:flutter/semantics.dart';

import './main.dart';
import './stream.dart';


class Page2 extends StatefulWidget {

  DataStream stream;
  Page2({this.stream});

  @override 
  State<StatefulWidget> createState() => new PageState();
}

class PageState extends State<Page2> {


bool textChanger = false;
bool firstText = true;

Text myText() {
  if (textChanger) {
    Text text1 = new Text('Text One', 
      style: TextStyle(color: Color.fromARGB(255, 0, 0, 0)));
    return text1;
  } else {
    Text text1 = new Text('Text Two', 
      style: TextStyle(color: Color.fromARGB(255, 0, 0, 0)));
    return text1;
  }
}

void changeText() {
  if (!firstText) {
    if (textChanger) {
      print('Change One');
      setState(() { 
        textChanger = false;      
      });
    } else {
      print('Change Two');
      setState(() {  
        textChanger = true;    
      });
    }  
  } else {
    firstText = false;
  }
}


  @override 
  Widget build(BuildContext context) {
    return new Scaffold(
      body: new Container(
        child: Center(
          child: myText()
        ) 
      ),);
  }

@override
  void initState() {
    super.initState();
    widget.stream.dataStream.listen((onData) {
      changeText();
    });
  }


}

Effectively, in this example, you can click on the text, and goto the second page, which will correctly change text when told, and return to the original page once completed. That would be a single "Cycle" of my program.

And you can see that this program then immediately disposes of the stream.

The issue is that if I click the text a second time, it is still trying to listen to the original stream, rather than creating a brand new stream and starting fresh.

Why? And how do I fix this?

Answer 1

The StreamController default constructor creates a stream that allows only a single listener.

StreamController({void onListen(), void onPause(), void onResume(), dynamic onCancel(), bool sync: false })
A controller with a stream that supports only one single subscriber. [...]

If you want to have multipler listener use broadcast named constructor.

factory StreamController.broadcast({void onListen(), void onCancel(), bool sync: false })
A controller where stream can be listened to more than once. [...]

If you want your stream to have only one subscriber, remember to cancel your subscription in the widget's dispose method.

DataStream stream;
StreamSubscription subscription;

@override
void initState() {
  super.initState();
  subsription = widget.stream.listen((onData) {
    changeText();
  });
}

@override
void dispose() {
  subscription?.cancel();
  super.dispose();
}

Just keep in mind that's not a proper way of rebuilding your UI based on stream events. Take a look at the Stream Builder class.

Answer 2

What I would do is to move stream to StatefulWidget and recreate it on "to Splash" tap

In real case scenario put it in to stateful widget in widget tree where all widgets that needs access will be able to find it (in your case even higher than navigator).

import 'package:flutter/cupertino.dart';
import 'dart:async';
import 'package:flutter/material.dart';

void main() => runApp(MyApp());

class MyApp extends StatelessWidget {
  // This widget is the root of your application.
  @override
  Widget build(BuildContext context) {
    return CupertinoApp(
      title: 'Splash Test',
      theme: CupertinoThemeData(
        primaryColor: Color.fromARGB(255, 0, 0, 255),
      ),
      home: MyHomePage(title: 'Splash Test Home Page'),
    );
  }
}

class MyHomePage extends StatefulWidget {
  MyHomePage({Key key, this.title}) : super(key: key);

  final String title;

  @override
  _MyHomePageState createState() => _MyHomePageState();
}

class _MyHomePageState extends State<MyHomePage> {
  bool textBool = false;
  int counter = 0;

  DataStream stream = DataStream();

  void changeTest(context) async {
    setState(() {
      stream = DataStream();
    });
    int counter = 0;
    Timer.periodic(Duration (seconds: 2), (Timer t) {
      counter++;
      stream.dataSink.add(true);
      if (counter >= 3) {
        t.cancel();
        stream.dispose();
        Navigator.pop(context);
      }
    },);
    Navigator.push(context, CupertinoPageRoute(builder: (context) => Page2(stream: stream)));
  }

  @override
  Widget build(BuildContext context) {
    return CupertinoPageScaffold(
      child: Center(
        child: CupertinoButton(
          child: Text('To Splash'),
          onPressed: () => changeTest(context),
        ),
      ),
    );
  }
}



class DataStream {
  StreamController _streamController;

  StreamSink<bool> get dataSink =>
      _streamController.sink;

  Stream<bool> get dataStream =>
      _streamController.stream;

  DataStream() {
    _streamController = StreamController<bool>();
  }

  dispose() {
    _streamController?.close();
  }

}



class Page2 extends StatefulWidget {

  DataStream stream;
  Page2({this.stream});

  @override
  State<StatefulWidget> createState() => new PageState();
}

class PageState extends State<Page2> {


  bool textChanger = false;
  bool firstText = true;

  Text myText() {
    if (textChanger) {
      Text text1 = new Text('Text One',
          style: TextStyle(color: Color.fromARGB(255, 0, 0, 0)));
      return text1;
    } else {
      Text text1 = new Text('Text Two',
          style: TextStyle(color: Color.fromARGB(255, 0, 0, 0)));
      return text1;
    }
  }

  void changeText() {
    if (!firstText) {
      if (textChanger) {
        print('Change One');
        setState(() {
          textChanger = false;
        });
      } else {
        print('Change Two');
        setState(() {
          textChanger = true;
        });
      }
    } else {
      firstText = false;
    }
  }


  @override
  Widget build(BuildContext context) {
    return new Scaffold(
      body: new Container(
          child: Center(
              child: myText()
          )
      ),);
  }

  @override
  void initState() {
    super.initState();
    widget.stream.dataStream.listen((onData) {
      changeText();
    });
  }


}

Answer 3

Without being able to grasp your problem in its entirety, I'd just like to state that I've been having headaches pretty much every time I've used "normal" streams myself and RxDart has been like Aspirin in the world of streams for me :) Not sure if this is the answer you're looking for, but I thought I'd post it anyway - you never know!