Method SetBodyReader

src/brpc/details/http_message.cpp:377–425

static void CreateFailAllRead() { s_fail_all_read = new FailAllRead; }

void HttpMessage::SetBodyReader(ProgressiveReader* r) {
    if (!_read_body_progressively) {
        return r->OnEndOfMessage(
            butil::Status(EPERM, "Call SetBodyReader on HttpMessage with"
                         " read_body_progressively=false"));
    }
    const int MAX_TRY = 3;
    int ntry = 0;
    do {
        std::unique_lock<butil::Mutex> mu(_body_mutex);
        if (_body_reader != NULL) {
            mu.unlock();
            return r->OnEndOfMessage(
                butil::Status(EPERM, "SetBodyReader is called more than once"));
        }
        if (_body.empty()) {
            if (_stage <= HTTP_ON_BODY) {
                _body_reader = r;
                return;
            } else {  // The body is complete and successfully consumed.
                mu.unlock();
                return r->OnEndOfMessage(butil::Status());
            }
        } else if (_stage <= HTTP_ON_BODY && ++ntry >= MAX_TRY) {
            // Stop making _body empty after we've tried several times.
            // If _stage is greater than HTTP_ON_BODY, neither OnBody() nor
            // OnMessageComplete() will be called in future, we have to spin
            // another time to empty _body.
            _body_reader = r;
            return;
        }
        butil::IOBuf body_seen = _body.movable();
        mu.unlock();
        for (size_t i = 0; i < body_seen.backing_block_num(); ++i) {
            butil::StringPiece blk = body_seen.backing_block(i);
            butil::Status st = r->OnReadOnePart(blk.data(), blk.size());
            if (!st.ok()) {
                r->OnEndOfMessage(st);
                // Make OnBody() or OnMessageComplete() fail on next call to
                // close the socket. If the message was already complete, the
                // socket will not be closed.
                pthread_once(&s_fail_all_read_once, CreateFailAllRead);
                r = s_fail_all_read;
                ntry = MAX_TRY;
                break;
            }
        }
    } while (true);
}

const http_parser_settings g_parser_settings = {
    &HttpMessage::on_message_begin,