Add readMessageAsync(…), tryReadMessageAsync(…), AsyncBufferedReader

Author: JonasWanke

capnproto/lib/capnproto.dart

@@ -1,4 +1,5 @@
 export 'src/any_pointer.dart';
+export 'src/async_buffered_reader.dart';
 export 'src/constants.dart';
 export 'src/data_list.dart';
 export 'src/error.dart';

capnproto/lib/src/async_buffered_reader.dart

@@ -0,0 +1,82 @@
+import 'dart:async';
+import 'dart:collection';
+import 'dart:math' as math;
+import 'dart:typed_data';
+
+import 'package:meta/meta.dart';
+import 'package:oxidized/oxidized.dart';
+
+import 'error.dart';
+
+class AsyncBufferedReader {
+  AsyncBufferedReader(Stream<List<int>> stream) {
+    _subscription = stream.listen(
+      (event) {
+        _pendingData.add(event);
+        _notifyNewEvent();
+      },
+      onDone: _notifyNewEvent,
+    );
+  }
+
+  late final StreamSubscription<List<int>> _subscription;
+  final _pendingData = Queue<List<int>>();
+  var _offsetInFirstBuffer = 0;
+
+  var _nextEvent = Completer<void>();
+  void _notifyNewEvent() {
+    final nextEvent = _nextEvent;
+    _nextEvent = Completer();
+    nextEvent.complete();
+  }
+
+  @useResult
+  Future<int> read(Uint8List buffer) async {
+    if (buffer.isEmpty) return 0;
+
+    var didResumeSubscription = false;
+    var offset = 0;
+    while (offset < buffer.length) {
+      if (_pendingData.isEmpty) {
+        if (!didResumeSubscription) {
+          didResumeSubscription = true;
+          _subscription.resume();
+        }
+
+        await _nextEvent.future;
+        if (_pendingData.isEmpty) break;
+      }
+
+      final newData = _pendingData.first;
+      final length = math.min(
+        buffer.length - offset,
+        newData.length - _offsetInFirstBuffer,
+      );
+      buffer.setRange(
+        offset,
+        offset + length,
+        newData.skip(_offsetInFirstBuffer),
+      );
+      offset += length;
+      if (_offsetInFirstBuffer + length == newData.length) {
+        _pendingData.removeFirst();
+        _offsetInFirstBuffer = 0;
+      }
+    }
+    _subscription.pause();
+    return offset;
+  }
+
+  @useResult
+  Future<Result<void, PrematureEndOfInputCapnpError>> readExact(
+    Uint8List buffer,
+  ) async {
+    final length = await read(buffer);
+    if (length < buffer.length) {
+      return const Err(PrematureEndOfInputCapnpError());
+    }
+    return const Ok(null);
+  }
+
+  Future<void> cancel() => _subscription.cancel();
+}

capnproto/lib/src/error.dart

@@ -20,8 +20,8 @@ sealed class CapnpError {
   const CapnpError();
 }
 
-class PrematureEndOfFileCapnpError extends CapnpError {
-  const PrematureEndOfFileCapnpError();
+class PrematureEndOfInputCapnpError extends CapnpError {
+  const PrematureEndOfInputCapnpError();
 }
 
 class InvalidNumberOfSegmentsCapnpError extends CapnpError {

capnproto/lib/src/serialize.dart

@@ -3,6 +3,7 @@ import 'dart:typed_data';
 import 'package:meta/meta.dart';
 import 'package:oxidized/oxidized.dart';
 
+import 'async_buffered_reader.dart';
 import 'constants.dart';
 import 'error.dart';
 import 'message.dart';
@@ -76,6 +77,74 @@ CapnpResult<MessageReader?> tryReadMessage(
   );
 }
 
+/// Reads a serialized message (including a segment table) from [reader].
+///
+/// [reader] is allowed to extend beyond the end of the message.
+///
+/// The segment table format for streams is defined in the Cap'n Proto
+/// [encoding spec](https://capnproto.org/encoding.html).
+Future<CapnpResult<MessageReader>> readMessageAsync(
+  AsyncBufferedReader reader, {
+  ReaderOptions options = const ReaderOptions(),
+}) async {
+  switch (await tryReadMessageAsync(reader, options: options)) {
+    case Ok(value: null):
+      return const Err(PrematureEndOfInputCapnpError());
+    case Ok(:final value?):
+      return Ok(value);
+    case Err(:final error):
+      return Err(error);
+  }
+}
+
+/// Like [readMessageAsync], but returns `Ok(null)` instead of an error if the
+/// [reader] is empty.
+Future<CapnpResult<MessageReader?>> tryReadMessageAsync(
+  AsyncBufferedReader reader, {
+  ReaderOptions options = const ReaderOptions(),
+}) async {
+  final segmentCountBuffer = Uint8List(4);
+  final readBytes = await reader.read(segmentCountBuffer);
+  if (readBytes == 0) return const Ok(null);
+  if (readBytes < 4) return const Err(PrematureEndOfInputCapnpError());
+
+  final int segmentCount;
+  switch (segmentCountBuffer.asByteData.getSegmentCount()) {
+    case Ok(:final value):
+      segmentCount = value;
+    case Err(:final error):
+      return Err(error);
+  }
+
+  final segmentLengthsBuffer =
+      Uint8List(4 * (segmentCount.isEven ? segmentCount + 1 : segmentCount));
+  if (await reader.readExact(segmentCountBuffer) case Err(:final error)) {
+    return Err(error);
+  }
+
+  final SegmentLengthsBuilder builder;
+  switch (segmentLengthsBuffer.asByteData
+      .getSegmentLengths(segmentCount, options)) {
+    case Ok(:final value):
+      builder = value.builder;
+    case Err(:final error):
+      return Err(error);
+  }
+
+  final dataBuffer =
+      Uint8List(builder.totalWords * CapnpConstants.bytesPerWord);
+  if (await reader.readExact(dataBuffer) case Err(:final error)) {
+    return Err(error);
+  }
+
+  return Ok(
+    MessageReader(
+      builder.intoSegments(dataBuffer.asByteData),
+      options: options,
+    ),
+  );
+}
+
 extension on ByteData {
   CapnpResult<int> getSegmentCount() {
     final segmentCount = getUint32(0, Endian.little) + 1;

capnproto/lib/src/utils.dart

@@ -72,3 +72,7 @@ extension ByteDataCapnp on ByteData {
         );
   }
 }
+
+extension Uint8ListCapnp on Uint8List {
+  ByteData get asByteData => buffer.asByteData(offsetInBytes, lengthInBytes);
+}