drain method

Iterable<RemoteFrame> drain()

Pull out every complete frame currently in the buffer. Partial frames stay buffered until the next feed.

Implementation

Iterable<RemoteFrame> drain() sync* {
  final pendingFatalError = _pendingFatalError;
  if (pendingFatalError != null) {
    _pendingFatalError = null;
    _clearBuffer();
    throw pendingFatalError;
  }
  while (true) {
    if (_bufferedLength < 5) return;
    final frameStart = _start;
    final type = FrameType.fromCode(_buffer[frameStart]);
    final length = ByteData.sublistView(
      _buffer,
      frameStart + 1,
      frameStart + 5,
    ).getUint32(0);
    final effectiveLimit = _effectivePayloadLimit(type);
    if (length > effectiveLimit) {
      // Clearing the buffer discards the current framing position with no
      // known next-frame boundary, so the stream can no longer be
      // resynchronized: mark this unrecoverable so a peer tears down and
      // reconnects instead of resyncing forever against a mirror that will
      // never advance (the "frozen, no banner" hang). A single frame whose
      // length was VALID but whose payload is malformed throws a recoverable
      // exception from _decode below (framing intact), so the two cases are
      // distinguishable at the catch site.
      _clearBuffer();
      throw _oversizedFrame(type, length, effectiveLimit);
    }
    final total = 5 + length;
    if (_bufferedLength < total) return;
    final before = _bufferedLength;
    final payloadStart = frameStart + 5;
    final payloadEnd = frameStart + total;
    _start = payloadEnd;
    if (type == null) {
      // The wire is lockstep, so an unknown discriminator is a protocol
      // violation, not a newer peer to tolerate. Framing has already
      // advanced past the frame, so the error is recoverable in the same
      // sense as a malformed known frame: it cannot recur on these bytes.
      final code = _buffer[frameStart];
      _releaseBufferIfEmpty();
      throw RemoteProtocolException(
        'unknown frame type 0x${code.toRadixString(16).padLeft(2, '0')}',
      );
    }
    // Copy exactly one completed payload. Partial feeds are retained in the
    // amortized byte buffer above instead of repeatedly materializing and
    // restoring the entire prefix (the old fragmented-frame O(n²) path).
    final payload = Uint8List.fromList(
      Uint8List.sublistView(_buffer, payloadStart, payloadEnd),
    );
    _releaseBufferIfEmpty();
    // Framing has advanced past this frame (the buffer now holds only the
    // trailing bytes), so a RECOVERABLE _decode throw below cannot recur on
    // these same bytes when the peer feeds again — the F19 non-loop
    // invariant a recoverable classification rests on.
    assert(
      _bufferedLength < before,
      'drain() must consume a frame from the buffer before decoding its '
      'payload, so a recoverable decode error cannot re-parse the same '
      'bytes forever (F19 non-loop invariant).',
    );
    yield _decode(type, payload);
  }
}