在函数之间传递流（flutter_libserialport 数据到 COBS 解码器）

Question

我一直在尝试让 COBS 0.2.0 decodeCOBSStream 功能正常工作。 我正在将来自 PC 蓝牙连接的数据从 Hercules 终端程序（禁用 CR/LF 的十六进制值）发送到通过 FTDI 电缆连接到 USB 端口的 HC-05 蓝牙模块。 flutter_libserialport 可以很好地处理原始值，我可以将它们打印到调试控制台，一切看起来都很好。

但是，当我尝试将流传递到 decodeCOBSStream 并将其打印出来时，第一条消息打印得很好。 有时第二个也是如此。 然后它停止打印消息，最终 flutter_libserialport 的 read 例程在 return 语句处抛出异常。

ArgumentError（参数无效：长度必须在 [0, 4611686018427387903] 范围内。）

这是 flutter_libserialport 中的相关函数：

static Uint8List read(int bytes, UtilFunc<ffi.Uint8> readFunc) {
    return ffi.using((arena) {
      final ptr = arena<ffi.Uint8>(bytes);
      final len = call(() => readFunc(ptr));
      return Uint8List.fromList(ptr.asTypedList(len));
    });
  }

这是我的代码。 如果我创建一个列表器，将来自 upcomingData 的值发送到 onDataReceived （将其更改为采用 Uint8List 作为输入），那就没问题了。 我很确定这就是我在其之间放置 decodeCOBSStream 的方式。 flutter_libserialport 适用于 Uint8List，decodeCOBSStream 适用于 ByteData。

import 'dart:typed_data';
import 'dart:async';
import 'package:flutter/material.dart';
import 'package:flutter_libserialport/flutter_libserialport.dart';
import 'package:cobs2/cobs2.dart';

const kDebugMode = true;

void main() {
  runApp(const MyApp());
}

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

  @override
  State<MyApp> createState() => _MyAppState();
}

class _MyAppState extends State<MyApp> {
  @override
  Widget build(BuildContext context) {
    List<String> availablePort = SerialPort.availablePorts;
    SerialPort port = SerialPort('COM4');
    SerialPortReader reader = SerialPortReader(port);
    final controller = StreamController<ByteData>();

    debugPrint('Available Ports: $availablePort');

/*    
    Stream<ByteData> upcomingData = reader.stream.map((data) {
      ByteData byteData = data.buffer.asByteData();
      return byteData;
    });
*/
    // Received the serial data from the SerialPortReader 'reader'
    Stream<Uint8List> upcomingData = reader.stream.map((data) {
      return data;
    });

    controller.addStream(decodeCOBSStream(upcomingData.byteData));

    // When data is received
    void onDataReceived(ByteData data) {
      var decodeUint8 = data.buffer.asUint8List();
      var decodedPrt =
          getHexString(decodeUint8, offset: 0, length: decodeUint8.length);
      debugPrint('Decoded: $decodedPrt');
    }

/*    // When data is received
    void onDataReceived(Uint8List data) {
      var decodedPrt = getHexString(data, offset: 0, length: data.length);
      debugPrint('Decoded: $decodedPrt');
    }
*/
    try {
      port.openReadWrite();
      initSerial(port);

      // Send data to serial port
      serialSend(port, 'Ready for data!');

      // Creates a listener to receive the decoded data
      //controller.stream.listen((data) => onDataReceived(data));
      controller.stream.listen(onDataReceived).onDone(() {});
      //upcomingData.listen(onDataReceived).onDone(() {});
    } on SerialPortError catch (err, _) {
      debugPrint('${SerialPort.lastError}');
      port.close();
    }

    return Container();
  }
}

extension on Stream<Uint8List> {
  Stream<ByteData> get byteData => transform(Uint8ListToByteData());
}

// Stream transformer that converts Uint8List data to ByteData.
class Uint8ListToByteData extends StreamTransformerBase<Uint8List, ByteData> {
  @override
  Stream<ByteData> bind(Stream<Uint8List> stream) {
    return stream.map((data) => data.buffer.asByteData());
  }
}

Uint8List _stringToUint8List(String data) {
  List<int> codeUnits = data.codeUnits;
  Uint8List uint8list = Uint8List.fromList(codeUnits);
  return uint8list;
}

// Converts a String of multiple integers into HEX
String getHexString(
  List<int> list, {
  required int offset,
  required int length,
}) {
  var sublist = list.getRange(offset, offset + length);
  return [
    for (var byte in sublist)
      byte.toRadixString(16).padLeft(2, '0').toUpperCase()
  ].join();
}

void initSerial(SerialPort port) {
  var config = port.config;
  config.baudRate = 9600;
  config.bits = 8;
  config.stopBits = 1;
  config.parity = 0;
  port.config = config;

  debugPrint('BaudRate = 9600');
}

void serialSend(SerialPort port, String data) {
  debugPrint('Written Bytes: ${port.write(_stringToUint8List(data))}');
}

我尝试以不同的方式将数据传递到flutter_libserialport。 我尝试在 upcomingData 函数中将其从 Uint8List 更改为 ByteData，并且我尝试使用流转换器来完成此操作。 显示了我尝试过的一些变体，但已注释掉。

Answer 1

ByteDataReader

是一个有用的类，用于收集您将逐步处理的字节。这允许您在字节通过网络传输时添加字节。如果没有足够的字节来形成完整的消息，它还有另一个“阻止”功能 - 即在您的情况下，如果还没有零字节分隔符。

这使我们能够采用维基百科示例 C 代码，并快速实现一个

async

方法，一旦有可用消息，该方法就会返回完整的消息。随着字节的添加，将其包装在

async*

方法中会产生此类消息流。

完整的工作示例如下所示：

import 'dart:typed_data';

import 'package:buffer/buffer.dart';
import 'package:convert/convert.dart';

void main() async {
  final reader = ByteDataReader(copy: true);
  cobsDecode(reader).listen((o) => print(o));

  reader.add(hex.decode('01010100'));
  reader.add(hex.decode('031122023300'));
  reader.add(hex.decode('031122')); // the above block split in 2
  reader.add(hex.decode('023300'));

  // add some more bytes later
  Future.delayed(
    Duration(seconds: 2),
    () => reader.add(hex.decode('031122023300')),
  );
}

Stream<Uint8List> cobsDecode(ByteDataReader reader) async* {
  while (true) {
    yield await _readBlock(reader);
  }
}

Future<Uint8List> _readBlock(ByteDataReader reader) async {
  final output = BytesBuilder();
  var code = 0xff;
  var block = 0;
  while (true) {
    await reader.readAhead(1); // ensure at least one byte available
    if (block != 0) {
      output.addByte(reader.readUint8()); // todo - add non-zero check
      block--;
    } else {
      block = reader.readUint8();
      if (block != 0 && code != 0xff) {
        output.addByte(0);
      }
      code = block;
      if (code == 0) {
        break;
      }
      block--;
    }
  }
  return output.toBytes();
}

请注意，这个简单的示例没有一些典型的错误处理，例如检查块内是否有零。