在Dart中向/从套接字数据写入/读取字节数。

Question

在以下的服务器/客户端套接字代码中，服务器尝试发送从0到1023的整数。客户端接收到了正确数量的整数，但它们在255处截断了。为什么会这样，有什么正确的方法吗？

* 服务器 *

import 'dart:async';
import 'dart:io';

main() async {
  final server = await ServerSocket.bind('127.0.0.1', 4041);
  server.listen((Socket socket) {
    print('Got connected ${socket.remoteAddress}');
    for(int i=0; i<1024; i++) {
      socket.add([i]);
    }
    socket.close();
    print('Closed ${socket.remoteAddress}');
  });
}

* 客户端 *

import 'dart:async';
import 'dart:io';

main() async {
  final client = await Socket.connect('127.0.0.1', 4041);
  client.listen(
      (var data) => print('Got $data'),
      onDone: () { print('Done'); client.close(); },
      onError: (e) { print('Got error $e'); client.close(); });
  print('main done');
}

此外，感觉 Socket api 与其他语言不同，更具有 Dart 特色，支持作为流进行读取。但是，如果您只想读取 N 个字节，该怎么做？我看到了 take 方法（Stream take(int count)），但我不清楚 count 到底取了多少个字节，特别是已经读取了多少个字节。因为 Socket 实现了 Stream< List < int > > ，而 int 并不是一个固定大小的实体（我认为是这样的？），你如何知道已经读取了多少个字节？

- user1338952

Answer 1

需要澄清的一件事是，尽管dart:io中的大多数方法都是针对List<int>进行操作，但在Dart中所有IO的基本假设是这些列表是字节值列表。文档可能没有那么精确。

这意味着，发送字节时可以使用任何您喜欢的List实现，包括类型化数据。

