mirror of
https://github.com/qicosmos/rest_rpc.git
synced 2026-08-29 16:40:48 +08:00
qicosmos
671 lines
17 KiB
C++
671 lines
17 KiB
C++
#pragma once
|
|
#include <string>
|
|
#include <deque>
|
|
#include <future>
|
|
#include "use_asio.hpp"
|
|
#include "client_util.hpp"
|
|
#include "const_vars.h"
|
|
#include "meta_util.hpp"
|
|
|
|
using namespace rest_rpc::rpc_service;
|
|
|
|
namespace rest_rpc {
|
|
class req_result {
|
|
public:
|
|
req_result() = default;
|
|
req_result(string_view data) : data_(data.data(), data.length()) {}
|
|
bool success() const {
|
|
return !has_error(data_);
|
|
}
|
|
|
|
template<typename T>
|
|
T as() {
|
|
if (has_error(data_)) {
|
|
throw std::logic_error(get_error_msg(data_));
|
|
}
|
|
|
|
return get_result<T>(data_);
|
|
}
|
|
|
|
void as() {
|
|
if (has_error(data_)) {
|
|
throw std::logic_error(get_error_msg(data_));
|
|
}
|
|
}
|
|
|
|
private:
|
|
std::string data_;
|
|
};
|
|
|
|
enum class CallModel {
|
|
future,
|
|
callback
|
|
};
|
|
const constexpr auto FUTURE = CallModel::future;
|
|
|
|
const constexpr size_t DEFAULT_TIMEOUT = 5000; //milliseconds
|
|
|
|
class rpc_client : private asio::noncopyable {
|
|
public:
|
|
rpc_client() : socket_(ios_), work_(ios_),
|
|
deadline_(ios_), body_(INIT_BUF_SIZE) {
|
|
thd_ = std::make_shared<std::thread>([this] {
|
|
ios_.run();
|
|
});
|
|
}
|
|
|
|
rpc_client(const std::string& host, unsigned short port) : socket_(ios_), work_(ios_),
|
|
deadline_(ios_), host_(host), port_(port), body_(INIT_BUF_SIZE) {
|
|
thd_ = std::make_shared<std::thread>([this] {
|
|
ios_.run();
|
|
});
|
|
}
|
|
|
|
~rpc_client() {
|
|
close();
|
|
stop();
|
|
}
|
|
|
|
void run(){
|
|
thd_->join();
|
|
}
|
|
|
|
void set_connect_timeout(size_t milliseconds) {
|
|
connect_timeout_ = milliseconds;
|
|
}
|
|
|
|
void set_reconnect_count(int reconnect_count) {
|
|
reconnect_cnt_ = reconnect_count;
|
|
}
|
|
|
|
bool connect(size_t timeout = 1) {
|
|
if (has_connected_)
|
|
return true;
|
|
|
|
assert(port_ != 0);
|
|
async_connect();
|
|
return wait_conn(timeout);
|
|
}
|
|
|
|
bool connect(const std::string& host, unsigned short port, size_t timeout = 1) {
|
|
if (port_==0) {
|
|
host_ = host;
|
|
port_ = port;
|
|
}
|
|
|
|
return connect(timeout);
|
|
}
|
|
|
|
void async_connect(const std::string& host, unsigned short port) {
|
|
if (port_ == 0) {
|
|
host_ = host;
|
|
port_ = port;
|
|
}
|
|
|
|
async_connect();
|
|
}
|
|
|
|
bool wait_conn(size_t timeout) {
|
|
if (has_connected_) {
|
|
return true;
|
|
}
|
|
|
|
has_wait_ = true;
|
|
std::unique_lock<std::mutex> lock(conn_mtx_);
|
|
bool result = conn_cond_.wait_for(lock, std::chrono::seconds(timeout),
|
|
[this] {return has_connected_.load(); });
|
|
has_wait_ = false;
|
|
return has_connected_;
|
|
}
|
|
|
|
void enable_auto_reconnect(bool enable = true) {
|
|
enable_reconnect_ = enable;
|
|
}
|
|
|
|
void enable_auto_heartbeat(bool enable = true) {
|
|
if (enable) {
|
|
reset_deadline_timer(5);
|
|
}
|
|
else {
|
|
deadline_.cancel();
|
|
}
|
|
}
|
|
|
|
void update_addr(const std::string& host, unsigned short port) {
|
|
host_ = host;
|
|
port_ = port;
|
|
}
|
|
|
|
void close() {
|
|
has_connected_ = false;
|
|
if (socket_.is_open()) {
|
|
boost::system::error_code ignored_ec;
|
|
socket_.shutdown(asio::ip::tcp::socket::shutdown_both, ignored_ec);
|
|
socket_.close(ignored_ec);
|
|
}
|
|
clear_cache();
|
|
}
|
|
|
|
void set_error_callback(std::function<void(boost::system::error_code)> f) {
|
|
err_cb_ = std::move(f);
|
|
}
|
|
|
|
uint64_t reqest_id() {
|
|
return temp_req_id_;
|
|
}
|
|
|
|
bool has_connected() const {
|
|
return has_connected_;
|
|
}
|
|
|
|
//sync call
|
|
#if __cplusplus > 201402L
|
|
template<size_t TIMEOUT, typename T = void, typename... Args>
|
|
auto call(const std::string& rpc_name, Args&& ... args) {
|
|
std::future<req_result> future = async_call<FUTURE>(rpc_name, std::forward<Args>(args)...);
|
|
auto status = future.wait_for(std::chrono::milliseconds(TIMEOUT));
|
|
if (status == std::future_status::timeout || status == std::future_status::deferred) {
|
|
throw std::out_of_range("timeout or deferred");
|
|
}
|
|
|
|
if constexpr (std::is_void_v<T>) {
|
|
future.get().as();
|
|
}
|
|
else {
|
|
return future.get().as<T>();
|
|
}
|
|
}
|
|
|
|
template<typename T = void, typename... Args>
|
|
auto call(const std::string& rpc_name, Args&& ... args) {
|
|
return call<DEFAULT_TIMEOUT, T>(rpc_name, std::forward<Args>(args)...);
|
|
}
|
|
#else
|
|
template<size_t TIMEOUT, typename T=void, typename... Args>
|
|
typename std::enable_if<std::is_void<T>::value>::type call(const std::string& rpc_name, Args&& ... args) {
|
|
std::future<req_result> future = async_call<FUTURE>(rpc_name, std::forward<Args>(args)...);
|
|
auto status = future.wait_for(std::chrono::milliseconds(TIMEOUT));
|
|
if (status == std::future_status::timeout || status == std::future_status::deferred) {
|
|
throw std::out_of_range("timeout or deferred");
|
|
}
|
|
|
|
future.get().as();
|
|
}
|
|
|
|
template<typename T = void, typename... Args>
|
|
typename std::enable_if<std::is_void<T>::value>::type call(const std::string& rpc_name, Args&& ... args) {
|
|
call<DEFAULT_TIMEOUT, T>(rpc_name, std::forward<Args>(args)...);
|
|
}
|
|
|
|
template<size_t TIMEOUT, typename T, typename... Args>
|
|
typename std::enable_if<!std::is_void<T>::value, T>::type call(const std::string& rpc_name, Args&& ... args) {
|
|
std::future<req_result> future = async_call<FUTURE>(rpc_name, std::forward<Args>(args)...);
|
|
auto status = future.wait_for(std::chrono::milliseconds(TIMEOUT));
|
|
if (status == std::future_status::timeout || status == std::future_status::deferred) {
|
|
throw std::out_of_range("timeout or deferred");
|
|
}
|
|
|
|
return future.get().as<T>();
|
|
}
|
|
|
|
template<typename T, typename... Args>
|
|
typename std::enable_if<!std::is_void<T>::value, T>::type call(const std::string& rpc_name, Args&& ... args) {
|
|
return call<DEFAULT_TIMEOUT, T>(rpc_name, std::forward<Args>(args)...);
|
|
}
|
|
#endif
|
|
|
|
template<CallModel model, typename... Args>
|
|
std::future<req_result> async_call(const std::string& rpc_name, Args&&... args) {
|
|
auto p = std::make_shared<std::promise<req_result>>();
|
|
std::future<req_result> future = p->get_future();
|
|
|
|
uint64_t fu_id = 0;
|
|
{
|
|
std::unique_lock<std::mutex> lock(cb_mtx_);
|
|
fu_id_++;
|
|
fu_id = fu_id_;
|
|
future_map_.emplace(fu_id, std::move(p));
|
|
}
|
|
|
|
msgpack_codec codec;
|
|
auto ret = codec.pack_args(rpc_name, std::forward<Args>(args)...);
|
|
write(fu_id, request_type::req_res, std::move(ret));
|
|
return future;
|
|
}
|
|
|
|
template<size_t TIMEOUT = DEFAULT_TIMEOUT, typename... Args>
|
|
void async_call(const std::string& rpc_name, std::function<void(boost::system::error_code, string_view)> cb, Args&& ... args) {
|
|
uint64_t cb_id = 0;
|
|
{
|
|
std::unique_lock<std::mutex> lock(cb_mtx_);
|
|
callback_id_++;
|
|
callback_id_ |= (uint64_t(1) << 63);
|
|
cb_id = callback_id_;
|
|
auto call = std::make_shared<call_t>(ios_, std::move(cb), TIMEOUT);
|
|
call->start_timer();
|
|
callback_map_.emplace(cb_id, call);
|
|
}
|
|
|
|
msgpack_codec codec;
|
|
auto ret = codec.pack_args(rpc_name, std::forward<Args>(args)...);
|
|
write(cb_id, request_type::req_res, std::move(ret));
|
|
}
|
|
|
|
void stop() {
|
|
if (thd_ != nullptr) {
|
|
ios_.stop();
|
|
thd_->join();
|
|
thd_ = nullptr;
|
|
}
|
|
}
|
|
|
|
template<typename Func>
|
|
void subscribe(std::string key, Func f) {
|
|
auto it = sub_map_.find(key);
|
|
if (it != sub_map_.end()) {
|
|
assert("duplicated subscribe");
|
|
return;
|
|
}
|
|
|
|
sub_map_.emplace(key, std::move(f));
|
|
msgpack_codec codec;
|
|
auto ret = codec.pack_args(key, "");
|
|
write(0, request_type::sub_pub, std::move(ret));
|
|
}
|
|
|
|
template<typename Func>
|
|
void subscribe(std::string key, std::string token, Func f) {
|
|
auto composite_key = key + token;
|
|
auto it = sub_map_.find(composite_key);
|
|
if (it != sub_map_.end()) {
|
|
assert("duplicated subscribe");
|
|
return;
|
|
}
|
|
|
|
sub_map_.emplace(std::move(composite_key), std::move(f));
|
|
msgpack_codec codec;
|
|
auto ret = codec.pack_args(key, token);
|
|
write(0, request_type::sub_pub, std::move(ret));
|
|
}
|
|
|
|
template<typename T, size_t TIMEOUT = 3>
|
|
void publish(std::string key, T&& t) {
|
|
call<TIMEOUT>("publish", std::move(key), "", std::forward<T>(t));
|
|
}
|
|
|
|
template<typename T, size_t TIMEOUT=3>
|
|
void publish_by_token(std::string key, std::string token, T&& t) {
|
|
call<TIMEOUT>("publish_by_token", std::move(key), std::move(token), std::forward<T>(t));
|
|
}
|
|
|
|
private:
|
|
void async_connect() {
|
|
assert(port_ != 0);
|
|
auto addr = boost::asio::ip::address::from_string(host_);
|
|
socket_.async_connect({ addr, port_ }, [this](const boost::system::error_code& ec) {
|
|
if (has_connected_) {
|
|
return;
|
|
}
|
|
|
|
if (ec) {
|
|
//std::cout << ec.message() << std::endl;
|
|
|
|
has_connected_ = false;
|
|
|
|
if (reconnect_cnt_ == 0) {
|
|
return;
|
|
}
|
|
|
|
if (reconnect_cnt_ > 0) {
|
|
reconnect_cnt_--;
|
|
}
|
|
|
|
async_reconnect();
|
|
}
|
|
else {
|
|
//std::cout<<"connected ok"<<std::endl;
|
|
has_connected_ = true;
|
|
do_read();
|
|
|
|
if (has_wait_)
|
|
conn_cond_.notify_one();
|
|
}
|
|
});
|
|
}
|
|
|
|
void async_reconnect() {
|
|
reset_socket();
|
|
async_connect();
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(connect_timeout_));
|
|
}
|
|
|
|
void reset_deadline_timer(size_t timeout) {
|
|
deadline_.expires_from_now(std::chrono::seconds(timeout));
|
|
deadline_.async_wait([this, timeout](const boost::system::error_code& ec) {
|
|
if (!ec) {
|
|
if (has_connected_) {
|
|
write(0, request_type::req_res, buffer_type(0));
|
|
}
|
|
}
|
|
|
|
reset_deadline_timer(timeout);
|
|
});
|
|
}
|
|
|
|
void write(std::uint64_t req_id, request_type type, buffer_type&& message) {
|
|
size_t size = message.size();
|
|
assert(size < MAX_BUF_LEN);
|
|
client_message_type msg{ req_id, type, {message.release(), size} };
|
|
|
|
std::unique_lock<std::mutex> lock(write_mtx_);
|
|
outbox_.emplace_back(std::move(msg));
|
|
if (outbox_.size() > 1) {
|
|
// outstanding async_write
|
|
return;
|
|
}
|
|
|
|
write();
|
|
}
|
|
|
|
void write() {
|
|
auto& msg = outbox_[0];
|
|
write_size_ = (uint32_t)msg.content.length();
|
|
std::array<boost::asio::const_buffer, 4> write_buffers;
|
|
write_buffers[0] = boost::asio::buffer(&write_size_, sizeof(int32_t));
|
|
write_buffers[1] = boost::asio::buffer(&msg.req_id, sizeof(uint64_t));
|
|
write_buffers[2] = boost::asio::buffer(&msg.req_type, sizeof(request_type));
|
|
write_buffers[3] = boost::asio::buffer((char*)msg.content.data(), write_size_);
|
|
|
|
boost::asio::async_write(socket_, write_buffers,
|
|
[this](const boost::system::error_code& ec, const size_t length) {
|
|
if (ec) {
|
|
has_connected_ = false;
|
|
close();
|
|
error_callback(ec);
|
|
|
|
return;
|
|
}
|
|
|
|
std::unique_lock<std::mutex> lock(write_mtx_);
|
|
if (outbox_.empty()) {
|
|
return;
|
|
}
|
|
|
|
::free((char*)outbox_.front().content.data());
|
|
outbox_.pop_front();
|
|
|
|
if (!outbox_.empty()) {
|
|
// more messages to send
|
|
this->write();
|
|
}
|
|
});
|
|
}
|
|
|
|
void do_read() {
|
|
boost::asio::async_read(socket_, boost::asio::buffer(head_),
|
|
[this](const boost::system::error_code& ec, const size_t length) {
|
|
if (!socket_.is_open()) {
|
|
//LOG(INFO) << "socket already closed";
|
|
has_connected_ = false;
|
|
return;
|
|
}
|
|
|
|
if (!ec) {
|
|
//const uint32_t body_len = *((uint32_t*)(head_));
|
|
//auto req_id = *((std::uint64_t*)(head_ + sizeof(int32_t)));
|
|
//auto req_type = *(request_type*)(head_ + sizeof(int32_t) + sizeof(int64_t));
|
|
rpc_header* header = (rpc_header*)(head_);
|
|
const uint32_t body_len = header->body_len;
|
|
if (body_len > 0 && body_len < MAX_BUF_LEN) {
|
|
if (body_.size() < body_len) { body_.resize(body_len); }
|
|
read_body(header->req_id, header->req_type, body_len);
|
|
return;
|
|
}
|
|
|
|
if (body_len == 0 || body_len > MAX_BUF_LEN) {
|
|
//LOG(INFO) << "invalid body len";
|
|
close();
|
|
if (err_cb_) {
|
|
err_cb_(asio::error::make_error_code(asio::error::message_size));
|
|
}
|
|
return;
|
|
}
|
|
}
|
|
else {
|
|
//LOG(INFO) << ec.message();
|
|
has_connected_ = false;
|
|
close();
|
|
error_callback(ec);
|
|
}
|
|
});
|
|
}
|
|
|
|
void read_body(std::uint64_t req_id, request_type req_type, size_t body_len) {
|
|
boost::asio::async_read(
|
|
socket_, boost::asio::buffer(body_.data(), body_len),
|
|
[this, req_id, req_type, body_len](boost::system::error_code ec, std::size_t length) {
|
|
//cancel_timer();
|
|
|
|
if (!socket_.is_open()) {
|
|
//LOG(INFO) << "socket already closed";
|
|
call_back(req_id, asio::error::make_error_code(asio::error::connection_aborted), {});
|
|
return;
|
|
}
|
|
|
|
if (!ec) {
|
|
//entier body
|
|
if (req_type == request_type::req_res) {
|
|
call_back(req_id, ec, { body_.data(), body_len });
|
|
}
|
|
else if (req_type == request_type::sub_pub) {
|
|
callback_sub(ec, { body_.data(), body_len });
|
|
}
|
|
else {
|
|
close();
|
|
if (err_cb_) {
|
|
err_cb_(asio::error::make_error_code(asio::error::invalid_argument));
|
|
}
|
|
return;
|
|
}
|
|
|
|
do_read();
|
|
}
|
|
else {
|
|
//LOG(INFO) << ec.message();
|
|
has_connected_ = false;
|
|
close();
|
|
error_callback(ec);
|
|
}
|
|
});
|
|
}
|
|
|
|
void call_back(uint64_t req_id, const boost::system::error_code& ec, string_view data) {
|
|
temp_req_id_ = req_id;
|
|
auto cb_flag = req_id >> 63;
|
|
if (cb_flag) {
|
|
std::shared_ptr<call_t> cl = nullptr;
|
|
{
|
|
std::unique_lock<std::mutex> lock(cb_mtx_);
|
|
cl = std::move(callback_map_[req_id]);
|
|
}
|
|
|
|
assert(cl);
|
|
if (!cl->has_timeout()) {
|
|
cl->cancel();
|
|
cl->callback(ec, data);
|
|
}
|
|
else {
|
|
cl->callback(asio::error::make_error_code(asio::error::timed_out), {});
|
|
}
|
|
|
|
std::unique_lock<std::mutex> lock(cb_mtx_);
|
|
callback_map_.erase(req_id);
|
|
}
|
|
else {
|
|
std::unique_lock<std::mutex> lock(cb_mtx_);
|
|
auto& f = future_map_[req_id];
|
|
if (ec) {
|
|
//LOG<<ec.message();
|
|
if (!f) {
|
|
//std::cout << "invalid req_id" << std::endl;
|
|
return;
|
|
}
|
|
}
|
|
|
|
assert(f);
|
|
f->set_value(req_result{ data });
|
|
future_map_.erase(req_id);
|
|
}
|
|
}
|
|
|
|
void callback_sub(const boost::system::error_code& ec, string_view result) {
|
|
rpc_service::msgpack_codec codec;
|
|
try {
|
|
auto tp = codec.unpack<std::tuple<int, std::string, std::string>>(result.data(), result.size());
|
|
auto code = std::get<0>(tp);
|
|
auto& key = std::get<1>(tp);
|
|
auto& data = std::get<2>(tp);
|
|
|
|
auto it = sub_map_.find(key);
|
|
if (it == sub_map_.end()) {
|
|
return;
|
|
}
|
|
|
|
it->second(data);
|
|
}
|
|
catch (const std::exception& ex) {
|
|
if (err_cb_) {
|
|
err_cb_(asio::error::make_error_code(asio::error::invalid_argument));
|
|
}
|
|
std::cout << ex.what() << "\n";
|
|
}
|
|
}
|
|
|
|
void clear_cache() {
|
|
{
|
|
std::unique_lock<std::mutex> lock(write_mtx_);
|
|
while (!outbox_.empty()) {
|
|
::free((char*)outbox_.front().content.data());
|
|
outbox_.pop_front();
|
|
}
|
|
}
|
|
|
|
{
|
|
std::unique_lock<std::mutex> lock(cb_mtx_);
|
|
callback_map_.clear();
|
|
future_map_.clear();
|
|
}
|
|
}
|
|
|
|
void reset_socket(){
|
|
boost::system::error_code igored_ec;
|
|
socket_.close(igored_ec);
|
|
socket_ = decltype(socket_)(ios_);
|
|
if(!socket_.is_open()){
|
|
socket_.open(boost::asio::ip::tcp::v4());
|
|
}
|
|
}
|
|
|
|
class call_t : asio::noncopyable, public std::enable_shared_from_this<call_t> {
|
|
public:
|
|
call_t(asio::io_service& ios, std::function<void(boost::system::error_code, string_view)> cb, size_t timeout) : timer_(ios),
|
|
cb_(std::move(cb)), timeout_(timeout){
|
|
}
|
|
|
|
void start_timer() {
|
|
if (timeout_ == 0) {
|
|
return;
|
|
}
|
|
|
|
timer_.expires_from_now(std::chrono::milliseconds(timeout_));
|
|
auto self = this->shared_from_this();
|
|
timer_.async_wait([this, self](boost::system::error_code ec) {
|
|
if (ec) {
|
|
return;
|
|
}
|
|
|
|
has_timeout_ = true;
|
|
});
|
|
}
|
|
|
|
void callback(boost::system::error_code ec, string_view data) {
|
|
cb_(ec, data);
|
|
}
|
|
|
|
bool has_timeout() const {
|
|
return has_timeout_;
|
|
}
|
|
|
|
void cancel() {
|
|
if (timeout_ == 0) {
|
|
return;
|
|
}
|
|
|
|
boost::system::error_code ec;
|
|
timer_.cancel(ec);
|
|
}
|
|
|
|
private:
|
|
boost::asio::steady_timer timer_;
|
|
std::function<void(boost::system::error_code, string_view)> cb_;
|
|
size_t timeout_;
|
|
bool has_timeout_ = false;
|
|
};
|
|
|
|
void error_callback(const boost::system::error_code& ec) {
|
|
if (err_cb_) {
|
|
err_cb_(ec);
|
|
}
|
|
|
|
if (enable_reconnect_) {
|
|
async_connect();
|
|
}
|
|
}
|
|
|
|
void set_default_error_cb() {
|
|
err_cb_ = [this](boost::system::error_code){
|
|
async_connect();
|
|
};
|
|
}
|
|
|
|
boost::asio::io_service ios_;
|
|
asio::ip::tcp::socket socket_;
|
|
boost::asio::io_service::work work_;
|
|
std::shared_ptr<std::thread> thd_ = nullptr;
|
|
|
|
std::string host_;
|
|
unsigned short port_ = 0;
|
|
size_t connect_timeout_ = 1000;//s
|
|
int reconnect_cnt_ = -1;
|
|
std::atomic_bool has_connected_ = { false };
|
|
std::mutex conn_mtx_;
|
|
std::condition_variable conn_cond_;
|
|
bool has_wait_ = false;
|
|
|
|
asio::steady_timer deadline_;
|
|
|
|
struct client_message_type {
|
|
std::uint64_t req_id;
|
|
request_type req_type;
|
|
string_view content;
|
|
};
|
|
std::deque<client_message_type> outbox_;
|
|
uint32_t write_size_ = 0;
|
|
std::mutex write_mtx_;
|
|
uint64_t fu_id_ = 0;
|
|
std::function<void(boost::system::error_code)> err_cb_;
|
|
bool enable_reconnect_ = false;
|
|
|
|
std::unordered_map<std::uint64_t, std::shared_ptr<std::promise<req_result>>> future_map_;
|
|
std::unordered_map<std::uint64_t, std::shared_ptr<call_t>> callback_map_;
|
|
std::mutex cb_mtx_;
|
|
uint64_t callback_id_ = 0;
|
|
|
|
uint64_t temp_req_id_ = 0;
|
|
|
|
char head_[HEAD_LEN] = {};
|
|
std::vector<char> body_;
|
|
|
|
std::unordered_map<std::string, std::function<void(string_view)>> sub_map_;
|
|
};
|
|
} |