Coverage Report

Created: 2024-11-21 13:02

/root/doris/be/src/util/thrift_client.h
Line
Count
Source (jump to first uncovered line)
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
20
#include <thrift/protocol/TBinaryProtocol.h>
21
#include <thrift/transport/TBufferTransports.h>
22
#include <thrift/transport/TSocket.h>
23
24
#include <memory>
25
#include <sstream>
26
#include <string>
27
28
#include "common/logging.h"
29
#include "common/status.h"
30
#include "util/thrift_server.h"
31
32
namespace apache {
33
namespace thrift {
34
namespace transport {
35
class TTransport;
36
} // namespace transport
37
} // namespace thrift
38
} // namespace apache
39
40
namespace doris {
41
// Super class for templatized thrift clients.
42
class ThriftClientImpl {
43
public:
44
0
    virtual ~ThriftClientImpl() { close(); }
45
0
    const std::string& ipaddress() { return _ipaddress; }
46
0
    int port() const { return _port; }
47
48
    // Open the connection to the remote server. May be called
49
    // repeatedly, is idempotent unless there is a failure to connect.
50
    Status open();
51
52
    // Retry the Open num_retries time waiting wait_ms milliseconds between retries.
53
    Status open_with_retry(int num_retries, int wait_ms);
54
55
    // close the connection with the remote server. May be called
56
    // repeatedly.
57
    void close();
58
59
    // Set the connect timeout
60
0
    void set_conn_timeout(int ms) { _socket->setConnTimeout(ms); }
61
62
    // Set the receive timeout
63
0
    void set_recv_timeout(int ms) { _socket->setRecvTimeout(ms); }
64
65
    // Set the send timeout
66
0
    void set_send_timeout(int ms) { _socket->setSendTimeout(ms); }
67
68
protected:
69
    ThriftClientImpl(const std::string& ipaddress, int port)
70
            : _ipaddress(ipaddress),
71
              _port(port),
72
0
              _socket(new apache::thrift::transport::TSocket(ipaddress, port)) {}
73
74
    std::string _ipaddress;
75
    int _port;
76
77
    // All shared pointers, because Thrift requires them to be
78
    std::shared_ptr<apache::thrift::transport::TSocket> _socket;
79
    std::shared_ptr<apache::thrift::transport::TTransport> _transport;
80
    std::shared_ptr<apache::thrift::protocol::TBinaryProtocol> _protocol;
81
};
82
83
// Utility client to a Thrift server. The parameter type is the
84
// Thrift interface type that the server implements.
85
template <class InterfaceType>
86
class ThriftClient : public ThriftClientImpl {
87
public:
88
    ThriftClient(const std::string& ipaddress, int port);
89
90
    ThriftClient(const std::string& ipaddress, int port, ThriftServer::ServerType server_type);
91
92
    // Returns the object used to actually make RPCs against the remote server
93
0
    InterfaceType* iface() { return _iface.get(); }
Unexecuted instantiation: _ZN5doris12ThriftClientINS_21FrontendServiceClientEE5ifaceEv
Unexecuted instantiation: _ZN5doris12ThriftClientINS_24TPaloBrokerServiceClientEE5ifaceEv
Unexecuted instantiation: _ZN5doris12ThriftClientINS_20BackendServiceClientEE5ifaceEv
94
95
private:
96
    std::shared_ptr<InterfaceType> _iface;
97
};
98
99
template <class InterfaceType>
100
ThriftClient<InterfaceType>::ThriftClient(const std::string& ipaddress, int port)
101
        : ThriftClientImpl(ipaddress, port) {
102
    _transport.reset(new apache::thrift::transport::TBufferedTransport(_socket));
103
    _protocol.reset(new apache::thrift::protocol::TBinaryProtocol(_transport));
104
    _iface.reset(new InterfaceType(_protocol));
105
}
106
107
template <class InterfaceType>
108
ThriftClient<InterfaceType>::ThriftClient(const std::string& ipaddress, int port,
109
                                          ThriftServer::ServerType server_type)
110
0
        : ThriftClientImpl(ipaddress, port) {
111
0
    switch (server_type) {
112
0
    case ThriftServer::NON_BLOCKING:
113
0
        _transport.reset(new apache::thrift::transport::TFramedTransport(_socket));
114
0
        break;
115
0
    case ThriftServer::THREADED:
116
0
    case ThriftServer::THREAD_POOL:
117
0
        _transport.reset(new apache::thrift::transport::TBufferedTransport(_socket));
118
0
        break;
119
0
    default:
120
0
        std::stringstream error_msg;
121
0
        error_msg << "Unsupported server type: " << server_type;
122
0
        LOG(ERROR) << error_msg.str();
123
0
        DCHECK(false);
124
0
        break;
125
0
    }
126
127
0
    _protocol.reset(new apache::thrift::protocol::TBinaryProtocol(_transport));
128
0
    _iface.reset(new InterfaceType(_protocol));
129
0
}
Unexecuted instantiation: _ZN5doris12ThriftClientINS_21FrontendServiceClientEEC2ERKNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEEiNS_12ThriftServer10ServerTypeE
Unexecuted instantiation: _ZN5doris12ThriftClientINS_24TPaloBrokerServiceClientEEC2ERKNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEEiNS_12ThriftServer10ServerTypeE
Unexecuted instantiation: _ZN5doris12ThriftClientINS_20BackendServiceClientEEC2ERKNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEEiNS_12ThriftServer10ServerTypeE
130
131
} // namespace doris