var socket = ...
socket.add([1, 2, 3]);
socket.add(new List<int>.generate(3, (int index) => index * index);
socket.add('ASCII'.codeUnits);
socket.add(UTF8.encode('Søren'));

var typedData = new Uint8List(8);
var data = new ByteData.view(typedData.buffer);
data.setInt32(0, 256 * 256 - 1);
data.setInt32(4, 256 * 256 - 1, Endianness.LITTLE_ENDIAN);
socket.add(typedData)

当从套接字接收数据时，Socket 流中传递的数据是从操作系统传递的字节块。您无法预测每次接收多少字节。如果您想能够读取特定数量的字节，则可以实现一个字节阅读器，其具有以下方法：

Future<List<int>> read(int numBytes);

这个字节读取器可以用于处理 List<int> 流并处理框架。你可以使用一个缓冲区（也许使用来自 dart:io 的 BytesBuilder），该缓冲区会在接收到数据时收集它。当调用 read 时，如果有足够的字节可用，则 Future 完成。这样的字节读取器还可以在缓冲数据量达到一定限制时暂停流，并在读取数据时恢复。

- sgjesse

Answer 2

这对我有用，但我不确定这是最优雅的解决方案。

import 'dart:io';
import 'dart:async';
import 'dart:typed_data';

main() async {
  final server = await ServerSocket.bind('127.0.0.1', 4041);
  server.listen((Socket socket) {
    print('Got connected ${socket.remoteAddress}');

    var toByte = new StreamTransformer<List<int>, Uint8List>.fromHandlers(
        handleData: (data, sink) {
      sink.add(new Uint64List.fromList(data).buffer.asUint8List());
    });

    var streamController = new StreamController<List<int>>();
    streamController.stream.transform(toByte).pipe(socket);

    for (int i = 0; i < 1024; i++) {
      streamController.add([i]);
    }
    streamController.close();
    print('Closed ${socket.remoteAddress}');
  });
}

import 'dart:io';
import 'dart:async';

main() async {
  final Socket client = await Socket.connect('127.0.0.1', 4041);

  var fromByte = new StreamTransformer<List<int>, List<int>>.fromHandlers(
      handleData: (data, sink) {
    sink.add(data.buffer.asInt64List());
  });

  client.transform(fromByte).listen((e) => e.forEach(print));
  print('main done');
}

原始尝试：

import 'dart:io';
import 'dart:typed_data';

main() async {
  final server = await ServerSocket.bind('127.0.0.1', 4041);
  server.listen((Socket socket) {
    print('Got connected ${socket.remoteAddress}');
    for(int i=0; i<1024; i++) {
      socket.add(new Uint64List.fromList([i]).buffer.asUint8List());
    }
    socket.close();
    print('Closed ${socket.remoteAddress}');
  });
}

import 'dart:io';
import 'dart:typed_data';

main() async {
  final Socket client = await Socket.connect('127.0.0.1', 4041);
  client.listen(
      (var data) {
        var ints = new Uint8List.fromList(data).buffer.asInt64List();
        ints.forEach((i) => print('Got $i'));
      },
      onDone: () { print('Done'); client.close(); },
      onError: (e) { print('Got error $e'); client.close(); });
  print('main done');
}

如果你不需要Int64的范围，也可以使用Int32、Int16或一些UIntXX类型或适合你值范围最好的类型，以节省一些冗余传输的0字节。

我怀疑使用像这里https://www.dartlang.org/articles/converters-and-codecs/解释的转换器可能更优雅地实现它。这似乎类似于https://gist.github.com/xxgreg/9104926，但我必须承认我无法将这些示例应用于您的用例。

更新

我已经按上面链接的文章所示使其工作。

import 'dart:convert';
import 'dart:typed_data';

class IntConverter extends Converter<List<int>, List<int>> {
  const IntConverter();

  List<int> convert(List<int> data) {
    if (data is Uint8List) {
      return data.buffer.asInt64List();
    } else {
      return new Uint64List.fromList(data).buffer.asUint8List();
    }
  }

  IntSink startChunkedConversion(sink) {
    return new IntSink(sink);
  }
}

class IntSink extends ChunkedConversionSink<List<int>> {
  final _converter;
  // fales when this type is used
  // final ChunkedConversionSink<List<int>> _outSink;
  final _outSink;

  IntSink(this._outSink) : _converter = new IntConverter();

  void add(List<int> data) {
    _outSink.add(_converter.convert(data));
  }

  void close() {
    _outSink.close();
  }
}

import 'dart:io';
import 'dart:typed_data';

main() async {
  final server = await ServerSocket.bind('127.0.0.1', 4041);
  server.listen((Socket socket) {
    print('Got connected ${socket.remoteAddress}');

    for (int i = 0; i < 1024; i++) {
      socket.add(new Uint64List.fromList([i]).buffer.asUint8List());
    }
    socket.close();
    print('Closed ${socket.remoteAddress}');
  });
}

import 'dart:io';
import 'byte_converter.dart';

main() async {
  final Socket client = await Socket.connect('127.0.0.1', 4041);
  client.transform(new IntConverter()).listen((e) => e.forEach(print));
  print('main done');
}

- Günter Zöchbauer

---

谢谢Gunter。我想一个基本的后续问题是为什么socket实现像*Stream<List<int>>这样的东西，因为不清楚正在发生什么，因为int有一个层次结构。但Stream<ByteArray>*很明显是让更高的层来解释一些字节的数量。此外，如何从套接字中读取n个字节？它似乎具有内在的缓冲区，并且从日志中我看到数据中元素的数量是不确定的。 - user1338952

Uint8List 在dart:typed_data中，我想您不希望在dart:io中引入依赖项。如果Dart完全实现typedef，则可以“typedef byte int;”，然后它将成为Stream<List<byte>>。我有点喜欢流和列表的可组合性，所以我不确定是否喜欢“typedef ByteStream Stream<List<int>>;”，这是另一种执行相同操作的方法。 - Greg Lowe

@GregLowe 我不确定我完全理解你的问题。这取决于你使用的类型。如果我使用Uint32List，它会每个int插入4个字节而不是8个。 - Günter Zöchbauer

数字被转换为二进制并通过网络发送。网络层将二进制拆分为任意大小的数据包，然后接收软件从网络层获取数据包，其大小取决于操作系统的网络实现。如果网络决定将数据切成大小为1500字节的数据包，而这不是8的倍数会发生什么？您将在一个数据包中获得int64的一半，然后在下一个数据包中获得另一半。您的IntConverter类无法正确处理此问题。 - Greg Lowe

好的，我明白你的意思。 - Günter Zöchbauer