Flutter中如何用StreamTransformer缓冲多个事件合并输出单个事件？

完全可以通过StreamTransformer实现多个输入事件缓冲合并为单个输出事件，这是StreamTransformer的典型适用场景，核心逻辑是在Transformer内部维护缓冲区，累积收到的字节数据直到匹配到完整的JSON报文再输出。

实现原理 #

1. 在Transformer内部维护一个字节列表作为缓冲区，每次收到蓝牙推送的List<int>数据时，先追加到缓冲区
2. 按你设备的蓝牙通信协议规则，判断缓冲区中是否存在完整的JSON报文
3. 若存在完整报文，截取对应部分解析为Map<String, dynamic>输出，同时删除缓冲区中已处理的部分
4. 剩余未处理的字节保留在缓冲区，等待后续收到新数据后再匹配

代码示例 #

以下是基于分隔符标记报文结束场景的实现示例，可根据实际协议调整判断逻辑，比如改为按报文头长度字段判断：

1. 自定义StreamTransformer #

import 'dart:convert';
import 'package:flutter/foundation.dart';

StreamTransformer<List<int>, Map<String, dynamic>> bleJsonTransformer() {
  // 内部缓冲区，存储未处理的字节数据
  List<int> buffer = [];
  // 报文结束分隔符，需和蓝牙设备端约定一致，此处以换行符为例
  final List<int> separator = utf8.encode('\n');

  return StreamTransformer.fromHandlers(
    handleData: (List<int> data, EventSink<Map<String, dynamic>> sink) {
      // 追加新收到的数据到缓冲区
      buffer.addAll(data);

      int separatorPos;
      // 循环检测缓冲区中是否存在完整报文，支持单次收到多包的场景
      while ((separatorPos = buffer.indexOf(separator)) != -1) {
        // 截取分隔符前的完整报文内容
        final completeJsonBytes = buffer.sublist(0, separatorPos);
        // 移除已处理的报文和分隔符
        buffer = buffer.sublist(separatorPos + separator.length);

        try {
          final jsonStr = utf8.decode(completeJsonBytes);
          final jsonResult = jsonDecode(jsonStr) as Map<String, dynamic>;
          sink.add(jsonResult);
        } catch (parseErr) {
          // 解析错误处理逻辑，可根据需求调整
          if (kDebugMode) {
            print('JSON解析失败: $parseErr, 原始字节: $completeJsonBytes');
          }
        }
      }
    },
    handleDone: (EventSink<Map<String, dynamic>> sink) {
      // 流关闭时处理剩余未处理的缓冲区数据，可根据需求调整
      if (buffer.isNotEmpty && kDebugMode) {
        print('BLE流关闭，剩余未处理字节: $buffer');
      }
      sink.close();
    },
  );
}

2. 调用方式 #

直接把Transformer绑定到flutter_blue的特征值数据流上即可：

// 假设你已经获取到对应的BLE特征值实例 bleChar
bleChar.value
    .transform(bleJsonTransformer())
    .listen((Map<String, dynamic> jsonData) {
      // 此处即可拿到完整解析后的JSON数据
      print('收到完整JSON: $jsonData');
    });

注意事项 #

• 如果蓝牙协议是通过报文头携带总长度的方式标记报文范围，只需要把上述的分隔符检测逻辑替换为「读取报文头长度→判断缓冲区总长度是否达标→截取对应长度解析」即可，整体逻辑完全通用
• 内置的循环检测逻辑已经覆盖了单次收到多个完整JSON的粘包场景，不需要额外处理
• 编码格式需要和设备端保持一致，若设备端用的非UTF-8编码，替换对应的解码方法即可

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