Flutter nested StreamBuilders causing Bad state: Stream has already been listened to

Question

I'm trying to build a Flutter app using the BLoC pattern described in the video Flutter / AngularDart – Code sharing, better together (DartConf 2018)

A BLoC is basically a view model with Sink inputs and Stream outputs. In my example it looks a bit like this:

 class BLoC {
    // inputs
    Sink<String> inputTextChanges;
    Sink<Null> submitButtonClicks;

    // outputs
    Stream<bool> showLoading;
    Stream<bool> submitEnabled;
 }

I have the BLoC defined in a widget near the root of the hierarchy and it is passed down to widgets beneath it, including nested StreamBuilders. Like so:

The top StreamBuilder listens to a showLoading stream on the BLoC so that it can rebuild to show an overlaid progress spinner. The bottom StreamBuilder listens to a submitEnabled stream to enable/disable a button.

The problem is whenever the showLoading stream causes the top StreamBuilder to rebuild the widget it rebuilds nested widgets too. This in itself is fine and expected. However this results in the bottom StreamBuilder being recreated. When this happens it attempts to re-subscribe to the existing submitEnabled stream on the BLoC causing Bad state: Stream has already been listened to

Is there any way to accomplish this without making all of the outputs BroadcastStreams?

(There is also a chance that I'm fundamentally misunderstanding the BLoC pattern.)

---

Actual code example below:

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

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

class BlocExampleApp extends StatefulWidget {

  BlocExampleApp({Key key}) : super(key: key);

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

class _BlocExampleAppState extends State<BlocExampleApp> {

  Bloc bloc = Bloc();

  @override
  Widget build(BuildContext context) =>
      MaterialApp(
        home: Scaffold(
            appBar: AppBar(elevation: 0.0),
            body: new StreamBuilder<bool>(
                stream: bloc.showLoading,
                builder: (context, snapshot) =>
                snapshot.data
                    ? _overlayLoadingWidget(_buildContent(context))
                    : _buildContent(context)
            )
        ),
      );

  Widget _buildContent(context) =>
      Column(
          crossAxisAlignment: CrossAxisAlignment.center,
          children: <Widget>[
            TextField(
              onChanged: bloc.inputTextChanges.add,
            ),
            StreamBuilder<bool>(
                stream: bloc.submitEnabled,
                builder: ((context, snapshot) =>
                    MaterialButton(
                      onPressed: snapshot.data ? () => bloc.submitClicks.add(null) : null,
                      child: Text('Submit'),
                    )
                )
            )
          ]
      );

  Widget _overlayLoadingWidget(Widget content) =>
      Stack(
        children: <Widget>[
          content,
          Container(
            color: Colors.black54,
          ),
          Center(child: CircularProgressIndicator()),
        ],
      );
}

class Bloc {
  final StreamController<String> _inputTextChanges = StreamController<String>();
  final StreamController<Null> _submitClicks = StreamController();

  // Inputs
  Sink<String> get inputTextChanges => _inputTextChanges.sink;

  Sink<Null> get submitClicks => _submitClicks.sink;

  // Outputs
  Stream<bool> get submitEnabled =>
      Observable<String>(_inputTextChanges.stream)
          .distinct()
          .map(_isInputValid);

  Stream<bool> get showLoading => _submitClicks.stream.map((_) => true);

  bool _isInputValid(String input) => true;

  void dispose() {
    _inputTextChanges.close();
    _submitClicks.close();
  }
}

Answer 1

As i understand BLoC you should only have one output stream which is connected to a StreamBuilder. This output stream emits a model which contains all required state.

You can see how its done here: https://github.com/ReactiveX/rxdart/blob/master/example/flutter/github_search/lib/github_search_widget.dart

New Link: https://github.com/ReactiveX/rxdart/blob/master/example/flutter/github_search/lib/search_widget.dart

If you need to combine multiple steams to generate you model (sowLoading and submitEnabled), you can use Observable.combineLatest from RxDart to merge multiple streams into one stream. I use this approach and it works really nice.