如何结合StreamBuilder使用realtime_dart包的Socket hooks（适配Supabase）

要将realtime_dart的回调式API和Flutter的StreamBuilder配合，核心是把实时事件回调转换成标准的Dart Stream，具体实现如下：

完整实现示例 #

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

class RealtimeDataPage extends StatefulWidget {
  const RealtimeDataPage({super.key});

  @override
  State<RealtimeDataPage> createState() => _RealtimeDataPageState();
}

class _RealtimeDataPageState extends State<RealtimeDataPage> {
  // 用于转发实时事件的StreamController
  late final StreamController<Map<String, dynamic>> _realtimeController;
  // realtime相关实例
  late final RealtimeClient _socket;
  late final RealtimeChannel _channel;
  // 用来存储当前的全量数据，可根据业务需求调整
  final List<Map<String, dynamic>> _currentData = [];

  @override
  void initState() {
    super.initState();
    _initRealtime();
  }

  void _initRealtime() {
    // 需多个监听者可使用broadcast创建多订阅流
    _realtimeController = StreamController.broadcast();

    // 初始化Realtime客户端，替换为自己的Supabase配置
    _socket = RealtimeClient(
      'ws://你的SUPABASE_API_ENDPOINT/realtime/v1',
      params: {'apikey': '你的SUPABSE_API_KEY'},
    );

    // 监听public schema的实时事件，也可指定表名，比如'realtime:public:users'只监听users表
    _channel = _socket.channel('realtime:public');

    // 监听插入事件，转发payload到stream
    _channel.on('INSERT', (payload, {ref}) {
      if (payload != null) {
        _realtimeController.sink.add({
          'event_type': 'INSERT',
          'data': payload,
        });
      }
    });

    // 监听删除事件
    _channel.on('DELETE', (payload, {ref}) {
      if (payload != null) {
        _realtimeController.sink.add({
          'event_type': 'DELETE',
          'data': payload,
        });
      }
    });

    // 监听更新事件
    _channel.on('UPDATE', (payload, {ref}) {
      if (payload != null) {
        _realtimeController.sink.add({
          'event_type': 'UPDATE',
          'data': payload,
        });
      }
    });

    // 连接socket并订阅频道
    _socket.connect();
    _channel.subscribe().receive('error', (error) {
      // 错误转发到stream，方便StreamBuilder处理
      _realtimeController.addError(error ?? '订阅失败');
    });
  }

  @override
  void dispose() {
    // 释放资源，避免内存泄漏
    _channel.unsubscribe();
    _socket.disconnect();
    _realtimeController.close();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('实时数据')),
      body: StreamBuilder<Map<String, dynamic>>(
        stream: _realtimeController.stream,
        // 初始数据可替换为首次加载的存量业务数据
        initialData: const {'event_type': 'INIT', 'data': []},
        builder: (context, snapshot) {
          // 处理错误状态
          if (snapshot.hasError) {
            return Center(child: Text('加载失败：${snapshot.error}'));
          }
          // 处理加载中状态
          if (!snapshot.hasData) {
            return const Center(child: CircularProgressIndicator());
          }

          // 处理不同事件类型，更新本地数据
          final event = snapshot.data!;
          final eventType = event['event_type'];
          final payload = event['data'];

          switch(eventType) {
            case 'INIT':
              // 初始状态可在此加载历史存量数据
              break;
            case 'INSERT':
              _currentData.add(payload['record']);
              break;
            case 'DELETE':
              _currentData.removeWhere((item) => item['id'] == payload['old_record']['id']);
              break;
            case 'UPDATE':
              final index = _currentData.indexWhere((item) => item['id'] == payload['record']['id']);
              if (index != -1) {
                _currentData[index] = payload['record'];
              }
              break;
          }

          // 渲染数据列表
          return ListView.builder(
            itemCount: _currentData.length,
            itemBuilder: (context, index) {
              final item = _currentData[index];
              return ListTile(
                title: Text(item['title'] ?? '无标题'),
                subtitle: Text(item['content'] ?? '无内容'),
              );
            },
          );
        },
      ),
    );
  }
}

注意事项 #

• 必须使用StatefulWidget管理生命周期，在dispose阶段释放所有资源，避免内存泄漏和无效socket连接占用资源。
• 可根据业务需求调整channel的监听范围，比如指定表名、行级过滤条件，减少不必要的事件推送。
• 示例中的_currentData是本地维护的全量数据列表，可根据自身业务逻辑调整数据更新规则。

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