Flutter GetX WebSocket 数据流传递到Provider与Controller的故障排查

问题背景 #

需求说明 #

• Flutter应用通过WebSockets调用GraphQL API
• 应用视图层调用Controller，Controller调用Provider，Provider通过WebSockets或者HTTP API套接字调用AWS AppSync API
• 后端会通过AppSync API或HTTP API套接字的WebSockets通道不定期推送数据流
• 数据流需要按顺序回传给Provider，再传递到Controller
• 由Controller而非Provider更新obs响应式变量，驱动UI同步变更

问题表现 #

调用端通过WebSockets成功接收数据，但数据流未回传给Provider和Controller，无法触发UI更新。

根本原因 #

核心问题是async*生成器和listen回调混用导致的事件失效：你在listen的回调函数中写的yield完全不生效，就算给回调加了async*修饰，这个内部生成器也没有被任何代码消费，数据根本不会传递到外层返回的Stream中。同时你在async*函数中没有将原始Stream的事件转发到外层返回的Stream，导致上层拿到的Stream没有任何事件输出。

修复方案 #

1. 调用实现层（orderdata.dart） #

不要嵌套listen，直接用await for遍历原始Stream，处理后向外层yield数据：

@override
Stream<dynamic> subscribe({
  String query,
  Map<String, dynamic> variables,
}) async* {
  debugPrint('===->subscribe===');
  // it can be any stream here, http or file or image or media
  final Stream<GraphQLResponse<String>> operation = Amplify.API.subscribe(
    GraphQLRequest<String>(
      document: query,
      variables: variables,
    ),
    onEstablished: () {
      debugPrint(
        '===->subscribe onEstablished ===',
      );
    },
  );

  // 直接遍历Stream事件，处理后yield到外层
  await for (final event in operation) {
    final jsonData = json.decode(event.data.toString());
    debugPrint('===->subscription data $jsonData');
    yield jsonData;
  }
}

2. Provider层（orderprovider.dart） #

同样移除嵌套的listen，直接转发上游Stream的事件，按需做业务实体转换：

Stream<Order> orderSubscription(String placeId) async* {
  debugPrint('===->=== $placeId');
  subscriptionResponseStream = orderData.subscribe(
    query: subscribeToMenuOrder,
    variables: {"place_id": placeId},
  );

  // 遍历上游Stream，转成Order实体后向外层yield
  await for (final event in subscriptionResponseStream) {
    debugPrint(
      "===->=== yielded $event",
    );
    // 此处根据你的业务逻辑把event转换成Order实体
    yield Order.fromJson(event);
  }
  debugPrint('===->=== finished');
}

3. Controller层（homecontroller.dart） #

原有逻辑基本可用，只需要注意订阅的Stream要记得在页面销毁时取消，避免内存泄漏，拿到数据后直接更新你的obs响应式变量即可：

// 建议把Subscription存在成员变量里，方便销毁时取消
StreamSubscription? _orderSub;

Future<void> getSubscriptionData(String placeId) async {
  debugPrint('===HomeController->getSubscriptionData===');
  // 建议OrderProvider用单例，不要每次调用都新建实例
  _orderSub = OrderProvider().orderSubscription(placeId).listen(
        (data) {
          //this block is executed when data event is receivedby listener
          debugPrint('Data: $data');
          // 此处更新你的obs响应式变量即可驱动UI变更
          Get.snackbar('orderSubscription', data.toString());
        },
        onError: (err) {
          //this block is executed when error event is received by listener
          debugPrint('Error: $err');
        },
        cancelOnError:
            false, //this decides if subscription is cancelled on error or not
        onDone: () {
          //this block is executed when done event is received by listener
          debugPrint('Done!');
        },
      );
}

// Controller销毁时取消订阅
@override
void onClose() {
  _orderSub?.cancel();
  super.onClose();
}

内容的提问来源于stack exchange，提问作者raaone7