Flutter中StreamBuilder获取API JSON数据时snapshot.data为空的问题

使用StreamBuilder从API获取JSON数据时，snapshot.data始终为空。该接口在FutureBuilder中可以正常运行，但因需要持续获取数据，必须使用StreamBuilder。相关代码及API返回数据如下：

当前Stream实现 #

final Stream<Notifications> _fetchDataStream = (() {
  late final StreamController<Notifications> controller;
  controller = StreamController<Notifications>(
    onListen: () async {
      final request = http.StreamedRequest(
          'GET',
          Uri.https('ntfy.sh', '/mytopic/json',
              {'poll': '1'}));

      request.sink.close();
      var result = await request.send();
      var response = await http.Response.fromStream(result);
      Notifications nots = Notifications();
      if (response.statusCode == 200) {
        print(response.body);
        Notifications.fromJson(jsonDecode(response.body));
      } else {
        throw Exception('Failed to load album');
      }
      controller.add(Notifications());
      await controller.close();
    },
  );
  return controller.stream;
})();

当前StreamBuilder实现 #

StreamBuilder<Notifications>(
            stream: _fetchDataStream,
            builder:
                (BuildContext context, AsyncSnapshot<Notifications> snapshot) {
              if (snapshot.hasData) {
                return Text('${snapshot.data}');
              } else {
                return CircularProgressIndicator();
              }
            },
          ),

模型类（modal.dart） #

class Notifications {
  String? id;
  int? time;
  int? expires;
  String? event;
  String? topic;
  String? title;
  String? message;

  Notifications(
      {this.id,
      this.time,
      this.expires,
      this.event,
      this.topic,
      this.title,
      this.message});

  Notifications.fromJson(Map<String, dynamic> json) {
    id = json['id'];
    time = json['time'];
    expires = json['expires'];
    event = json['event'];
    topic = json['topic'];
    title = json['title'];
    message = json['message'];
  }

  Map<String, dynamic> toJson() {
    final Map<String, dynamic> data = new Map<String, dynamic>();
    data['id'] = this.id;
    data['time'] = this.time;
    data['expires'] = this.expires;
    data['event'] = this.event;
    data['topic'] = this.topic;
    data['title'] = this.title;
    data['message'] = this.message;
    return data;
  }
}

API返回的JSON数据 #

{"id":"coOhkyN1YxBD","time":1692893929,"expires":1698077929,"event":"message","topic":"mytopic","message":"This is a test notification from the ntfy Android app. It has a level 5 priority. If you send another, it may look different.","priority":5,"tags":["this-is-a-tag","rsync"]}
{"id":"0snbxhbGEehW","time":1697330037,"expires":1697373237,"event":"message","topic":"mytopic","message":"Backup successful 😀"}
{"id":"PvcsTrRxqb0m","time":1697352352,"expires":1697395552,"event":"message","topic":"mytopic","message":"Backup successful 😀"}
{"id":"ZDGew6HgIbue","time":1697352368,"expires":1697395568,"event":"message","topic":"mytopic","message":"Bing Chilling"}
{"id":"I20EDvVvwgMr","time":1697352425,"expires":1697395625,"event":"message","topic":"mytopic","message":"lsd done running"}
{"id":"GX02kf13Y0Xd","time":1697352645,"expires":1697395845,"event":"message","topic":"mytopic","message":"Hello Motu"}
{"id":"nN6Jnsrf5jyI","time":1697355413,"expires":1697398613,"event":"message","topic":"mytopic","message":"Backup successful 😀"}
{"id":"83L6jGQlSfYk","time":1697360074,"expires":1697403274,"event":"message","topic":"mytopic","message":"Backup successful 😀"}
{"id":"Dnj8RmVCFEwr","time":1697361179,"expires":1697404379,"event":"message","topic":"mytopic","message":"Would You Kill the Fat Man? [pdf] (utilitarianism.com) book"}
{"id":"QIzLhHdI5mww","time":1697362404,"expires":1697405604,"event":"message","topic":"mytopic","message":"card+scanned"}

关键错误分析 #

