在 TabBarView 中使用 WebSockets 流的 Streambuilder：错误状态：流已被监听

Question

我想收听一个显示在 TabBarView 中的 Widget 内的 websocket 流：

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

但是，当我更改选项卡时，我得到了一个例外：

════════ 小部件库捕获的异常════════════════════════════════ ═════════════════════ 在构建 Rabit（状态：_RabitState#cd14b）时引发了以下 StateError： 错误状态：Stream 已被收听。

我错过了什么？

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

有两种流

单一订阅流 最常见的流类型包含 作为更大整体的一部分的一系列事件。活动需要 以正确的顺序交付，没有遗漏任何一个。这 是您在阅读文件或接收网页时获得的那种流 请求。

这样的流只能听一次。稍后再听 可能意味着错过最初的事件，然后是其余的 流没有意义。当您开始收听时，数据将是 以块的形式获取和提供。

广播流 另一种流是供个人使用的 一次可以处理一个消息。这种流可以 例如，用于浏览器中的鼠标事件。

您可以随时开始收听这样的流，并且您会得到 在您收听时触发的事件。不止一位听众可以 同时收听，以后可以再收听 取消之前的订阅。