Function ProcessNsheadResponse

src/brpc/policy/nshead_protocol.cpp:356–389

}

void ProcessNsheadResponse(InputMessageBase* msg_base) {
    const int64_t start_parse_us = butil::cpuwide_time_us();
    DestroyingPtr<MostCommonMessage> msg(static_cast<MostCommonMessage*>(msg_base));
    
    // Fetch correlation id that we saved before in `PackNsheadRequest'
    const CallId cid = { static_cast<uint64_t>(msg->socket()->correlation_id()) };
    Controller* cntl = NULL;
    const int rc = bthread_id_lock(cid, (void**)&cntl);
    if (rc != 0) {
        LOG_IF(ERROR, rc != EINVAL && rc != EPERM)
            << "Fail to lock correlation_id=" << cid << ": " << berror(rc);
        return;
    }

    ControllerPrivateAccessor accessor(cntl);
    if (auto span = accessor.span()) {
        span->set_base_real_us(msg->base_real_us());
        span->set_received_us(msg->received_us());
        span->set_response_size(msg->payload.length());
        span->set_start_parse_us(start_parse_us);
    }
    // MUST be NsheadMessage (checked in SerializeNsheadRequest)
    NsheadMessage* response = (NsheadMessage*)cntl->response();
    const int saved_error = cntl->ErrorCode();
    if (response != NULL) {
        msg->meta.copy_to(&response->head, sizeof(nshead_t));
        msg->payload.swap(response->body);
    } // else just ignore the response.

    // Unlocks correlation_id inside. Revert controller's
    // error code if it version check of `cid' fails
    msg.reset();  // optional, just release resource ASAP
    accessor.OnResponse(cid, saved_error);
}

bool VerifyNsheadRequest(const InputMessageBase* msg_base) {
    Server* server = (Server*)msg_base->arg();