framer 1.0.1

Incremental framing for Stream<List<int>> with hard progress guarantees, spill/drain-safe codecs, and a fast ring buffer.

example/example.md

example/grpc_messages.dart

import 'dart:math' as math;
import 'dart:typed_data';

import 'package:framer/framer.dart';

Uint8List encodeGrpc(Uint8List payload, {bool compressed = false}) {
  final len = payload.length;
  final header = <int>[
    compressed ? 1 : 0,
    (len >> 24) & 0xFF,
    (len >> 16) & 0xFF,
    (len >> 8) & 0xFF,
    len & 0xFF,
  ];
  return Uint8List.fromList([...header, ...payload]);
}

Stream<Uint8List> chunk(Uint8List bytes,
    {int maxChunk = 9, int seed = 3}) async* {
  final rng = math.Random(seed);
  var off = 0;
  while (off < bytes.length) {
    final n = math.min(bytes.length - off, 1 + rng.nextInt(maxChunk));
    yield Uint8List.sublistView(bytes, off, off + n);
    off += n;
  }
}

Future<void> main() async {
  final m1 = encodeGrpc(Uint8List.fromList([1, 2, 3]), compressed: false);
  final m2 = encodeGrpc(Uint8List.fromList([4, 5]), compressed: true);
  final wire = Uint8List.fromList([...m1, ...m2]);

  await for (final msg in chunk(wire).decodeFrames(GrpcMessageCodec())) {
    print(msg);
  }
}

example/lsp_content_length.dart

import 'dart:convert';
import 'dart:math' as math;
import 'dart:typed_data';

import 'package:framer/framer.dart';

Uint8List encodeLsp(Uint8List bodyBytes) {
  final header = 'Content-Length: ${bodyBytes.length}\r\n\r\n';
  final headerBytes = ascii.encode(header);
  return Uint8List.fromList([...headerBytes, ...bodyBytes]);
}

Stream<Uint8List> chunk(Uint8List bytes,
    {int maxChunk = 11, int seed = 2}) async* {
  final rng = math.Random(seed);
  var off = 0;
  while (off < bytes.length) {
    final n = math.min(bytes.length - off, 1 + rng.nextInt(maxChunk));
    yield Uint8List.sublistView(bytes, off, off + n);
    off += n;
  }
}

Future<void> main() async {
  final msg = {
    'jsonrpc': '2.0',
    'id': 1,
    'method': 'initialize',
    'params': {
      'capabilities': {},
    },
  };

  final body = Uint8List.fromList(utf8.encode(json.encode(msg)));
  final wire = encodeLsp(body);

  await for (final payload in chunk(wire).decodeFrames(ContentLengthCodec())) {
    final text = utf8.decode(payload);
    print(text);
  }
}

example/lsp_json_rpc.dart

import 'dart:convert';
import 'dart:math' as math;
import 'dart:typed_data';

import 'package:framer/framer.dart';

Uint8List encodeLsp(Uint8List bodyBytes) {
  final header = 'Content-Length: ${bodyBytes.length}\r\n\r\n';
  return Uint8List.fromList([...ascii.encode(header), ...bodyBytes]);
}

Stream<Uint8List> chunk(Uint8List bytes,
    {int maxChunk = 11, int seed = 4}) async* {
  final rng = math.Random(seed);
  var off = 0;
  while (off < bytes.length) {
    final n = math.min(bytes.length - off, 1 + rng.nextInt(maxChunk));
    yield Uint8List.sublistView(bytes, off, off + n);
    off += n;
  }
}

Future<void> main() async {
  final msg = {
    'jsonrpc': '2.0',
    'id': 1,
    'method': 'initialize',
    'params': {
      'capabilities': {},
    },
  };

  final body = Uint8List.fromList(utf8.encode(json.encode(msg)));
  final wire = encodeLsp(body);

  await for (final obj in chunk(wire).decodeFrames(LspJsonRpcCodec())) {
    print(obj);
  }
}

example/sse.dart

import 'dart:convert';
import 'dart:math' as math;
import 'dart:typed_data';

import 'package:framer/framer.dart';

Stream<Uint8List> chunk(Uint8List bytes,
    {int maxChunk = 7, int seed = 1}) async* {
  final rng = math.Random(seed);
  var off = 0;
  while (off < bytes.length) {
    final n = math.min(bytes.length - off, 1 + rng.nextInt(maxChunk));
    yield Uint8List.sublistView(bytes, off, off + n);
    off += n;
  }
}

Future<void> main() async {
  final sse = [
    'event: greeting\n',
    'data: hello\n',
    '\n',
    'data: multi\n',
    'data: line\n',
    '\n',
  ].join();

  final bytes = Uint8List.fromList(utf8.encode(sse));

  await for (final e in chunk(bytes).decodeFrames(SseCodec())) {
    print('type=${e.type} data=${e.data} id=${e.id}');
  }
}

example/websocket_frames.dart

import 'dart:math' as math;
import 'dart:typed_data';

import 'package:framer/framer.dart';

Uint8List buildWsFrame({
  required Uint8List payload,
  int opcode = WebSocketFrame.opcodeBinary,
  bool fin = true,
  bool masked = false,
  int mask0 = 1,
  int mask1 = 2,
  int mask2 = 3,
  int mask3 = 4,
}) {
  final len = payload.length;
  final bytes = <int>[];

  bytes.add((fin ? 0x80 : 0x00) | (opcode & 0x0F));

  if (len <= 125) {
    bytes.add((masked ? 0x80 : 0x00) | len);
  } else if (len <= 0xFFFF) {
    bytes.add((masked ? 0x80 : 0x00) | 126);
    bytes.add((len >> 8) & 0xFF);
    bytes.add(len & 0xFF);
  } else {
    bytes.add((masked ? 0x80 : 0x00) | 127);
    // 8-byte length
    final v = len;
    for (int i = 7; i >= 0; i--) {
      bytes.add((v >> (8 * i)) & 0xFF);
    }
  }

  if (masked) {
    bytes.addAll([mask0, mask1, mask2, mask3]);
    for (int i = 0; i < payload.length; i++) {
      final m = switch (i & 3) {
        0 => mask0,
        1 => mask1,
        2 => mask2,
        _ => mask3,
      };
      bytes.add(payload[i] ^ m);
    }
  } else {
    bytes.addAll(payload);
  }

  return Uint8List.fromList(bytes);
}

Stream<Uint8List> chunk(Uint8List bytes,
    {int maxChunk = 13, int seed = 1}) async* {
  final rng = math.Random(seed);
  var off = 0;
  while (off < bytes.length) {
    final n = math.min(bytes.length - off, 1 + rng.nextInt(maxChunk));
    yield Uint8List.sublistView(bytes, off, off + n);
    off += n;
  }
}

Future<void> main() async {
  final frames = <Uint8List>[
    buildWsFrame(payload: Uint8List.fromList([1, 2, 3, 4])),
    buildWsFrame(payload: Uint8List.fromList([5, 6, 7])),
    buildWsFrame(payload: Uint8List(0)),
  ];

  final wire = Uint8List.fromList(frames.expand((f) => f).toList());

  final codec = WebSocketFrameCodec(requireMask: false);
  await for (final f in chunk(wire).decodeFrames(codec)) {
    print(f);
  }
}