Flutter event gets lost in stream

Question

I've recently started using state management in flutter and have pretty much settled on BloC. However I do not use the bloc package or any similar dependency for it since my codebase is not that complex and I like writing it on my own. But I've come across an issue i just can't seem to get fixed. In summary, I have a stream that seems to just loose a certain event everytime i put it in the sink.

I've built an example app that is much simpler than my actual codebase, but still has this issue. The app consists of two pages with the first (main)page displaying a list of strings. When you click on one of the list-items, the second page will open up and the string/the item you clicked on will be displayed on this page.

Each of the two pages has an own BloC, but since the two pages need to be somewhat connected to get the selected item from the first to the second page, there is a third AppBloC which gets injected into the other two BloCs. It exposes a sink and a stream to send data between the other two BloCs.

The only third party package used in this example is kiwi (0.2.0) for dependency injection.

my main.dart is pretty simple and looks like this:

import 'package:flutter/material.dart';
import 'package:kiwi/kiwi.dart' as kw; //renamed to reduce confusion with flutter's own Container widget
import 'package:streams_bloc_test/first.dart';
import 'package:streams_bloc_test/second.dart';
import 'bloc.dart';


kw.Container get container => kw.Container(); //Container is a singleton used for dependency injection with Kiwi

void main() {
  container.registerSingleton((c) => AppBloc()); //registering AppBloc as a singleton for dependency injection (will be injected into the other two blocs)
  runApp(MyApp());
}

class MyApp extends StatefulWidget {
  @override
  _MyAppState createState() => _MyAppState();
}

class _MyAppState extends State<MyApp> {
  final appBloc = container.resolve(); //injecting AppBloc here just to dispose it when the App gets closed

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

  @override
  Widget build(BuildContext context) {
    return MaterialApp( //basic MaterialApp with two routes
      title: 'Streams Test',
      theme: ThemeData.dark(),
      initialRoute: "first",
      routes: {
        "first": (context) => FirstPage(),
        "first/second": (context) => SecondPage(),
      },
    );
  }
}

then there are the two pages:
first.dart:

import 'package:flutter/material.dart';
import 'package:streams_bloc_test/bloc.dart';

class FirstPage extends StatefulWidget { //First page that just displays a simple list of strings
  @override
  _FirstPageState createState() => _FirstPageState();
}

class _FirstPageState extends State<FirstPage> {
  final bloc = FirstBloc();

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

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: Text("FirstPage")),
      body: StreamBuilder<List<String>>(
          initialData: [],
          stream: bloc.list,
          builder: (context, snapshot) {
            return ListView.builder( //displays list of strings from the stream
              itemBuilder: (context, i){
                return ListItem(
                  text: snapshot.data[i],
                  onTap: () { //list item got clicked
                    bloc.selectionClicked(i); //send selected item to second page
                    Navigator.pushNamed(context, "first/second"); //open up second page
                  },
                );
              },
              itemCount: snapshot.data.length,
            );
          }),
    );
  }
}

class ListItem extends StatelessWidget { //simple widget to display a string in the list
  final void Function() onTap;
  final String text;

  const ListItem({Key key, this.onTap, this.text}) : super(key: key);

  @override
  Widget build(BuildContext context) {
    return InkWell(
      child: Container(
        padding: EdgeInsets.all(16.0),
        child: Text(text),
      ),
      onTap: onTap,
    );
  }
}

second.dart:

import 'package:flutter/material.dart';
import 'package:streams_bloc_test/bloc.dart';

class SecondPage extends StatefulWidget { //Second page that displays a selected item
  @override
  _SecondPageState createState() => _SecondPageState();
}

class _SecondPageState extends State<SecondPage> {
  final bloc = SecondBloc();

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

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: StreamBuilder( //selected item is displayed as the AppBars title
          stream: bloc.title,
          initialData: "Nothing here :/", //displayed when the stream does not emit any event
          builder: (context, snapshot) {
            return Text(snapshot.data);
          },
        ),
      ),
    );
  }
}

and finally here are my three BloCs:
bloc.dart:

import 'dart:async';
import 'package:kiwi/kiwi.dart' as kw;

abstract class Bloc{
  void dispose();
}

class AppBloc extends Bloc{ //AppBloc for connecting the other two Blocs
  final _selectionController = StreamController<String>(); //"connection" used for passing selected list items from first to second page

