| 909 | } |
| 910 | |
| 911 | void ProcessRpcResponse(InputMessageBase* msg_base) { |
| 912 | const int64_t start_parse_us = butil::cpuwide_time_us(); |
| 913 | DestroyingPtr<MostCommonMessage> msg(static_cast<MostCommonMessage*>(msg_base)); |
| 914 | RpcMeta meta; |
| 915 | if (!ParsePbFromIOBuf(&meta, msg->meta)) { |
| 916 | LOG(WARNING) << "Fail to parse from response meta"; |
| 917 | return; |
| 918 | } |
| 919 | |
| 920 | const bthread_id_t cid = { static_cast<uint64_t>(meta.correlation_id()) }; |
| 921 | Controller* cntl = NULL; |
| 922 | |
| 923 | StreamId remote_stream_id = meta.has_stream_settings() ? meta.stream_settings().stream_id(): INVALID_STREAM_ID; |
| 924 | |
| 925 | const int rc = bthread_id_lock(cid, (void**)&cntl); |
| 926 | if (rc != 0) { |
| 927 | LOG_IF(ERROR, rc != EINVAL && rc != EPERM) |
| 928 | << "Fail to lock correlation_id=" << cid << ": " << berror(rc); |
| 929 | if (remote_stream_id != INVALID_STREAM_ID) { |
| 930 | SendStreamRst(msg->socket(), remote_stream_id); |
| 931 | const auto & extra_stream_ids = meta.stream_settings().extra_stream_ids(); |
| 932 | for (int i = 0; i < extra_stream_ids.size(); ++i) { |
| 933 | policy::SendStreamRst(msg->socket(), extra_stream_ids[i]); |
| 934 | } |
| 935 | } |
| 936 | return; |
| 937 | } |
| 938 | |
| 939 | ControllerPrivateAccessor accessor(cntl); |
| 940 | if (remote_stream_id != INVALID_STREAM_ID) { |
| 941 | accessor.set_remote_stream_settings( |
| 942 | new StreamSettings(meta.stream_settings())); |
| 943 | } |
| 944 | |
| 945 | if (!meta.user_fields().empty()) { |
| 946 | for (const auto& it : meta.user_fields()) { |
| 947 | (*cntl->response_user_fields())[it.first] = it.second; |
| 948 | } |
| 949 | } |
| 950 | |
| 951 | cntl->set_rpc_received_us(msg->received_us()); |
| 952 | if (auto span = accessor.span()) { |
| 953 | span->set_base_real_us(msg->base_real_us()); |
| 954 | span->set_received_us(msg->received_us()); |
| 955 | span->set_response_size(msg->meta.size() + msg->payload.size() + 12); |
| 956 | span->set_start_parse_us(start_parse_us); |
| 957 | } |
| 958 | const RpcResponseMeta &response_meta = meta.response(); |
| 959 | const int saved_error = cntl->ErrorCode(); |
| 960 | do { |
| 961 | if (response_meta.error_code() != 0) { |
| 962 | // If error_code is unset, default is 0 = success. |
| 963 | cntl->SetFailed(response_meta.error_code(), |
| 964 | "%s", response_meta.error_text().c_str()); |
| 965 | break; |
| 966 | } |
| 967 | // Parse response message iff error code from meta is 0 |
| 968 | butil::IOBuf res_buf; |
no test coverage detected