Skip to content

Commit b3b03a7

Browse files
tigersktigersk
committed
Merge branch 'feature_tcp_client' of git.tigerbrokers.net:server/openapi/openapi-cpp-sdk into feature_tcp_client
2 parents 4f21f25 + 700b6ba commit b3b03a7

File tree

4 files changed

+16
-61
lines changed

4 files changed

+16
-61
lines changed

‎include/tigerapi/push_socket/push_socket.h‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,9 @@ namespace TIGER_API
1616
{
1717
enum class SocketState
1818
{
19-
CONNECTING, //正在连接
20-
CONNECTED, //已连接
21-
DISCONNECTED //已断开
19+
CONNECTING,
20+
CONNECTED,
21+
DISCONNECTED
2222
};
2323

2424
class PushSocket : public std::enable_shared_from_this<PushSocket>
@@ -82,8 +82,8 @@ namespace TIGER_API
8282
private:
8383
std::function<void()> connected_callback_;
8484
std::function<void()> disconnected_callback_;
85-
std::function<void(const std::shared_ptr<tigeropen::push::pb::Response>& response_pb_object)> on_message_callback_;
8685
std::function<void(std::string)> on_inner_error_callback_;
86+
std::function<void(const std::shared_ptr<tigeropen::push::pb::Response>& response_pb_object)> on_message_callback_;
8787
private:
8888
TIGER_API::ClientConfig client_config_;
8989

@@ -101,9 +101,9 @@ namespace TIGER_API
101101
private:
102102
std::time_t last_send_heart_beat_time_ = 0;
103103
std::time_t last_io_time_ = 0;
104-
int reconnect_interval_ = 10 * 1000; //单位：ms
105-
int send_interval_ = 10 * 1000; //单位：ms
106-
int recv_interval_ = 10 * 1000; //单位：ms
104+
int reconnect_interval_ = 10 * 1000; //ms
105+
int send_interval_ = 10 * 1000; //ms
106+
int recv_interval_ = 10 * 1000; //ms
107107
};
108108
}
109109
#endif // PUSH_SOCKET_H