1. 添加空对象到流：代码中controller.add(Notifications());添加的是空白实例，而非解析API返回后的对象；Notifications.fromJson(jsonDecode(response.body));仅调用构造函数但未赋值，完全无效。
2. 未处理多行JSON：API返回每行一个独立JSON对象，直接整体解析会报错，需逐行处理。
3. 未流式处理响应：http.Response.fromStream(result)一次性读取全部响应，无法实现持续获取数据的需求，应直接监听字节流逐行解析。
4. 流过早关闭：await controller.close();在第一次获取数据后就关闭流，无法接收后续推送的新数据。

修正后的代码 #

1. 优化模型类（改为单数类名更合理，工厂构造函数更规范） #

class Notification {
  String? id;
  int? time;
  int? expires;
  String? event;
  String? topic;
  String? title;
  String? message;

  Notification({
    this.id,
    this.time,
    this.expires,
    this.event,
    this.topic,
    this.title,
    this.message,
  });

  factory Notification.fromJson(Map<String, dynamic> json) {
    return Notification(
      id: json['id'],
      time: json['time'],
      expires: json['expires'],
      event: json['event'],
      topic: json['topic'],
      title: json['title'],
      message: json['message'],
    );
  }

  Map<String, dynamic> toJson() {
    final Map<String, dynamic> data = {};
    data['id'] = id;
    data['time'] = time;
    data['expires'] = expires;
    data['event'] = event;
    data['topic'] = topic;
    data['title'] = title;
    data['message'] = message;
    return data;
  }

  @override
  String toString() {
    return 'Notification(id: $id, message: $message)';
  }
}

2. 修正Stream实现（支持持续获取数据） #

final Stream<Notification> _fetchDataStream = (() {
  late final StreamController<Notification> controller;
  controller = StreamController<Notification>(
    onListen: () async {
      try {
        final request = http.StreamedRequest(
          'GET',
          Uri.https('ntfy.sh', '/mytopic/json', {'poll': '1'}),
        );
        request.sink.close();
        
        final response = await request.send();
        if (response.statusCode != 200) {
          throw Exception('Failed to connect: ${response.statusCode}');
        }

        // 流式读取响应，按行解析JSON
        final stream = response.stream
            .transform(utf8.decoder)
            .transform(const LineSplitter());

        await for (final line in stream) {
          if (line.trim().isEmpty) continue;
          try {
            final jsonData = jsonDecode(line) as Map<String, dynamic>;
            final notification = Notification.fromJson(jsonData);
            controller.add(notification);
          } catch (e) {
            print('Failed to parse line: $e');
          }
        }
      } catch (e) {
        controller.addError(e);
      } finally {
        if (!controller.isClosed) {
          await controller.close();
        }
      }
    },
    onCancel: () {
      if (!controller.isClosed) {
        controller.close();
      }
    },
  );
  return controller.stream;
})();

3. 修正StreamBuilder（展示通知列表） #

// 建议使用StatefulWidget存储历史通知，这里示例仅展示最新数据
StreamBuilder<Notification>(
  stream: _fetchDataStream,
  builder: (context, snapshot) {
    if (snapshot.hasError) {
      return Text('Error: ${snapshot.error}');
    }

    if (snapshot.connectionState == ConnectionState.waiting) {
      return const CircularProgressIndicator();
    }

    final notifications = <Notification>[];
    if (snapshot.hasData) {
      notifications.add(snapshot.data!);
      // 实际开发中，建议用State管理列表，追加新数据
    }

    return ListView.builder(
      itemCount: notifications.length,
      itemBuilder: (context, index) {
        final notification = notifications[index];
        return ListTile(
          title: Text(notification.title ?? 'No Title'),
          subtitle: Text(notification.message ?? 'No Message'),
          trailing: Text(
            DateTime.fromMillisecondsSinceEpoch(notification.time! * 1000)
                .toString()
                .substring(0, 19),
          ),
        );
      },
    );
  },
)

说明 #

• 通过LineSplitter分割响应流，逐行解析JSON，实现持续接收新通知的长连接需求。
• 流控制器仅在连接断开、出错或取消订阅时关闭，保持长连接状态。
• 实际开发中建议使用StatefulWidget维护通知列表，确保新数据到来时能追加并更新UI。

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