Method OnChunks

src/brpc/policy/rtmp_protocol.cpp:1329–1365

}

ParseResult RtmpContext::OnChunks(butil::IOBuf* source, Socket* socket) {
    // Parse basic header.
    const char* p = (const char*)source->fetch1();
    if (NULL == p) {
        return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
    }
    const uint8_t first_byte = *p;
    // 2 bits, deciding type of following chunk message header.
    const RtmpChunkType fmt = (RtmpChunkType)(first_byte >> 6);
    // cs_id is short for "chunk stream id"
    uint32_t cs_id = (first_byte & 0x3F);
    uint8_t basic_header_len = 1u;
    if (cs_id == 0) { // 2-byte basic header
        if (source->size() < 2u) {
            return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
        }
        char basic_header_buf[2];
        const uint8_t* p = (const uint8_t*)source->fetch(basic_header_buf, 2);
        cs_id = ((uint32_t)p[1]) + 64;
        basic_header_len = 2u;
    } else if (cs_id == 1) { // 3-byte basic header
        if (source->size() < 3u) {
            return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
        }
        char basic_header_buf[3];
        const uint8_t* p = (const uint8_t*)source->fetch(basic_header_buf, 3);
        cs_id = ((uint32_t)p[2]) * 256 + ((uint32_t)p[1]) + 64;
        basic_header_len = 3u;
    } // else 1-byte basic header, keep cs_id as it is.
    RtmpBasicHeader bh = { cs_id, fmt, basic_header_len };
    RtmpChunkStream* cstream = GetChunkStream(cs_id);
    if (cstream == NULL) {
        LOG(ERROR) << "Invalid chunk_stream_id=" << cs_id;
        return MakeParseError(PARSE_ERROR_NO_RESOURCE);
    }
    return cstream->Feed(bh, source, socket);
}

static int SendAck(Socket* socket, uint64_t received_bytes) {
    const uint32_t data = butil::HostToNet32(received_bytes);