‎src/push_socket/push_client_impl.cpp‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@ TIGER_API::PushClientImpl::PushClientImpl(const TIGER_API::ClientConfig& client_
2727
void TIGER_API::PushClientImpl::connect()
2828
{
2929
LOG(INFO) << "create a worker thread to perform asynchronous network connections";
30-
//启动工作线程
3130
worker_thread_ = std::shared_ptr<std::thread>(new std::thread([this]
3231
{
3332

@@ -40,7 +39,7 @@ void TIGER_API::PushClientImpl::connect()
4039

4140
void TIGER_API::PushClientImpl::disconnect()
4241
{
43-
//跨线程调用，需要异步投递任务
42+
//post task to work thread
4443
io_service_.post(boost::bind(&PushClientImpl::do_disconnect, this));
4544
}
4645

@@ -112,7 +111,6 @@ bool TIGER_API::PushClientImpl::unsubscribe_asset(const std::string& account)
112111

113112
bool TIGER_API::PushClientImpl::send_frame(const tigeropen::push::pb::Request& request)
114113
{
115-
//序列化pb对象到字符串
116114
std::string packed_frame = request.SerializeAsString();
117115
if (packed_frame.empty())
118116
{
@@ -124,8 +122,6 @@ bool TIGER_API::PushClientImpl::send_frame(const tigeropen::push::pb::Request& r
124122
MessageToJsonString(request, &packed_frame_json, options).ok();
125123

126124
LOG(DEBUG) << "send frame:" << packed_frame_json;
127-
128-
//跨线程，异步投递任务
129125
io_service_.post(boost::bind(&PushClientImpl::do_write, this, packed_frame));
130126

131127
return true;

‎src/push_socket/push_frame_serialize.cpp‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -27,17 +27,13 @@ std::vector<unsigned char> TIGER_API::PushFrameEncoder::encode_frame(const std::
2727

2828
bool TIGER_API::PushFrameDecoder::push_byte(unsigned char value)
2929
{
30-
//逐个字节压入缓冲区
3130
frame_header_buffer_.push_back(value);
3231
if (!(value & 0x80))
3332
{
34-
//完整读取到包长度
3533
frame_len_ = decode_varint();
3634
frame_header_buffer_.clear();
37-
//frame头部已经读取完毕
3835
return true;
3936
}
40-
//frame头部还有数据，需要继续独取头部数据
4137
return false;
4238
}
4339

@@ -48,7 +44,7 @@ uint32_t TIGER_API::PushFrameDecoder::get_frame_size() const
4844

4945
uint32_t TIGER_API::PushFrameDecoder::decode_varint()
5046
{
51-
const uint32_t mask = (1U << 32) - 1; // 32位掩码
47+
const uint32_t mask = (1U << 32) - 1;
5248
uint32_t result = 0;
5349
int shift = 0;
5450

@@ -62,6 +58,6 @@ uint32_t TIGER_API::PushFrameDecoder::decode_varint()
6258
}
6359
}
6460

65-
result &= mask; // 应用32位掩码
61+
result &= mask;
6662
return result;
6763
}

‎src/push_socket/push_socket.cpp‎

Lines changed: 6 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -28,13 +28,8 @@ TIGER_API::PushSocket::PushSocket(boost::asio::io_service* io_service,
2828
send_interval_ = client_config_.send_interval;
2929
recv_interval_ = client_config_.receive_interval;
3030

31-
//初始化内存池，防止接收数据频繁的内存分配和释放引起的性能问题和内存碎片问题
3231
recv_buff_pool_.reset(new boost::pool<>(MEMORY_POOL_PAGE_SIZE, MEMORY_POOL_BLOCK_NUM));
33-
34-
//创建一个保活定时器
3532
keep_alive_timer_ = std::make_shared<boost::asio::deadline_timer>(*io_service);
36-
37-
//创建重连定时器
3833
reconnect_timer_ = std::make_shared<boost::asio::deadline_timer>(*io_service);
3934
}
4035

@@ -60,9 +55,7 @@ void TIGER_API::PushSocket::set_inner_error_callback(const std::function<void(st
6055

6156
void TIGER_API::PushSocket::connect()
6257
{
63-
//启动保活监测定时任务
6458
start_keep_alive();
65-
6659
try
6760
{
6861
socket_state_ = SocketState::CONNECTING;
@@ -71,8 +64,7 @@ void TIGER_API::PushSocket::connect()
7164
boost::asio::ip::tcp::resolver::query query(utility::conversions::to_utf8string(client_config_.socket_url),
7265
utility::conversions::to_utf8string(client_config_.socket_port));
7366
boost::asio::ip::tcp::resolver::iterator rit = resolver.resolve(query);
74-
75-
//打印dns解析之后的IP地址
67+
7668
std::string str_target_server_ip = rit->endpoint().address().to_string();
7769
LOG(INFO) << "resolved ip: " << str_target_server_ip;
7870

@@ -84,7 +76,6 @@ void TIGER_API::PushSocket::connect()
8476
catch (const boost::system::system_error& e)
8577
{
8678
LOG(ERROR) << e.what();
87-
//dns解析失败/异常连接状态修改为：DISCONNECTED
8879
socket_state_ = SocketState::DISCONNECTED;
8980
}
9081
}
@@ -110,11 +101,9 @@ bool TIGER_API::PushSocket::send_message(const std::string& msg)
110101
return false;
111102
}
112103

113-
//协议封包
114104
TIGER_API::PushFrameEncoder encoder;
115105
std::vector<unsigned char> data = encoder.encode_frame(msg);
116106

117-
//异步发送
118107
boost::asio::async_write(*socket_,
119108
boost::asio::buffer(data, data.size()),
120109
boost::bind(&PushSocket::handle_write, this,
@@ -125,7 +114,7 @@ bool TIGER_API::PushSocket::send_message(const std::string& msg)
125114

126115
void TIGER_API::PushSocket::init_socket()
127116
{
128-
//创建ssl上下文，指定ssl版本
117+
//set ssl version
129118
boost::asio::ssl::context ssl_content(boost::asio::ssl::context::sslv23);
130119
#if 1
131120
ssl_content.set_options(boost::asio::ssl::context::default_workarounds | boost::asio::ssl::context::single_dh_use);
@@ -134,12 +123,10 @@ void TIGER_API::PushSocket::init_socket()
134123
#endif
135124
if (client_config_.socket_ca_certs.empty())
136125
{
137-
//禁用证书验证
138126
ssl_content.set_verify_mode(boost::asio::ssl::verify_none);
139127
}
140128
else
141129
{
142-
//验证证书
143130
ssl_content.set_verify_mode(boost::asio::ssl::verify_peer);
144131
ssl_content.set_verify_mode(boost::asio::ssl::context::verify_peer
145132
| boost::asio::ssl::context::verify_fail_if_no_peer_cert);
@@ -213,15 +200,13 @@ void TIGER_API::PushSocket::close_session()
213200

214201
void TIGER_API::PushSocket::send_authentication()
215202
{
216-
//对tiger_id根据private_key进行签名
203+
//tiger_id sign by private_key
217204
utility::string_t sign = Utils::get_sign(client_config_.private_key, client_config_.tiger_id);
218-
219-
// 创建一个 Request 对象
205+
220206
tigeropen::push::pb::Request request;
221207
request.set_command(tigeropen::push::pb::SocketCommon_Command_CONNECT);
222208
request.set_id(get_next_id());
223209

224-
// 创建一个 Request_Connect 对象
225210
tigeropen::push::pb::Request_Connect* connect_request = request.mutable_connect();
226211
connect_request->set_tigerid(utility::conversions::to_utf8string(client_config_.tiger_id));
227212
connect_request->set_sign(utility::conversions::to_utf8string(sign));
@@ -231,7 +216,6 @@ void TIGER_API::PushSocket::send_authentication()
231216
connect_request->set_receiveinterval(client_config_.receive_interval);
232217
connect_request->set_usefulltick(client_config_.user_full_tick);
233218

234-
//序列化pb对象到字符串
235219
std::string packed_frame = request.SerializeAsString();
236220
if (!packed_frame.empty())
237221
{
@@ -302,7 +286,7 @@ void TIGER_API::PushSocket::handle_connect(const boost::system::error_code& erro
302286
if (!error)
303287
{
304288
LOG(INFO) << "connect success";
305-
//ssl连接成功之后需要async_handshake()，成功之后才能发起异步读写。
289+
//start handshake
306290
socket_->async_handshake(boost::asio::ssl::stream_base::client,
307291
boost::bind(&PushSocket::handle_handshake, this,
308292
boost::asio::placeholders::error));
@@ -319,8 +303,6 @@ void TIGER_API::PushSocket::handle_connect(const boost::system::error_code& erro
319303
{
320304
LOG(ERROR) << "[connect failed]: " << error;
321305
dispatch_inner_error_callback(error.message());
322-
323-
//连接失败关闭会话
324306
close_session();
325307
}
326308
}
@@ -339,27 +321,19 @@ void TIGER_API::PushSocket::handle_handshake(const boost::system::error_code& er
339321
if (!error)
340322
{
341323
LOG(INFO) << "handshake success";
342-
343-
//握手成功之后，设置状态为CONNECTED
344324
socket_state_ = SocketState::CONNECTED;
345-
346-
//设置socket选项
347325
socket_->lowest_layer().set_option(boost::asio::ip::tcp::acceptor::linger(true, 0));
348326
socket_->lowest_layer().set_option(boost::asio::socket_base::keep_alive(true));
349-
350-
//启动异步读任务，保持IO循环的运行
327+
351328
read_head();
352329

353-
//身份认证
354330
send_authentication();
355331

356-
//连接回调，只有建立连接成功之后才会回调
357332
dispatch_connected_callback();
358333
}
359334
else
360335
{
361336
LOG(ERROR) << "[handshake failed]: " << error;
362-
//握手失败关闭会话
363337
dispatch_inner_error_callback(error.message());
364338
close_session();
365339
}
@@ -395,7 +369,6 @@ void TIGER_API::PushSocket::handle_read_head(const boost::system::error_code& er
395369
else
396370
{
397371
#if 1
398-
// 循环打印每个字节的二进制值
399372
for (size_t i = 0; i < bytes_transferred; ++i)
400373
{
401374
std::bitset<8> binary(head_buff_[i]);
@@ -405,13 +378,11 @@ void TIGER_API::PushSocket::handle_read_head(const boost::system::error_code& er
405378
last_io_time_ = time(nullptr);
406379
if (frame_decoder_.push_byte(head_buff_[0]))
407380
{
408-
//头部已经独取完毕，开始读取body
409381
auto frame_len = frame_decoder_.get_frame_size();
410382
read_body(frame_len);
411383
}
412384
else
413385
{
414-
//继续读取头部
415386
read_head();
416387
}
417388
}
@@ -458,17 +429,14 @@ void TIGER_API::PushSocket::handle_read_body(const boost::system::error_code& er
458429

459430
if (on_message_callback_)
460431
{
461-
//下发数据
462432
on_message_callback_(response_pb_object);
463433
}
464434

465-
//缓存还回内存池
466435
if (recv_buff)
467436
{
468437
recv_buff_pool_->ordered_free(recv_buff, page_num);
469438
}
470439

471-
//继续读取下一包数据
472440
read_head();
473441
}
474442

@@ -479,12 +447,10 @@ void TIGER_API::PushSocket::handle_timer(const boost::system::error_code& error)
479447
{
480448
if (socket_state_ == SocketState::CONNECTED)
481449
{
482-
// 检查是否需要发送心跳
483450
if (now_time - last_send_heart_beat_time_ > send_interval_ / 1000)
484451
{
485452
send_heart_beat();
486453
}
487-
//检查空闲状态是否多长时间没有收到数据，断开重连
488454
if (now_time - last_io_time_ > recv_interval_ / 1000)
489455
{
490456
LOG(ERROR) << "heart beat timeout";
@@ -497,8 +463,6 @@ void TIGER_API::PushSocket::handle_timer(const boost::system::error_code& error)
497463
cancel_reconnect_timer();
498464
auto_reconnect();
499465
}
500-
501-
// 再次启动定时器
502466
start_keep_alive();
503467
}
504468
else
@@ -557,7 +521,6 @@ void TIGER_API::PushSocket::message_filter(const std::shared_ptr<tigeropen::push
557521
{
558522
if (response_pb_object->command() == tigeropen::push::pb::SocketCommon_Command_CONNECTED)
559523
{
560-
//连接成功消息
561524
const std::string& str_msg = response_pb_object->msg();
562525
if (str_msg.find(CONNECTED_HEART_BEAT_CFG_KEY) != std::wstring::npos)
563526
{

0 commit comments

Comments
 (0)