  Stream<String> selected;
  Sink<String> get select => _selectionController.sink;

  AppBloc(){
    selected = _selectionController.stream.asBroadcastStream(); //Broadcast stream needed if second page is opened/closed multiple times
  }

  @override
  void dispose() {
    _selectionController.close();
  }
}

class FirstBloc extends Bloc { //Bloc for first Page (used for displaying a simple list)
  final appBloc = kw.Container().resolve<AppBloc>(); //injected AppBloc
  final listItems = ["this", "is", "a", "list"]; //example list items

  final _listController = StreamController<List<String>>();

  Stream<List<String>> get list => _listController.stream;

  FirstBloc(){
    _listController.add(listItems); //initially adding list items
  }

  selectionClicked(int index){ //called when a list item got clicked
    final item = listItems[index]; //obtaining item
    appBloc.select.add(item); //adding the item to the "connection" in AppBloc
    print("item added: $item"); //debug print
  }

  @override
  dispose(){
    _listController.close();
  }
}

class SecondBloc extends Bloc { //Bloc for second Page (used for displaying a single list item)
  final appBloc = kw.Container().resolve<AppBloc>(); //injected AppBloc

  final _titleController = StreamController<String>(); //selected item is displayed as the AppBar title

  Stream<String> get title => _titleController.stream;

  SecondBloc(){
    awaitTitle(); //needs separate method because there are no async constructors
  }

  awaitTitle() async {
    final title = await appBloc.selected.first; //wait until the "connection" spits out the selected item
    print("recieved title: $title"); //debug print
    _titleController.add(title); //adding the item as the title
  }

  @override
  void dispose() {
    _titleController.close();
  }

}

The expected behavior would be, that everytime I click on one of the list-items, the second page would open up and display that item as its title. But that's not what is happening here. Executing the above code will look like this. The first time when you click on a list item, everything works just as intended and the string "this" is set as the second page's title. But closing the page and doing so again, "Nothing here :/" (the default string/initial value of the StreamBuilder) gets displayed. The third time however, as you can see in the screencap, the app starts to hang because of an exception:

Unhandled Exception: Bad state: Cannot add event after closing

The exception occurrs in the BloC of the second page when trying to add the recieved string into the sink so it can be displayed as the AppBar's title:

  awaitTitle() async {
    final title = await appBloc.selected.first;
    print("recieved title: $title");
    _titleController.add(title); //<-- thats where the exception get's thrown
  } 

This seems kind of weird at first. The StreamController (_titleController) is only getting closed when the page is also closed (and the page has clearly not gotten closed yet). So why is this exception getting thrown? So just for fun I uncommented the line where _titleController gets closed. It will probably create some memory leaks, but that's fine for debugging:

  @override
  void dispose() {
    //_titleController.close();
  }

Now that there are no more exceptions that will stop the app from executing, the following happens: The first time is the same as before (title gets displayed - expected behavior), but all the following times the default string gets displayed, not matter how often you try it. Now you may have noticed the two debug prints in bloc.dart. The first tells me when an event is added to the AppBloc's sink and the second one when the event is recieved. Here is the output:

//first time
  item added: this
  recieved title: this
//second time
  item added: this
//third time
  item added: this
  recieved title: this
//all the following times are equal to the third time...

So as you can clearly see, the second time the event somehow got lost somewhere. This also explains the exception I was getting before. Since the title never got to the second page on the second try, the BloC was still waiting for an event to come through the stream. So when i clicked on the item the third time, the previous bloc was still active and recieved the event. Of course then the page and the StreamController were already closed, ergo the exception. So everytime the default string is displayed the following times is basically just because the previous page was still alive and caught the string...

So the part I can't seem to figure out is, where did that second event go? Did i miss something really trivial or get something wrong somewhere? I tested this on the stable channel (v1.7.8) as well as on the master channel (v1.8.2-pre.59) on multiple different android versions. I used dart 2.4.0.

Answer 1

You can try to use Rxdart's BehaviorSubject instead of StreamController in your main AppBloc

final _selectionController = BehaviorSubject<String>();

And your stream listener can be a just stream instead of a broadcast stream

selected = _selectionController.stream;

The reason I am suggesting this is because RxDart's BehaviorSubject makes sure it always emits the last stream at every point in time wherever it is being listened to.