Streambuilder с потоком WebSockets в TabBarView: плохое состояние: поток уже прослушан

Question

Я хочу прослушать поток веб-сокетов внутри виджета, который отображается в TabBarView:

import 'package:flutter/material.dart';
import 'package:web_socket_channel/io.dart';
import 'package:web_socket_channel/web_socket_channel.dart';

class RabitHouse extends StatefulWidget {
  final channel = IOWebSocketChannel.connect('ws://echo.websocket.org');

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

class _RabitHouseState extends State<RabitHouse> {
  @override
  Widget build(BuildContext context) {
    return DefaultTabController(
        length: 2,
        child: Scaffold(
          appBar: AppBar(
            bottom: TabBar(tabs: [
              Tab(
                icon: Icon(
                  Icons.adb,
                ),
              ),
              Tab(
                  icon: Icon(
                Icons.android,
              )),
            ]),
          ),
          body: TabBarView(children: [
            Rabit(channel: widget.channel),
            Rabit(channel: widget.channel),
          ]),
        ));
  }

  @override
  void dispose() {
    widget.channel.sink.close();
    super.dispose();
  }
}

class Rabit extends StatefulWidget {
  final WebSocketChannel channel;

  const Rabit({Key key, this.channel}) : super(key: key);

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

class _RabitState extends State<Rabit> {
  @override
  Widget build(BuildContext context) {
    return StreamBuilder(
      stream: widget.channel.stream,
      builder: (context, snapshot) => Text('pew pew'),
    );
  }
}

Однако, когда я изменяю вкладку, я получаю исключение:

════════ Исключение поймано библиотекой виджетов ═════════════════════════════════════ ══════════════════ Следующее StateError было брошено в здание Rabit (состояние: _RabitState # cd14b): Плохое состояние: поток уже прослушан.

Чего мне не хватает?

Answer 1

Потоки в Dart по умолчанию предназначены для одного слушателя. Это означает, что если вы попытаетесь прослушать их более одного раза, они выдадут ошибку. Если вы хотите прослушать поток в нескольких местах, вам необходимо преобразовать его в поток вещания:

class _RabitHouseState extends State<RabitHouse> {
  Stream broadcastStream;

  @override
  void initState() {
    broadcastStream = widget.channel.stream.asBroadcastStream();
    super.initState();
  }

  @override
  Widget build(BuildContext context) {
    return DefaultTabController(
        length: 2,
        child: Scaffold(
          appBar: AppBar(
            bottom: TabBar(tabs: [
              Tab(
                icon: Icon(
                  Icons.adb,
                ),
              ),
              Tab(
                  icon: Icon(
                Icons.android,
              )),
            ]),
          ),
          body: TabBarView(children: [
            Rabit(stream: broadcastStream),
            Rabit(stream: broadcastStream),
          ]),
        ));
  }

  @override
  void dispose() {
    widget.channel.sink.close();
    super.dispose();
  }
}

class Rabit extends StatefulWidget {
  final Stream stream;

  const Rabit({Key key, this.stream}) : super(key: key);

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

class _RabitState extends State<Rabit> {
  @override
  Widget build(BuildContext context) {
    return StreamBuilder(
      stream: widget.stream,
      builder: (context, snapshot) => Text('pew pew'),
    );
  }
}

Answer 2

Вы можете прослушивать один и тот же поток, если дважды подписываетесь на него. Создание двух каналов WebSocket и последующее прослушивание обоих из них как asBroadcastStream():

class RabitHouse extends StatefulWidget {
  final channel1 = IOWebSocketChannel.connect('ws://echo.websocket.org');
  final channel2 = IOWebSocketChannel.connect('ws://echo.websocket.org');

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

class _RabitHouseState extends State<RabitHouse> {
  Stream stream1;
  Stream stream2;

  @override
  void initState() {
    stream1  = widget.channel1.stream.asBroadcastStream();
    stream2  = widget.channel2.stream.asBroadcastStream();

    super.initState();
  }

  @override
  Widget build(BuildContext context) {
    return DefaultTabController(
        length: 2,
        child: Scaffold(
          appBar: AppBar(
            bottom: TabBar(tabs: [
              Tab(
                icon: Icon(
                  Icons.adb,
                ),
              ),
              Tab(
                  icon: Icon(
                    Icons.android,
                  )
              ),
            ]),
          ),
          body: TabBarView(children: [
            Rabit(stream: stream1),
            Rabit(stream: stream2),
          ]),
        )
    );
  }

  @override
  void dispose() {
    widget.channel1.sink.close();
    widget.channel2.sink.close();
    super.dispose();
  }
}

class Rabit extends StatefulWidget {
  final Stream stream;

  const Rabit({Key key, this.stream}) : super(key: key);

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

class _RabitState extends State<Rabit> {
  @override
  Widget build(BuildContext context) {
    return StreamBuilder(
      stream: widget.stream,
      builder: (context, snapshot) => Text('pew pew'),
    );
  }
}

Если вы хотите узнать больше о потоках, из dart.dev

Существует два вида потоков

Одиночные потоки подписки Самый распространенный вид потока содержит последовательность событий, которые являются частями большего целого. События должны быть доставлены в правильном порядке и не пропустить ни одного из них. Этот тип потока вы получаете, когда читаете файл или получаете веб-запрос.

Такой поток можно прослушать только один раз. Повторное прослушивание позже может означать пропуск первоначальных событий, а затем весь остальной поток не имеет смысла. Когда вы начнете прослушивать, данные будут извлечены и предоставлены кусками.

Потоки широковещания Поток другого типа предназначен для отдельных сообщений, которые могут обрабатываться по одному за раз. Этот вид потока можно использовать, например, для событий мыши в браузере.

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