src/sofa/pbrpc/rpc_client_impl.cc
// Rpc call method to remote endpoint.
//
// The call can be done in following cases:
// * send failed
// * timeouted
// * received response
Rpc call method to remote endpoint.
The call can be done in following cases:
send failed
timeouted
received response
Rpc call method to remote endpoint . The call can be done in following cases : send failed timeouted received response
void RpcClientImpl::CallMethod(const google::protobuf::Message* request,
        google::protobuf::Message* response,
        const RpcControllerImplPtr& cntl)
{
    if (!_is_running)
    {
#if defined( LOG )
        LOG(ERROR) << "CallMethod(): client not in running, ignore";
#else
        SLOG(ERROR, "CallMethod(): client not in running, ignore");
#endif
        cntl->Done(RPC_ERROR_NOT_IN_RUNNING, "client not in running, should start it first");
        return;
    }
    RpcClientStreamPtr stream = FindOrCreateStream(cntl->RemoteEndpoint());
    if (!stream)
    {
#if defined( LOG )
        LOG(ERROR) << "CallMethod(): create socket stream failed: "
                   << RpcEndpointToString(cntl->RemoteEndpoint());
#else
        SLOG(ERROR, "CallMethod(): create socket stream failed: %s",
                RpcEndpointToString(cntl->RemoteEndpoint()).c_str());
#endif
        cntl->Done(RPC_ERROR_CREATE_STREAM, "create stream failed, maybe exceed connection limit");
        return;
    }
    cntl->SetRpcClientStream(stream);
    if (stream->pending_buffer_size() > stream->max_pending_buffer_size())
    {
#if defined( LOG )
#else
        SLOG(DEBUG, "CallMethod(): pending buffer full: %s",
                RpcEndpointToString(cntl->RemoteEndpoint()).c_str());
#endif
        cntl->Done(RPC_ERROR_SEND_BUFFER_FULL, "pending buffer full");
        return;
    }
    cntl->SetSequenceId(GenerateSequenceId());
    RpcMeta meta;
    meta.set_type(RpcMeta::REQUEST);
    meta.set_sequence_id(cntl->SequenceId());
    meta.set_method(cntl->MethodId());
    int64 timeout = cntl->Timeout();
    if (timeout > 0)
    {
        meta.set_server_timeout(timeout);
    }
    meta.set_compress_type(cntl->RequestCompressType());
    meta.set_expected_response_compress_type(cntl->ResponseCompressType());
    RpcMessageHeader header;
    int header_size = sizeof(header);
    WriteBuffer write_buffer;
    int64 header_pos = write_buffer.Reserve(header_size);
    if (header_pos < 0)
    {
#if defined( LOG )
        LOG(ERROR) << "CallMethod(): " << RpcEndpointToString(cntl->RemoteEndpoint())
                   << ": reserve rpc message header failed";
#else
        SLOG(ERROR, "CallMethod(): %s: reserve rpc message header failed",
                RpcEndpointToString(cntl->RemoteEndpoint()).c_str());
#endif
        cntl->Done(RPC_ERROR_SERIALIZE_REQUEST, "reserve rpc message header failed");
        return;
    }
    if (!meta.SerializeToZeroCopyStream(&write_buffer))
    {
#if defined( LOG )
        LOG(ERROR) << "CallMethod(): " << RpcEndpointToString(cntl->RemoteEndpoint())
                   << ": serialize rpc meta failed";
#else
        SLOG(ERROR, "CallMethod(): %s: serialize rpc meta failed",
                RpcEndpointToString(cntl->RemoteEndpoint()).c_str());
#endif
        cntl->Done(RPC_ERROR_SERIALIZE_REQUEST, "serialize rpc meta failed");
        return;
    }
    header.meta_size = static_cast<int>(write_buffer.ByteCount() - header_pos - header_size);
    bool serialize_request_return = false;
    if (meta.compress_type() == CompressTypeNone)
    {
        serialize_request_return = request->SerializeToZeroCopyStream(&write_buffer);
    }
    else
    {
        ::sofa::pbrpc::scoped_ptr<AbstractCompressedOutputStream> os(
                get_compressed_output_stream(&write_buffer, meta.compress_type()));
        serialize_request_return = request->SerializeToZeroCopyStream(os.get());
        os->Flush();
    }
    if (!serialize_request_return)
    {
#if defined( LOG )
        LOG(ERROR) << "CallMethod(): " << RpcEndpointToString(cntl->RemoteEndpoint())
                   << ": serialize request message failed";
#else
        SLOG(ERROR, "CallMethod(): %s: serialize request message failed",
                RpcEndpointToString(cntl->RemoteEndpoint()).c_str());
#endif
        cntl->Done(RPC_ERROR_SERIALIZE_REQUEST, "serialize request message failed");
        return;
    }
    header.data_size = write_buffer.ByteCount() - header_pos - header_size - header.meta_size;
    header.message_size = header.meta_size + header.data_size;
    write_buffer.SetData(header_pos, reinterpret_cast<const char*>(&header), header_size);
    ReadBufferPtr read_buffer(new ReadBuffer());
    write_buffer.SwapOut(read_buffer.get());
    cntl->SetRequestBuffer(read_buffer);
    cntl->PushDoneCallback(boost::bind(&RpcClientImpl::DoneCallback, shared_from_this(), response, _1));
    if (timeout > 0)
    {
        if (!_timeout_manager->add(cntl))
        {
#if defined( LOG )
            LOG(ERROR) << "CallMethod(): " << RpcEndpointToString(cntl->RemoteEndpoint())
                       << ": add to timeout manager failed: timeout=" << timeout << "ms";
#else
            SLOG(ERROR, "CallMethod(): %s: add to timeout manager failed: timeout=%lldms",
                    RpcEndpointToString(cntl->RemoteEndpoint()).c_str(), timeout);
#endif
            cntl->Done(RPC_ERROR_REQUEST_TIMEOUT, "add to timeout manager failed, maybe too short timeout");
            return;
        }
    }
    stream->call_method(cntl);
}
void RpcClientImpl :: CallMethod ( const google :: protobuf :: Message * request , google :: protobuf :: Message * response , const RpcControllerImplPtr & cntl ) { if ( ! _is_running ) { #if defined ( LOG ) LOG ( ERROR ) << " " ; #else SLOG ( ERROR , " " ) ; #endif cntl -> Done ( RPC_ERROR_NOT_IN_RUNNING , " " ) ; return ; } RpcClientStreamPtr stream = FindOrCreateStream ( cntl -> RemoteEndpoint ( ) ) ; if ( ! stream ) { #if defined ( LOG ) LOG ( ERROR ) << " " << RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) ; #else SLOG ( ERROR , " " , RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) . c_str ( ) ) ; #endif cntl -> Done ( RPC_ERROR_CREATE_STREAM , " " ) ; return ; } cntl -> SetRpcClientStream ( stream ) ; if ( stream -> pending_buffer_size ( ) > stream -> max_pending_buffer_size ( ) ) { #if defined ( LOG ) #else SLOG ( DEBUG , " " , RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) . c_str ( ) ) ; #endif cntl -> Done ( RPC_ERROR_SEND_BUFFER_FULL , " " ) ; return ; } cntl -> SetSequenceId ( GenerateSequenceId ( ) ) ; RpcMeta meta ; meta . set_type ( RpcMeta :: REQUEST ) ; meta . set_sequence_id ( cntl -> SequenceId ( ) ) ; meta . set_method ( cntl -> MethodId ( ) ) ; int64 timeout = cntl -> Timeout ( ) ; if ( timeout > 0 ) { meta . set_server_timeout ( timeout ) ; } meta . set_compress_type ( cntl -> RequestCompressType ( ) ) ; meta . set_expected_response_compress_type ( cntl -> ResponseCompressType ( ) ) ; RpcMessageHeader header ; int header_size = sizeof ( header ) ; WriteBuffer write_buffer ; int64 header_pos = write_buffer . Reserve ( header_size ) ; if ( header_pos < 0 ) { #if defined ( LOG ) LOG ( ERROR ) << " " << RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) << " " ; #else SLOG ( ERROR , " " , RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) . c_str ( ) ) ; #endif cntl -> Done ( RPC_ERROR_SERIALIZE_REQUEST , " " ) ; return ; } if ( ! meta . SerializeToZeroCopyStream ( & write_buffer ) ) { #if defined ( LOG ) LOG ( ERROR ) << " " << RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) << " " ; #else SLOG ( ERROR , " " , RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) . c_str ( ) ) ; #endif cntl -> Done ( RPC_ERROR_SERIALIZE_REQUEST , " " ) ; return ; } header . meta_size = static_cast < int > ( write_buffer . ByteCount ( ) - header_pos - header_size ) ; bool serialize_request_return = false ; if ( meta . compress_type ( ) == CompressTypeNone ) { serialize_request_return = request -> SerializeToZeroCopyStream ( & write_buffer ) ; } else { :: sofa :: pbrpc :: scoped_ptr < AbstractCompressedOutputStream > os ( get_compressed_output_stream ( & write_buffer , meta . compress_type ( ) ) ) ; serialize_request_return = request -> SerializeToZeroCopyStream ( os . get ( ) ) ; os -> Flush ( ) ; } if ( ! serialize_request_return ) { #if defined ( LOG ) LOG ( ERROR ) << " " << RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) << " " ; #else SLOG ( ERROR , " " , RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) . c_str ( ) ) ; #endif cntl -> Done ( RPC_ERROR_SERIALIZE_REQUEST , " " ) ; return ; } header . data_size = write_buffer . ByteCount ( ) - header_pos - header_size - header . meta_size ; header . message_size = header . meta_size + header . data_size ; write_buffer . SetData ( header_pos , reinterpret_cast < const char * > ( & header ) , header_size ) ; ReadBufferPtr read_buffer ( new ReadBuffer ( ) ) ; write_buffer . SwapOut ( read_buffer . get ( ) ) ; cntl -> SetRequestBuffer ( read_buffer ) ; cntl -> PushDoneCallback ( boost :: bind ( & RpcClientImpl :: DoneCallback , shared_from_this ( ) , response , _1 ) ) ; if ( timeout > 0 ) { if ( ! _timeout_manager -> add ( cntl ) ) { #if defined ( LOG ) LOG ( ERROR ) << " " << RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) << " " << timeout << " " ; #else SLOG ( ERROR , " " , RpcEndpointToString ( cntl -> RemoteEndpoint ( ) ) . c_str ( ) , timeout ) ; #endif cntl -> Done ( RPC_ERROR_REQUEST_TIMEOUT , " " ) ; return ; } } stream -> call_method ( cntl ) ; }
Rpc call method to remote endpoint.
Rpc call method to remote endpoint .
// get the stream // check the pending buffer full // generate sequence id // prepare request buffer // push callback // add to timeout manager if need // call method