decodeServerSentEvents function

Stream<String> decodeServerSentEvents(Stream<List<int>> bytes, { int maxEventBytes = 1024 * 1024, int maxResponseBytes = 16 * 1024 * 1024, })

Parses UTF-8 SSE independently of network chunk boundaries. Bounds are enforced before assembling an unbounded line or event, including comments.

Implementation

Stream<String> decodeServerSentEvents(
  Stream<List<int>> bytes, {
  int maxEventBytes = 1024 * 1024,
  int maxResponseBytes = 16 * 1024 * 1024,
}) async* {
  final line = <int>[];
  final data = <String>[];
  var eventBytes = 0;
  var totalBytes = 0;
  var previousCr = false;
  var firstLine = true;
  String? finishLine() {
    var text = utf8.decode(line);
    line.clear();
    if (firstLine) {
      text = text.replaceFirst(RegExp('^\uFEFF'), '');
      firstLine = false;
    }
    if (text.isEmpty) {
      eventBytes = 0;
      if (data.isEmpty) return null;
      final event = data.join('\n');
      data.clear();
      return event;
    }
    if (text.startsWith('data:')) {
      var value = text.substring(5);
      if (value.startsWith(' ')) value = value.substring(1);
      data.add(value);
    }
    return null;
  }

  await for (final chunk in bytes) {
    totalBytes += chunk.length;
    if (totalBytes > maxResponseBytes) {
      throw const ChatApiException(
        'The model response exceeded its size limit.',
      );
    }
    for (final byte in chunk) {
      if (previousCr && byte == 10) {
        previousCr = false;
        continue;
      }
      previousCr = byte == 13;
      eventBytes++;
      if (eventBytes > maxEventBytes) {
        throw const ChatApiException('A stream event exceeded its size limit.');
      }
      if (byte == 10 || byte == 13) {
        final event = finishLine();
        if (event != null) yield event;
      } else {
        line.add(byte);
      }
    }
  }
  if (line.isNotEmpty) {
    final event = finishLine();
    if (event != null) yield event;
  }
  if (data.isNotEmpty) yield data.join('\n');
}