writing/reading number of bytes to/from socket data in Dart

Question

In the following server/client socket code, the server tries to send ints from 0 to 1023. The client receives the right number of ints but they wrap at 255. Why is this and what is the proper way?

* Server *

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}');
  });
}

* Client *

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');
}

Also, it feels like the Socket api is a bit different - being more Dartish than other languages the way it supports reading as a stream. But what if you just want to read N bytes? How do you achieve that? I see the take method (Stream take(int count)) but I'm not clear on what exactly count is taking, especially in terms of how many bytes have been read. Because Socket implements Stream< List < int > > and because int is not really a fixed size entity (I think??), how do you know how many bytes have been read?

Answer 1

One thing to make clear is that even though most methods in dart:io operate on List<int> the underlying assumption for all IO in Dart is that these lists are list of byte values. The documentation it probably not that precise about that.

This means that you can use whatever List implementation you like when sending bytes - including typed data, e.g.

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)

When receiving data from a socket the data delivered on the Socket stream are the byte-chunks as delivered from the OS. You can make no predictions of how many bytes are received at the time. If you want to be able to read a specific number of bytes one option is to implement a byte reader which have a method like this:

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

This byte reader can then be initialized with a stream of List<int> and handle the framing. What you could do is have a buffer (maybe use BytesBuilder from dart:io) which collect data as it is received. When read is called the Future completed when enough bytes are available. Such a byte reader can also pause the stream when the amount of buffered data reaches a certain limit and resume it again when data is read.

Answer 2

This worked for me, I'm not sure this is the most elegant solution

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');
}

Original attempt:

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');
}

If you don't need the range of an Int64 you can also use Int32, Int16 or some of the UIntXX types or whatever fits your value range best, to save some redundantly transmitted 0-bytes.

I have the suspicion that using a converter like it's explained here https://www.dartlang.org/articles/converters-and-codecs/ could be used to do it more elegantly. This seems similar https://gist.github.com/xxgreg/9104926 but I have to admit that I'm not able to apply these examples for your use case.

Update

I got it working like shown in the articles linked above

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');
}