Function ProcessSofaRequest

src/brpc/policy/sofa_pbrpc_protocol.cpp:321–490

    ::google::protobuf::Closure* done);

void ProcessSofaRequest(InputMessageBase* msg_base) {
    const int64_t start_parse_us = butil::cpuwide_time_us();
    DestroyingPtr<MostCommonMessage> msg(static_cast<MostCommonMessage*>(msg_base));
    SocketUniquePtr socket_guard(msg->ReleaseSocket());
    Socket* socket = socket_guard.get();
    const Server* server = static_cast<const Server*>(msg_base->arg());
    ScopedNonServiceError non_service_error(server);

    SofaRpcMeta meta;
    if (!ParsePbFromIOBuf(&meta, msg->meta)) {
        LOG(WARNING) << "Fail to parse SofaRpcMeta from " << *socket;
        socket->SetFailed(EREQUEST, "Fail to parse SofaRpcMeta from %s",
                          socket->description().c_str());
        return;
    }
    const CompressType req_cmp_type = Sofa2CompressType(meta.compress_type());

    SampledRequest* sample = AskToBeSampled();
    if (sample) {
        sample->meta.set_method_name(meta.method());
        sample->meta.set_compress_type(req_cmp_type);
        sample->meta.set_protocol_type(PROTOCOL_SOFA_PBRPC);
        sample->request = msg->payload;
        sample->submit(start_parse_us);
    }

    std::unique_ptr<Controller> cntl(new (std::nothrow) Controller);
    if (NULL == cntl.get()) {
        LOG(WARNING) << "Fail to new Controller";
        return;
    }
    std::unique_ptr<google::protobuf::Message> req;
    std::unique_ptr<google::protobuf::Message> res;

    ControllerPrivateAccessor accessor(cntl.get());
    ServerPrivateAccessor server_accessor(server);
    const int64_t correlation_id = meta.sequence_id();
    const bool security_mode = server->options().security_mode() &&
                               socket->user() == server_accessor.acceptor();

    cntl->set_request_compress_type(req_cmp_type);
    accessor.set_server(server)
        .set_security_mode(security_mode)
        .set_peer_id(socket->id())
        .set_remote_side(socket->remote_side())
        .set_local_side(socket->local_side())
        .set_auth_context(socket->auth_context())
        .set_request_protocol(PROTOCOL_SOFA_PBRPC)
        .set_begin_time_us(msg->received_us())
        .move_in_server_receiving_sock(socket_guard);

    // Tag the bthread with this server's key for thread_local_data().
    if (server->thread_local_options().thread_local_data_factory) {
        bthread_assign_data((void*)&server->thread_local_options());
    }

    std::shared_ptr<Span> span;
    if (IsTraceable(false)) {