| 499 | } |
| 500 | |
| 501 | void ProcessSofaResponse(InputMessageBase* msg_base) { |
| 502 | const int64_t start_parse_us = butil::cpuwide_time_us(); |
| 503 | DestroyingPtr<MostCommonMessage> msg(static_cast<MostCommonMessage*>(msg_base)); |
| 504 | SofaRpcMeta meta; |
| 505 | if (!ParsePbFromIOBuf(&meta, msg->meta)) { |
| 506 | LOG(WARNING) << "Fail to parse response meta"; |
| 507 | return; |
| 508 | } |
| 509 | |
| 510 | const bthread_id_t cid = { static_cast<uint64_t>(meta.sequence_id()) }; |
| 511 | Controller* cntl = NULL; |
| 512 | const int rc = bthread_id_lock(cid, (void**)&cntl); |
| 513 | if (rc != 0) { |
| 514 | LOG_IF(ERROR, rc != EINVAL && rc != EPERM) |
| 515 | << "Fail to lock correlation_id=" << cid << ": " << berror(rc); |
| 516 | return; |
| 517 | } |
| 518 | |
| 519 | ControllerPrivateAccessor accessor(cntl); |
| 520 | if (auto span = accessor.span()) { |
| 521 | span->set_base_real_us(msg->base_real_us()); |
| 522 | span->set_received_us(msg->received_us()); |
| 523 | span->set_response_size(msg->meta.size() + msg->payload.size() + 24); |
| 524 | span->set_start_parse_us(start_parse_us); |
| 525 | } |
| 526 | const int saved_error = cntl->ErrorCode(); |
| 527 | if (meta.error_code() != 0) { |
| 528 | // If error_code is unset, default is 0 = success. |
| 529 | cntl->SetFailed(meta.error_code(), |
| 530 | "%s", meta.reason().c_str()); |
| 531 | } else if (cntl->response()) { |
| 532 | // Parse response message iff error code from meta is 0 |
| 533 | CompressType res_cmp_type = Sofa2CompressType(meta.compress_type()); |
| 534 | if (!ParseFromCompressedData( |
| 535 | msg->payload, cntl->response(), res_cmp_type)) { |
| 536 | cntl->SetFailed( |
| 537 | ERESPONSE, "Fail to parse response message, " |
| 538 | "CompressType=%d, response_size=%" PRIu64, |
| 539 | res_cmp_type, (uint64_t)msg->payload.length()); |
| 540 | } else { |
| 541 | cntl->set_response_compress_type(res_cmp_type); |
| 542 | } |
| 543 | } // else silently ignore the response. |
| 544 | |
| 545 | // Unlocks correlation_id inside. Revert controller's |
| 546 | // error code if it version check of `cid' fails |
| 547 | msg.reset(); // optional, just release resource ASAP |
| 548 | accessor.OnResponse(cid, saved_error); |
| 549 | } |
| 550 | |
| 551 | void PackSofaRequest(butil::IOBuf* req_buf, |
| 552 | SocketMessage**, |
nothing calls this directly
no test coverage detected