| 452 | }; |
| 453 | |
| 454 | void ProcessThriftRequest(InputMessageBase* msg_base) { |
| 455 | const int64_t start_parse_us = butil::cpuwide_time_us(); |
| 456 | |
| 457 | DestroyingPtr<MostCommonMessage> msg(static_cast<MostCommonMessage*>(msg_base)); |
| 458 | SocketUniquePtr socket_guard(msg->ReleaseSocket()); |
| 459 | Socket* socket = socket_guard.get(); |
| 460 | const Server* server = static_cast<const Server*>(msg_base->arg()); |
| 461 | ScopedNonServiceError non_service_error(server); |
| 462 | |
| 463 | ThriftClosure* thrift_done = new ThriftClosure; |
| 464 | ClosureGuard done_guard(thrift_done); |
| 465 | Controller* cntl = &(thrift_done->_controller); |
| 466 | ThriftFramedMessage* req = &(thrift_done->_request); |
| 467 | ThriftFramedMessage* res = &(thrift_done->_response); |
| 468 | thrift_done->_received_us = msg->received_us(); |
| 469 | |
| 470 | ServerPrivateAccessor server_accessor(server); |
| 471 | const bool security_mode = server->options().security_mode() && |
| 472 | socket->user() == server_accessor.acceptor(); |
| 473 | ControllerPrivateAccessor accessor(cntl); |
| 474 | accessor.set_server(server) |
| 475 | .set_security_mode(security_mode) |
| 476 | .set_peer_id(socket->id()) |
| 477 | .set_remote_side(socket->remote_side()) |
| 478 | .set_local_side(socket->local_side()) |
| 479 | .set_request_protocol(PROTOCOL_THRIFT) |
| 480 | .set_begin_time_us(msg_base->received_us()) |
| 481 | .move_in_server_receiving_sock(socket_guard); |
| 482 | |
| 483 | uint32_t seq_id; |
| 484 | ::apache::thrift::protocol::TMessageType mtype; |
| 485 | butil::Status st = ReadThriftMessageBegin( |
| 486 | &msg->payload, &cntl->_thrift_method_name, &mtype, &seq_id); |
| 487 | if (!st.ok()) { |
| 488 | return cntl->SetFailed(EREQUEST, "%s", st.error_cstr()); |
| 489 | } |
| 490 | msg->payload.swap(req->body); |
| 491 | req->field_id = THRIFT_REQUEST_FID; |
| 492 | cntl->set_log_id(seq_id); // Pass seq_id by log_id |
| 493 | |
| 494 | ThriftService* service = server->options().thrift_service; |
| 495 | if (service == NULL) { |
| 496 | LOG_EVERY_SECOND(ERROR) |
| 497 | << "Received thrift request however the server does not set" |
| 498 | " ServerOptions.thrift_service, close the connection."; |
| 499 | return cntl->SetFailed(EINTERNAL, "ServerOptions.thrift_service is NULL"); |
| 500 | } |
| 501 | |
| 502 | // Switch to service-specific error. |
| 503 | non_service_error.release(); |
| 504 | MethodStatus* method_status = service->_status; |
| 505 | if (method_status) { |
| 506 | if (!method_status->OnRequested()) { |
| 507 | return cntl->SetFailed(ELIMIT, "Reached %s's max_concurrency=%d", |
| 508 | cntl->thrift_method_name().c_str(), |
| 509 | method_status->MaxConcurrency()); |
| 510 | } |
| 511 | } |
nothing calls this directly
no test coverage detected