Coverage Report

Created: 2024-11-21 14:00

/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
42
#define THRIFT_MOVE_VALUES(thrift, member, value) \
43
0
    thrift.__isset.member = true;                 \
44
0
    thrift.member = std::move(value);
45
46
// Super class for templatized thrift clients.
47
class ThriftClientImpl {
48
public:
49
0
    virtual ~ThriftClientImpl() { close(); }
50
0
    const std::string& ipaddress() { return _ipaddress; }
51
0
    int port() const { return _port; }
52
53
    // Open the connection to the remote server. May be called
54
    // repeatedly, is idempotent unless there is a failure to connect.
55
    Status open();
56
57
    // Retry the Open num_retries time waiting wait_ms milliseconds between retries.
58
    Status open_with_retry(int num_retries, int wait_ms);
59
60
    // close the connection with the remote server. May be called
61
    // repeatedly.
62
    void close();
63
64
    // Set the connect timeout
65
0
    void set_conn_timeout(int ms) { _socket->setConnTimeout(ms); }
66
67
    // Set the receive timeout
68
0
    void set_recv_timeout(int ms) { _socket->setRecvTimeout(ms); }
69
70
    // Set the send timeout
71
0
    void set_send_timeout(int ms) { _socket->setSendTimeout(ms); }
72
73
protected:
74
    ThriftClientImpl(const std::string& ipaddress, int port)
75
            : _ipaddress(ipaddress),
76
              _port(port),
77
0
              _socket(new apache::thrift::transport::TSocket(ipaddress, port)) {}
78
79
    std::string _ipaddress;
80
    int _port;
81
82
    // All shared pointers, because Thrift requires them to be
83
    std::shared_ptr<apache::thrift::transport::TSocket> _socket;
84
    std::shared_ptr<apache::thrift::transport::TTransport> _transport;
85
    std::shared_ptr<apache::thrift::protocol::TBinaryProtocol> _protocol;
86
};
87
88
// Utility client to a Thrift server. The parameter type is the
89
// Thrift interface type that the server implements.
90
template <class InterfaceType>
91
class ThriftClient : public ThriftClientImpl {
92
public:
93
    ThriftClient(const std::string& ipaddress, int port);
94
95
    ThriftClient(const std::string& ipaddress, int port, ThriftServer::ServerType server_type);
96
97
    // Returns the object used to actually make RPCs against the remote server
98
0
    InterfaceType* iface() { return _iface.get(); }
Unexecuted instantiation: _ZN5doris12ThriftClientINS_21FrontendServiceClientEE5ifaceEv
Unexecuted instantiation: _ZN5doris12ThriftClientINS_24TPaloBrokerServiceClientEE5ifaceEv
Unexecuted instantiation: _ZN5doris12ThriftClientINS_20BackendServiceClientEE5ifaceEv
99
100
private:
101
    std::shared_ptr<InterfaceType> _iface;
102
};
103
104
template <class InterfaceType>
105
ThriftClient<InterfaceType>::ThriftClient(const std::string& ipaddress, int port)
106
        : ThriftClientImpl(ipaddress, port) {
107
    _transport.reset(new apache::thrift::transport::TBufferedTransport(_socket));
108
    _protocol.reset(new apache::thrift::protocol::TBinaryProtocol(_transport));
109
    _iface.reset(new InterfaceType(_protocol));
110
}
111
112
template <class InterfaceType>
113
ThriftClient<InterfaceType>::ThriftClient(const std::string& ipaddress, int port,
114
                                          ThriftServer::ServerType server_type)
115
0
        : ThriftClientImpl(ipaddress, port) {
116
0
    switch (server_type) {
117
0
    case ThriftServer::NON_BLOCKING:
118
0
        _transport.reset(new apache::thrift::transport::TFramedTransport(_socket));
119
0
        break;
120
0
    case ThriftServer::THREADED:
121
0
    case ThriftServer::THREAD_POOL:
122
0
        _transport.reset(new apache::thrift::transport::TBufferedTransport(_socket));
123
0
        break;
124
0
    default:
125
0
        std::stringstream error_msg;
126
0
        error_msg << "Unsupported server type: " << server_type;
127
0
        LOG(ERROR) << error_msg.str();
128
0
        DCHECK(false);
129
0
        break;
130
0
    }
131
132
0
    _protocol.reset(new apache::thrift::protocol::TBinaryProtocol(_transport));
133
0
    _iface.reset(new InterfaceType(_protocol));
134
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
135
136
} // namespace doris