onPushStreamData method

void onPushStreamData(int streamId, Uint8List data)

Process incoming data on a server push stream.

Per RFC 9114 §6.2.2, a push stream begins with the stream type varint (0x01) followed by a varint push ID, then a sequence of HTTP/3 frames carrying the push response. This method parses the push ID, matches it to a previously registered promise (via registerPushPromise or a received PUSH_PROMISE frame), and decodes the HEADERS/DATA frames into an Http3Response made available via getPushResponse.

If no promise is registered for the parsed push ID, the stream data is ignored. The streamId is the QUIC stream ID of the incoming push stream; data is the full stream payload including the leading stream type varint.

Implementation

void onPushStreamData(int streamId, Uint8List data) {
  if (data.isEmpty) return;
  // Skip the stream-type varint (already classified as a push stream).
  final typeLength = VarInt.decodeLength(data[0]);
  if (data.length < typeLength) return;
  // Read the push ID varint immediately after the stream type.
  final pushIdOffset = typeLength;
  if (data.length <= pushIdOffset) return;
  final pushIdLength = VarInt.decodeLength(data[pushIdOffset]);
  if (data.length < pushIdOffset + pushIdLength) return;
  final pushId = VarInt.decode(data.buffer, offset: pushIdOffset);

  var offset = pushIdOffset + pushIdLength;
  Http3Response? response;
  Uint8List? body;
  while (offset < data.length) {
    try {
      final (frame, consumed) = Http3Frame.parse(data, offset: offset);
      switch (frame.type) {
        case Http3FrameType.headers:
          final encoded = Uint8List.fromList(frame.payload);
          qpackDecoder.dynamicTable.resetRequiredInsertCount();
          response =
              Http3Response.decodeHeaders(encoded, decoder: qpackDecoder);
          _pendingDecoderInstructions.add(
            SectionAcknowledgment(streamId: streamId),
          );
          break;
        case Http3FrameType.data:
          final dataFrame = DataFrame.fromPayload(frame.payload);
          if (dataFrame.data.isNotEmpty) {
            final existing = body ?? Uint8List(0);
            final combined =
                Uint8List(existing.length + dataFrame.data.length);
            combined.setRange(0, existing.length, existing);
            combined.setRange(
              existing.length,
              combined.length,
              dataFrame.data,
            );
            body = combined;
          }
          break;
        default:
          // Ignore unrelated frame types on a push stream.
          break;
      }
      offset += consumed;
    } catch (_) {
      break;
    }
  }

  if (response != null) {
    final withBody = body != null && body.isNotEmpty
        ? Http3Response(
            statusCode: response.statusCode,
            headers: response.headers,
            body: body,
          )
        : response;
    _pushResponses[pushId] = withBody;
    _pushStreamIds[pushId] = streamId;
  }
}