saberbud/rest_rpc
1
0
Fork 0
mirror of https://github.com/qicosmos/rest_rpc.git synced 2026-08-29 08:34:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

format code with clang-format11

Browse Source
This commit is contained in:
qicosmos
2021-08-20 20:16:05 +08:00
parent 7735bf3678
commit 7e62c3238b
17 changed files with 2839 additions and 2757 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/client/main.cpp
+497 -493
View File
File diff suppressed because it is too large Load Diff
examples/server/main.cpp
+114 -110
View File
@@ -5,153 +5,157 @@ using namespace rpc_service;
#include "qps.h"
struct dummy{
int add(rpc_conn conn, int a, int b) {
auto shared_conn = conn.lock();
if (shared_conn) {
shared_conn->set_user_data(std::string("aa"));
auto s = conn.lock()->get_user_data<std::string>();
std::cout << s << '\n'; //aa
}
struct dummy {
int add(rpc_conn conn, int a, int b) {
auto shared_conn = conn.lock();
if (shared_conn) {
shared_conn->set_user_data(std::string("aa"));
auto s = conn.lock()->get_user_data<std::string>();
std::cout << s << '\n'; // aa
}
return a + b;
}
return a + b;
}
};
std::string translate(rpc_conn conn, const std::string& orignal) {
std::string temp = orignal;
for (auto& c : temp) {
c = std::toupper(c);
}
return temp;
std::string translate(rpc_conn conn, const std::string &orignal) {
std::string temp = orignal;
for (auto &c : temp) {
c = std::toupper(c);
}
return temp;
}
void hello(rpc_conn conn, const std::string& str) {
std::cout << "hello " << str << std::endl;
void hello(rpc_conn conn, const std::string &str) {
std::cout << "hello " << str << std::endl;
}
struct person {
int id;
std::string name;
int age;
int id;
std::string name;
int age;
MSGPACK_DEFINE(id, name, age);
MSGPACK_DEFINE(id, name, age);
};
std::string get_person_name(rpc_conn conn, const person& p) {
return p.name;
std::string get_person_name(rpc_conn conn, const person &p) { return p.name; }
person get_person(rpc_conn conn) { return {1, "tom", 20}; }
void upload(rpc_conn conn, const std::string &filename,
const std::string &content) {
std::cout << content.size() << std::endl;
std::ofstream file(filename, std::ios::binary);
file.write(content.data(), content.size());
}
person get_person(rpc_conn conn) {
return { 1, "tom", 20 };
}
std::string download(rpc_conn conn, const std::string &filename) {
std::ifstream file(filename, std::ios::binary);
if (!file) {
return "";
}
void upload(rpc_conn conn, const std::string& filename, const std::string& content) {
std::cout << content.size() << std::endl;
std::ofstream file(filename, std::ios::binary);
file.write(content.data(), content.size());
}
file.seekg(0, std::ios::end);
size_t file_len = file.tellg();
file.seekg(0, std::ios::beg);
std::string content;
content.resize(file_len);
file.read(&content[0], file_len);
std::cout << file_len << std::endl;
std::string download(rpc_conn conn, const std::string& filename) {
std::ifstream file(filename, std::ios::binary);
if (!file) {
return "";
}
file.seekg(0, std::ios::end);
size_t file_len = file.tellg();
file.seekg(0, std::ios::beg);
std::string content;
content.resize(file_len);
file.read(&content[0], file_len);
std::cout << file_len << std::endl;
return content;
return content;
}
qps g_qps;
std::string get_name(rpc_conn conn, const person& p) {
g_qps.increase();
return p.name;
}
//if you want to response later, you can use async model, you can control when to response
void async_echo(rpc_conn conn, const std::string& src) {
auto req_id = conn.lock()->request_id();//note: you need keep the request id at that time, and pass it into the async thread
std::thread thd([conn, req_id, src] {
std::this_thread::sleep_for(std::chrono::seconds(1));
auto conn_sp = conn.lock();
if (conn_sp) {
conn_sp->pack_and_response(req_id, std::move(src));
}
});
thd.detach();
}
std::string echo(rpc_conn conn, const std::string& src) {
std::string get_name(rpc_conn conn, const person &p) {
g_qps.increase();
return src;
return p.name;
}
int get_int(rpc_conn conn, int val) {
return val;
// if you want to response later, you can use async model, you can control when
// to response
void async_echo(rpc_conn conn, const std::string &src) {
auto req_id =
conn.lock()->request_id(); // note: you need keep the request id at that
// time, and pass it into the async thread
std::thread thd([conn, req_id, src] {
std::this_thread::sleep_for(std::chrono::seconds(1));
auto conn_sp = conn.lock();
if (conn_sp) {
conn_sp->pack_and_response(req_id, std::move(src));
}
});
thd.detach();
}
std::string echo(rpc_conn conn, const std::string &src) {
g_qps.increase();
return src;
}
int get_int(rpc_conn conn, int val) { return val; }
void test_ssl() {
rpc_server server(9000, std::thread::hardware_concurrency(), { "server.crt", "server.key" });
server.register_handler("hello", hello);
server.register_handler("echo", echo);
server.run();
rpc_server server(9000, std::thread::hardware_concurrency(),
{"server.crt", "server.key"});
server.register_handler("hello", hello);
server.register_handler("echo", echo);
server.run();
}
void benchmark_test(){
void benchmark_test() {
rpc_server server(9000, std::thread::hardware_concurrency());
server.register_handler("echo", echo);
server.run();
}
int main() {
// benchmark_test();
rpc_server server(9000, std::thread::hardware_concurrency());
// benchmark_test();
rpc_server server(9000, std::thread::hardware_concurrency());
dummy d;
server.register_handler("add", &dummy::add, &d);
server.register_handler("translate", translate);
server.register_handler("hello", hello);
server.register_handler("get_person_name", get_person_name);
server.register_handler("get_person", get_person);
server.register_handler("upload", upload);
server.register_handler("download", download);
server.register_handler("get_name", get_name);
server.register_handler<Async>("async_echo", async_echo);
server.register_handler("echo", echo);
server.register_handler("get_int", get_int);
dummy d;
server.register_handler("add", &dummy::add, &d);
server.register_handler("translate", translate);
server.register_handler("hello", hello);
server.register_handler("get_person_name", get_person_name);
server.register_handler("get_person", get_person);
server.register_handler("upload", upload);
server.register_handler("download", download);
server.register_handler("get_name", get_name);
server.register_handler<Async>("async_echo", async_echo);
server.register_handler("echo", echo);
server.register_handler("get_int", get_int);
server.register_handler("publish_by_token", [&server](rpc_conn conn, std::string key, std::string token, std::string val) {
server.publish_by_token(std::move(key), std::move(token), std::move(val));
});
server.register_handler("publish_by_token", [&server](rpc_conn conn,
std::string key,
std::string token,
std::string val) {
server.publish_by_token(std::move(key), std::move(token), std::move(val));
});
server.register_handler("publish", [&server](rpc_conn conn, std::string key, std::string token, std::string val) {
server.publish(std::move(key), std::move(val));
});
server.register_handler("publish",
[&server](rpc_conn conn, std::string key,
std::string token, std::string val) {
server.publish(std::move(key), std::move(val));
});
std::thread thd([&server] {
person p{ 1, "tom", 20 };
while (true) {
server.publish("key", "hello subscriber");
auto list = server.get_token_list();
for (auto& token : list) {
server.publish_by_token("key", token, p);
server.publish_by_token("key1", token, "hello subscriber1");
}
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
});
std::thread thd([&server] {
person p{1, "tom", 20};
while (true) {
server.publish("key", "hello subscriber");
auto list = server.get_token_list();
for (auto &token : list) {
server.publish_by_token("key", token, p);
server.publish_by_token("key1", token, "hello subscriber1");
}
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
});
server.run();
server.run();
std::string str;
std::cin >> str;
std::string str;
std::cin >> str;
}
examples/server/qps.h
+19 -20
View File
@@ -1,31 +1,30 @@
#pragma once
#include <atomic>
#include <thread>
#include <chrono>
#include <thread>
class qps {
public:
void increase() {
counter_.fetch_add(1, std::memory_order_release);
}
void increase() { counter_.fetch_add(1, std::memory_order_release); }
qps() : counter_(0) {
thd_ = std::thread([this] {
while (!stop_) {
std::cout << "qps: " << counter_.load(std::memory_order_acquire) << '\n';
std::this_thread::sleep_for(std::chrono::seconds(1));
//counter_.store(0, std::memory_order_release);
}
});
}
qps() : counter_(0) {
thd_ = std::thread([this] {
while (!stop_) {
std::cout << "qps: " << counter_.load(std::memory_order_acquire)
<< '\n';
std::this_thread::sleep_for(std::chrono::seconds(1));
// counter_.store(0, std::memory_order_release);
}
});
}
~qps() {
stop_ = true;
thd_.join();
}
~qps() {
stop_ = true;
thd_.join();
}
private:
bool stop_ = false;
std::thread thd_;
std::atomic<uint32_t> counter_;
bool stop_ = false;
std::thread thd_;
std::atomic<uint32_t> counter_;
};
include/rest_rpc.hpp
+1 -1
View File
@@ -1,2 +1,2 @@
#include "rest_rpc/rpc_server.h"
#include "rest_rpc/rpc_client.hpp"
#include "rest_rpc/rpc_server.h"
include/rest_rpc/client_util.hpp
+29 -31
View File
@@ -16,37 +16,35 @@ using string_view = boost::string_view;
#include "codec.h"
namespace rest_rpc {
inline bool has_error(string_view result) {
if (result.empty()) {
return true;
}
inline bool has_error(string_view result) {
if (result.empty()) {
return true;
}
rpc_service::msgpack_codec codec;
auto tp = codec.unpack<std::tuple<int>>(result.data(), result.size());
rpc_service::msgpack_codec codec;
auto tp = codec.unpack<std::tuple<int>>(result.data(), result.size());
return std::get<0>(tp) != 0;
}
template<typename T>
inline T get_result(string_view result) {
rpc_service::msgpack_codec codec;
auto tp = codec.unpack<std::tuple<int, T>>(result.data(), result.size());
return std::get<1>(tp);
}
inline std::string get_error_msg(string_view result) {
rpc_service::msgpack_codec codec;
auto tp = codec.unpack<std::tuple<int, std::string>>(result.data(), result.size());
return std::get<1>(tp);
}
template<typename T>
inline T as(string_view result) {
if (has_error(result)) {
throw std::logic_error(get_error_msg(result));
}
return get_result<T>(result);
}
return std::get<0>(tp) != 0;
}
template <typename T> inline T get_result(string_view result) {
rpc_service::msgpack_codec codec;
auto tp = codec.unpack<std::tuple<int, T>>(result.data(), result.size());
return std::get<1>(tp);
}
inline std::string get_error_msg(string_view result) {
rpc_service::msgpack_codec codec;
auto tp =
codec.unpack<std::tuple<int, std::string>>(result.data(), result.size());
return std::get<1>(tp);
}
template <typename T> inline T as(string_view result) {
if (has_error(result)) {
throw std::logic_error(get_error_msg(result));
}
return get_result<T>(result);
}
} // namespace rest_rpc
include/rest_rpc/codec.h
+21 -21
View File
@@ -6,44 +6,44 @@
namespace rest_rpc {
namespace rpc_service {
using buffer_type = msgpack::sbuffer;
struct msgpack_codec {
using buffer_type = msgpack::sbuffer;
struct msgpack_codec {
const static size_t init_size = 2 * 1024;
template<typename... Args>
static buffer_type pack_args(Args&&... args) {
buffer_type buffer(init_size);
template <typename... Args> static buffer_type pack_args(Args &&...args) {
buffer_type buffer(init_size);
msgpack::pack(buffer, std::forward_as_tuple(std::forward<Args>(args)...));
return buffer;
}
template<typename Arg, typename... Args,
typename = typename std::enable_if<std::is_enum<Arg>::value>::type>
static std::string pack_args_str(Arg arg, Args&&... args) {
buffer_type buffer(init_size);
msgpack::pack(buffer, std::forward_as_tuple((int)arg, std::forward<Args>(args)...));
return std::string(buffer.data(), buffer.size());
template <typename Arg, typename... Args,
typename = typename std::enable_if<std::is_enum<Arg>::value>::type>
static std::string pack_args_str(Arg arg, Args &&...args) {
buffer_type buffer(init_size);
msgpack::pack(buffer,
std::forward_as_tuple((int)arg, std::forward<Args>(args)...));
return std::string(buffer.data(), buffer.size());
}
template<typename T>
buffer_type pack(T&& t) const {
buffer_type buffer;
template <typename T> buffer_type pack(T &&t) const {
buffer_type buffer;
msgpack::pack(buffer, std::forward<T>(t));
return buffer;
}
template<typename T>
T unpack(char const* data, size_t length) {
template <typename T> T unpack(char const *data, size_t length) {
try {
msgpack::unpack(msg_, data, length);
return msg_.get().as<T>();
} catch (...) { throw std::invalid_argument("unpack failed: Args not match!"); }
} catch (...) {
throw std::invalid_argument("unpack failed: Args not match!");
}
}
private:
private:
msgpack::unpacked msg_;
};
} // namespace rpc_service
} // namespace rest_rpc
} // namespace rpc_service
} // namespace rest_rpc
#endif // REST_RPC_CODEC_H_
#endif // REST_RPC_CODEC_H_
include/rest_rpc/connection.h
+317 -306
View File
@@ -1,383 +1,394 @@
#ifndef REST_RPC_CONNECTION_H_
#define REST_RPC_CONNECTION_H_
#include <iostream>
#include <memory>
#include <array>
#include <deque>
#include "use_asio.hpp"
#include "const_vars.h"
#include "router.h"
#include "cplusplus_14.h"
#include "nonstd_any.hpp"
#include "router.h"
#include "use_asio.hpp"
#include <array>
#include <deque>
#include <iostream>
#include <memory>
using boost::asio::ip::tcp;
namespace rest_rpc {
namespace rpc_service {
struct ssl_configure {
std::string cert_file;
std::string key_file;
};
namespace rpc_service {
struct ssl_configure {
std::string cert_file;
std::string key_file;
};
class connection : public std::enable_shared_from_this<connection>, private asio::noncopyable {
public:
connection(boost::asio::io_service& io_service, std::size_t timeout_seconds, router& router)
: socket_(io_service),
body_(INIT_BUF_SIZE),
timer_(io_service),
timeout_seconds_(timeout_seconds),
has_closed_(false),
router_(router){
}
class connection : public std::enable_shared_from_this<connection>,
private asio::noncopyable {
public:
connection(boost::asio::io_service &io_service, std::size_t timeout_seconds,
router &router)
: socket_(io_service), body_(INIT_BUF_SIZE), timer_(io_service),
timeout_seconds_(timeout_seconds), has_closed_(false), router_(router) {
}
~connection() {
close();
}
~connection() { close(); }
void start() {
if (is_ssl() && !has_shake_) {
async_handshake();
}
else {
read_head();
}
}
void start() {
if (is_ssl() && !has_shake_) {
async_handshake();
} else {
read_head();
}
}
tcp::socket& socket() { return socket_; }
tcp::socket &socket() { return socket_; }
bool has_closed() const { return has_closed_; }
uint64_t request_id() const {
return req_id_;
}
bool has_closed() const { return has_closed_; }
uint64_t request_id() const { return req_id_; }
void response(uint64_t req_id, std::string data, request_type req_type = request_type::req_res) {
auto len = data.size();
assert(len < MAX_BUF_LEN);
void response(uint64_t req_id, std::string data,
request_type req_type = request_type::req_res) {
auto len = data.size();
assert(len < MAX_BUF_LEN);
std::unique_lock<std::mutex> lock(write_mtx_);
write_queue_.emplace_back(message_type{ req_id, req_type, std::make_shared<std::string>(std::move(data)) });
if (write_queue_.size() > 1) {
return;
}
std::unique_lock<std::mutex> lock(write_mtx_);
write_queue_.emplace_back(message_type{
req_id, req_type, std::make_shared<std::string>(std::move(data))});
if (write_queue_.size() > 1) {
return;
}
write();
}
write();
}
template<typename T>
void pack_and_response(uint64_t req_id, T data) {
auto result = msgpack_codec::pack_args_str(result_code::OK, std::move(data));
response(req_id, std::move(result));
}
template <typename T> void pack_and_response(uint64_t req_id, T data) {
auto result =
msgpack_codec::pack_args_str(result_code::OK, std::move(data));
response(req_id, std::move(result));
}
void set_conn_id(int64_t id) { conn_id_ = id; }
void set_conn_id(int64_t id) { conn_id_ = id; }
int64_t conn_id() const { return conn_id_; }
int64_t conn_id() const { return conn_id_; }
template<typename T>
void set_user_data(const T &data) {
user_data_ = data;
}
template <typename T> void set_user_data(const T &data) { user_data_ = data; }
template<typename T>
T get_user_data() {
return nonstd::any_cast<T>(user_data_);
}
template <typename T> T get_user_data() {
return nonstd::any_cast<T>(user_data_);
}
const std::vector<char>& body() const {
return body_;
}
const std::vector<char> &body() const { return body_; }
std::string remote_address() const {
if (has_closed_) {
return "";
}
std::string remote_address() const {
if (has_closed_) {
return "";
}
return socket_.remote_endpoint().address().to_string();
}
return socket_.remote_endpoint().address().to_string();
}
void publish(const std::string& key, const std::string& data) {
auto result = msgpack_codec::pack_args_str(result_code::OK, key, data);
response(0, std::move(result), request_type::sub_pub);
}
void publish(const std::string &key, const std::string &data) {
auto result = msgpack_codec::pack_args_str(result_code::OK, key, data);
response(0, std::move(result), request_type::sub_pub);
}
void set_callback(std::function<void(std::string, std::string, std::weak_ptr<connection>)> callback) {
callback_ = std::move(callback);
}
void set_callback(
std::function<void(std::string, std::string, std::weak_ptr<connection>)>
callback) {
callback_ = std::move(callback);
}
void init_ssl_context(const ssl_configure& ssl_conf) {
void init_ssl_context(const ssl_configure &ssl_conf) {
#ifdef CINATRA_ENABLE_SSL
unsigned long ssl_options = boost::asio::ssl::context::default_workarounds
| boost::asio::ssl::context::no_sslv2
| boost::asio::ssl::context::single_dh_use;
try {
boost::asio::ssl::context ssl_context(boost::asio::ssl::context::sslv23);
ssl_context.set_options(ssl_options);
ssl_context.set_password_callback([](std::size_t size,
boost::asio::ssl::context_base::password_purpose purpose) {return "123456"; });
unsigned long ssl_options = boost::asio::ssl::context::default_workarounds |
boost::asio::ssl::context::no_sslv2 |
boost::asio::ssl::context::single_dh_use;
try {
boost::asio::ssl::context ssl_context(boost::asio::ssl::context::sslv23);
ssl_context.set_options(ssl_options);
ssl_context.set_password_callback(
[](std::size_t size,
boost::asio::ssl::context_base::password_purpose purpose) {
return "123456";
});
boost::system::error_code ec;
if (rpcfs::exists(ssl_conf.cert_file, ec)) {
ssl_context.use_certificate_chain_file(ssl_conf.cert_file);
}
boost::system::error_code ec;
if (rpcfs::exists(ssl_conf.cert_file, ec)) {
ssl_context.use_certificate_chain_file(ssl_conf.cert_file);
}
if (rpcfs::exists(ssl_conf.key_file, ec))
ssl_context.use_private_key_file(ssl_conf.key_file, boost::asio::ssl::context::pem);
if (rpcfs::exists(ssl_conf.key_file, ec))
ssl_context.use_private_key_file(ssl_conf.key_file,
boost::asio::ssl::context::pem);
//ssl_context_callback(ssl_context);
ssl_stream_ = std::make_unique<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>>(socket_, ssl_context);
}
catch (const std::exception& e) {
print(e);
}
// ssl_context_callback(ssl_context);
ssl_stream_ = std::make_unique<
boost::asio::ssl::stream<boost::asio::ip::tcp::socket &>>(
socket_, ssl_context);
} catch (const std::exception &e) {
print(e);
}
#endif
}
private:
void read_head() {
reset_timer();
auto self(this->shared_from_this());
async_read_head(
[this, self](boost::system::error_code ec, std::size_t length) {
if (!socket_.is_open()) {
// LOG(INFO) << "socket already closed";
return;
}
if (!ec) {
// const uint32_t body_len = *((int*)(head_));
// req_id_ = *((std::uint64_t*)(head_ + sizeof(int32_t)));
rpc_header *header = (rpc_header *)(head_);
req_id_ = header->req_id;
const uint32_t body_len = header->body_len;
req_type_ = header->req_type;
if (body_len > 0 && body_len < MAX_BUF_LEN) {
if (body_.size() < body_len) {
body_.resize(body_len);
}
read_body(body_len);
return;
}
private:
void read_head() {
reset_timer();
auto self(this->shared_from_this());
async_read_head([this, self](boost::system::error_code ec, std::size_t length) {
if (!socket_.is_open()) {
//LOG(INFO) << "socket already closed";
return;
}
if (body_len == 0) { // nobody, just head, maybe as heartbeat.
cancel_timer();
read_head();
} else {
print("invalid body len");
close();
}
} else {
print(ec);
close();
}
});
}
if (!ec) {
//const uint32_t body_len = *((int*)(head_));
//req_id_ = *((std::uint64_t*)(head_ + sizeof(int32_t)));
rpc_header* header = (rpc_header*)(head_);
req_id_ = header->req_id;
const uint32_t body_len = header->body_len;
req_type_ = header->req_type;
if (body_len > 0 && body_len < MAX_BUF_LEN) {
if (body_.size() < body_len) { body_.resize(body_len); }
read_body(body_len);
return;
}
void read_body(std::size_t size) {
auto self(this->shared_from_this());
async_read(
size, [this, self](boost::system::error_code ec, std::size_t length) {
cancel_timer();
if (body_len == 0) { // nobody, just head, maybe as heartbeat.
cancel_timer();
read_head();
}
else {
print("invalid body len");
close();
}
}
else {
print(ec);
close();
}
});
}
if (!socket_.is_open()) {
// LOG(INFO) << "socket already closed";
return;
}
void read_body(std::size_t size) {
auto self(this->shared_from_this());
async_read(size, [this, self](boost::system::error_code ec, std::size_t length) {
cancel_timer();
if (!ec) {
read_head();
if (req_type_ == request_type::req_res) {
router_.route<connection>(body_.data(), length,
this->shared_from_this());
} else if (req_type_ == request_type::sub_pub) {
try {
msgpack_codec codec;
auto p = codec.unpack<std::tuple<std::string, std::string>>(
body_.data(), length);
callback_(std::move(std::get<0>(p)), std::move(std::get<1>(p)),
this->shared_from_this());
} catch (const std::exception &ex) {
print(ex);
}
}
} else {
// LOG(INFO) << ec.message();
}
});
}
if (!socket_.is_open()) {
//LOG(INFO) << "socket already closed";
return;
}
void write() {
auto &msg = write_queue_.front();
write_size_ = (uint32_t)msg.content->size();
std::array<boost::asio::const_buffer, 4> write_buffers;
write_buffers[0] = boost::asio::buffer(&write_size_, sizeof(uint32_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(msg.content->data(), write_size_);
if (!ec) {
read_head();
if (req_type_ == request_type::req_res) {
router_.route<connection>(body_.data(), length, this->shared_from_this());
}
else if (req_type_ == request_type::sub_pub) {
try {
msgpack_codec codec;
auto p = codec.unpack<std::tuple<std::string, std::string>>(body_.data(), length);
callback_(std::move(std::get<0>(p)), std::move(std::get<1>(p)), this->shared_from_this());
}
catch (const std::exception& ex) {
print(ex);
}
}
}
else {
//LOG(INFO) << ec.message();
}
});
}
void write() {
auto& msg = write_queue_.front();
write_size_ = (uint32_t)msg.content->size();
std::array<boost::asio::const_buffer, 4> write_buffers;
write_buffers[0] = boost::asio::buffer(&write_size_, sizeof(uint32_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(msg.content->data(), write_size_);
auto self = this->shared_from_this();
async_write(write_buffers,
[this, self](boost::system::error_code ec, std::size_t length) {
on_write(ec, length);
});
}
void on_write(boost::system::error_code ec, std::size_t length) {
if (ec) {
print(ec);
close(false);
return;
}
if (has_closed()) { return; }
std::unique_lock<std::mutex> lock(write_mtx_);
write_queue_.pop_front();
if (!write_queue_.empty()) {
write();
}
}
void async_handshake() {
#ifdef CINATRA_ENABLE_SSL
auto self = this->shared_from_this();
ssl_stream_->async_handshake(boost::asio::ssl::stream_base::server,
[this, self](const boost::system::error_code& error) {
if (error) {
print(error);
close();
return;
}
has_shake_ = true;
read_head();
auto self = this->shared_from_this();
async_write(write_buffers,
[this, self](boost::system::error_code ec, std::size_t length) {
on_write(ec, length);
});
#endif
}
}
bool is_ssl() const {
void on_write(boost::system::error_code ec, std::size_t length) {
if (ec) {
print(ec);
close(false);
return;
}
if (has_closed()) {
return;
}
std::unique_lock<std::mutex> lock(write_mtx_);
write_queue_.pop_front();
if (!write_queue_.empty()) {
write();
}
}
void async_handshake() {
#ifdef CINATRA_ENABLE_SSL
return ssl_stream_ != nullptr;
auto self = this->shared_from_this();
ssl_stream_->async_handshake(
boost::asio::ssl::stream_base::server,
[this, self](const boost::system::error_code &error) {
if (error) {
print(error);
close();
return;
}
has_shake_ = true;
read_head();
});
#endif
}
bool is_ssl() const {
#ifdef CINATRA_ENABLE_SSL
return ssl_stream_ != nullptr;
#else
return false;
return false;
#endif
}
}
template<typename Handler>
void async_read_head(Handler handler) {
if (is_ssl()) {
template <typename Handler> void async_read_head(Handler handler) {
if (is_ssl()) {
#ifdef CINATRA_ENABLE_SSL
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
boost::asio::async_read(*ssl_stream_,
boost::asio::buffer(head_, HEAD_LEN),
std::move(handler));
#endif
}
else {
boost::asio::async_read(socket_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
}
}
} else {
boost::asio::async_read(socket_, boost::asio::buffer(head_, HEAD_LEN),
std::move(handler));
}
}
template<typename Handler>
void async_read(size_t size_to_read, Handler handler) {
if (is_ssl()) {
template <typename Handler>
void async_read(size_t size_to_read, Handler handler) {
if (is_ssl()) {
#ifdef CINATRA_ENABLE_SSL
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
boost::asio::async_read(*ssl_stream_,
boost::asio::buffer(body_.data(), size_to_read),
std::move(handler));
#endif
}
else {
boost::asio::async_read(socket_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
}
}
} else {
boost::asio::async_read(socket_,
boost::asio::buffer(body_.data(), size_to_read),
std::move(handler));
}
}
template<typename BufferType, typename Handler>
void async_write(const BufferType& buffers, Handler handler) {
if (is_ssl()) {
template <typename BufferType, typename Handler>
void async_write(const BufferType &buffers, Handler handler) {
if (is_ssl()) {
#ifdef CINATRA_ENABLE_SSL
boost::asio::async_write(*ssl_stream_, buffers, std::move(handler));
boost::asio::async_write(*ssl_stream_, buffers, std::move(handler));
#endif
}
else {
boost::asio::async_write(socket_, buffers, std::move(handler));
}
}
} else {
boost::asio::async_write(socket_, buffers, std::move(handler));
}
}
void reset_timer() {
if (timeout_seconds_ == 0) { return; }
void reset_timer() {
if (timeout_seconds_ == 0) {
return;
}
auto self(this->shared_from_this());
timer_.expires_from_now(std::chrono::seconds(timeout_seconds_));
timer_.async_wait([this, self](const boost::system::error_code& ec) {
if (has_closed()) { return; }
auto self(this->shared_from_this());
timer_.expires_from_now(std::chrono::seconds(timeout_seconds_));
timer_.async_wait([this, self](const boost::system::error_code &ec) {
if (has_closed()) {
return;
}
if (ec) { return; }
if (ec) {
return;
}
//LOG(INFO) << "rpc connection timeout";
close(false);
});
}
// LOG(INFO) << "rpc connection timeout";
close(false);
});
}
void cancel_timer() {
if (timeout_seconds_ == 0) { return; }
void cancel_timer() {
if (timeout_seconds_ == 0) {
return;
}
timer_.cancel();
}
timer_.cancel();
}
void close(bool close_ssl = true) {
void close(bool close_ssl = true) {
#ifdef CINATRA_ENABLE_SSL
if (close_ssl && ssl_stream_) {
boost::system::error_code ec;
ssl_stream_->shutdown(ec);
ssl_stream_ = nullptr;
}
if (close_ssl && ssl_stream_) {
boost::system::error_code ec;
ssl_stream_->shutdown(ec);
ssl_stream_ = nullptr;
}
#endif
if (has_closed_) {
return;
}
if (has_closed_) {
return;
}
boost::system::error_code ignored_ec;
socket_.shutdown(tcp::socket::shutdown_both, ignored_ec);
socket_.close(ignored_ec);
has_closed_ = true;
has_shake_ = false;
}
boost::system::error_code ignored_ec;
socket_.shutdown(tcp::socket::shutdown_both, ignored_ec);
socket_.close(ignored_ec);
has_closed_ = true;
has_shake_ = false;
}
template<typename... Args>
void print(Args... args) {
template <typename... Args> void print(Args... args) {
#ifdef _DEBUG
std::initializer_list<int>{( std::cout << args << ' ', 0)...};
std::cout << "\n";
std::initializer_list<int>{(std::cout << args << ' ', 0)...};
std::cout << "\n";
#endif
}
}
void print(const boost::system::error_code& ec) {
print(ec.value(), ec.message());
}
void print(const boost::system::error_code &ec) {
print(ec.value(), ec.message());
}
void print(const std::exception& ex) {
print(ex.what());
}
void print(const std::exception &ex) { print(ex.what()); }
tcp::socket socket_;
tcp::socket socket_;
#ifdef CINATRA_ENABLE_SSL
std::unique_ptr<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>> ssl_stream_ = nullptr;
std::unique_ptr<boost::asio::ssl::stream<boost::asio::ip::tcp::socket &>>
ssl_stream_ = nullptr;
#endif
bool has_shake_ = false;
char head_[HEAD_LEN];
std::vector<char> body_;
std::uint64_t req_id_;
request_type req_type_;
bool has_shake_ = false;
char head_[HEAD_LEN];
std::vector<char> body_;
std::uint64_t req_id_;
request_type req_type_;
uint32_t write_size_ = 0;
std::mutex write_mtx_;
uint32_t write_size_ = 0;
std::mutex write_mtx_;
asio::steady_timer timer_;
std::size_t timeout_seconds_;
int64_t conn_id_ = 0;
bool has_closed_;
asio::steady_timer timer_;
std::size_t timeout_seconds_;
int64_t conn_id_ = 0;
bool has_closed_;
std::deque<message_type> write_queue_;
std::function<void(std::string, std::string, std::weak_ptr<connection>)> callback_;
router& router_;
nonstd::any user_data_;
};
} // namespace rpc_service
} // namespace rest_rpc
std::deque<message_type> write_queue_;
std::function<void(std::string, std::string, std::weak_ptr<connection>)>
callback_;
router &router_;
nonstd::any user_data_;
};
} // namespace rpc_service
} // namespace rest_rpc
#endif // REST_RPC_CONNECTION_H_
#endif // REST_RPC_CONNECTION_H_
include/rest_rpc/const_vars.h
+29 -33
View File
@@ -3,41 +3,37 @@
namespace rest_rpc {
enum class result_code : std::int16_t {
OK = 0,
FAIL = 1,
};
enum class result_code : std::int16_t {
OK = 0,
FAIL = 1,
};
enum class error_code {
OK,
UNKNOWN,
FAIL,
TIMEOUT,
CANCEL,
BADCONNECTION,
};
enum class error_code {
OK,
UNKNOWN,
FAIL,
TIMEOUT,
CANCEL,
BADCONNECTION,
};
enum class request_type : uint8_t {
req_res,
sub_pub
};
enum class request_type : uint8_t { req_res, sub_pub };
struct message_type {
std::uint64_t req_id;
request_type req_type;
std::shared_ptr<std::string> content;
};
struct message_type {
std::uint64_t req_id;
request_type req_type;
std::shared_ptr<std::string> content;
};
#pragma pack(1)
struct rpc_header {
uint32_t body_len;
uint64_t req_id;
request_type req_type;
};
#pragma pack()
#pragma pack (1)
struct rpc_header {
uint32_t body_len;
uint64_t req_id;
request_type req_type;
};
#pragma pack ()
static const size_t MAX_BUF_LEN = 1048576 * 10;
static const size_t HEAD_LEN = 13;
static const size_t INIT_BUF_SIZE = 2 * 1024;
}
static const size_t MAX_BUF_LEN = 1048576 * 10;
static const size_t HEAD_LEN = 13;
static const size_t INIT_BUF_SIZE = 2 * 1024;
} // namespace rest_rpc
include/rest_rpc/cplusplus_14.h
+38 -47
View File
@@ -1,44 +1,37 @@
#ifndef REST_RPC_CPLUSPLUS_14_H_
#define REST_RPC_CPLUSPLUS_14_H_
#include <type_traits>
#include <memory>
#include <tuple>
#include <type_traits>
#if __cplusplus == 201103L
namespace std {
template<class T>
struct unique_if {
typedef unique_ptr<T> single_object;
};
template <class T> struct unique_if { typedef unique_ptr<T> single_object; };
template<class T>
struct unique_if<T[]> {
template <class T> struct unique_if<T[]> {
typedef unique_ptr<T[]> unknown_bound;
};
template<class T, size_t N>
struct unique_if<T[N]> {
template <class T, size_t N> struct unique_if<T[N]> {
typedef void known_bound;
};
template<class T, class... Args>
typename unique_if<T>::single_object make_unique(Args&&... args) {
template <class T, class... Args>
typename unique_if<T>::single_object make_unique(Args &&...args) {
return unique_ptr<T>(new T(forward<Args>(args)...));
}
template<class T>
typename unique_if<T>::unknown_bound make_unique(size_t n) {
template <class T> typename unique_if<T>::unknown_bound make_unique(size_t n) {
typedef typename remove_extent<T>::type U;
return unique_ptr<T>(new U[n]());
}
template<class T, class... Args>
typename unique_if<T>::known_bound make_unique(Args&&...) = delete;
template <class T, class... Args>
typename unique_if<T>::known_bound make_unique(Args &&...) = delete;
template<size_t... Ints>
struct index_sequence {
template <size_t... Ints> struct index_sequence {
using type = index_sequence;
using value_type = size_t;
static constexpr std::size_t size() noexcept { return sizeof...(Ints); }
@@ -46,63 +39,61 @@ struct index_sequence {
// --------------------------------------------------------------
template<class Sequence1, class Sequence2>
struct _merge_and_renumber;
template <class Sequence1, class Sequence2> struct _merge_and_renumber;
template<size_t... I1, size_t... I2>
template <size_t... I1, size_t... I2>
struct _merge_and_renumber<index_sequence<I1...>, index_sequence<I2...>>
: index_sequence<I1..., (sizeof...(I1) + I2)...> {};
// --------------------------------------------------------------
template<size_t N>
struct make_index_sequence : _merge_and_renumber<typename make_index_sequence<N / 2>::type,
typename make_index_sequence<N - N / 2>::type> {};
template <size_t N>
struct make_index_sequence
: _merge_and_renumber<typename make_index_sequence<N / 2>::type,
typename make_index_sequence<N - N / 2>::type> {};
template<>
struct make_index_sequence<0> : index_sequence<> {};
template<>
struct make_index_sequence<1> : index_sequence<0> {};
template <> struct make_index_sequence<0> : index_sequence<> {};
template <> struct make_index_sequence<1> : index_sequence<0> {};
template<typename... T>
template <typename... T>
using index_sequence_for = make_index_sequence<sizeof...(T)>;
template<bool B, class T = void>
template <bool B, class T = void>
using enable_if_t = typename enable_if<B, T>::type;
template<typename T>
using remove_const_t = typename remove_const<T>::type;
template <typename T> using remove_const_t = typename remove_const<T>::type;
template<typename T>
template <typename T>
using remove_reference_t = typename remove_reference<T>::type;
template<int I, typename T>
template <int I, typename T>
using tuple_element_t = typename tuple_element<I, T>::type;
template<typename T>
using decay_t = typename decay<T>::type;
template <typename T> using decay_t = typename decay<T>::type;
template<typename F, typename Tuple, size_t... Idx>
auto apply_helper(F&& f, Tuple&& tp, std::index_sequence<Idx...>)
template <typename F, typename Tuple, size_t... Idx>
auto apply_helper(F &&f, Tuple &&tp, std::index_sequence<Idx...>)
-> decltype(std::forward<F>(f)(std::get<Idx>(std::forward<Tuple>(tp))...)) {
return std::forward<F>(f)(std::get<Idx>(std::forward<Tuple>(tp))...);
}
template<typename F, typename Tuple>
auto apply(F&& f, Tuple&& tp)
-> decltype(apply_helper(std::forward<F>(f), std::forward<Tuple>(tp),
std::make_index_sequence<std::tuple_size<decay_t<Tuple>>::value>{})) {
return apply_helper(std::forward<F>(f), std::forward<Tuple>(tp),
std::make_index_sequence<std::tuple_size<decay_t<Tuple>>::value>{});
template <typename F, typename Tuple>
auto apply(F &&f, Tuple &&tp) -> decltype(apply_helper(
std::forward<F>(f), std::forward<Tuple>(tp),
std::make_index_sequence<std::tuple_size<decay_t<Tuple>>::value>{})) {
return apply_helper(
std::forward<F>(f), std::forward<Tuple>(tp),
std::make_index_sequence<std::tuple_size<decay_t<Tuple>>::value>{});
}
template<typename F, typename... Args>
auto invoke(F&& f, Args&&... args) -> decltype(std::forward<F>(f)(std::forward<Args>(args)...)) {
template <typename F, typename... Args>
auto invoke(F &&f, Args &&...args)
-> decltype(std::forward<F>(f)(std::forward<Args>(args)...)) {
return std::forward<F>(f)(std::forward<Args>(args)...);
}
} // namespace std
} // namespace std
#endif
#endif // REST_RPC_CPLUSPLUS_14_H_
#endif // REST_RPC_CPLUSPLUS_14_H_
include/rest_rpc/io_service_pool.h
+17 -14
View File
@@ -1,16 +1,17 @@
#ifndef REST_RPC_IO_SERVICE_POOL_H_
#define REST_RPC_IO_SERVICE_POOL_H_
#include <vector>
#include <memory>
#include "use_asio.hpp"
#include <memory>
#include <vector>
namespace rest_rpc {
namespace rpc_service {
class io_service_pool : private asio::noncopyable {
public:
public:
explicit io_service_pool(std::size_t pool_size) : next_io_service_(0) {
if (pool_size == 0) throw std::runtime_error("io_service_pool size is 0");
if (pool_size == 0)
throw std::runtime_error("io_service_pool size is 0");
for (std::size_t i = 0; i < pool_size; ++i) {
io_service_ptr io_service(new boost::asio::io_service);
@@ -23,11 +24,12 @@ class io_service_pool : private asio::noncopyable {
void run() {
std::vector<std::shared_ptr<std::thread>> threads;
for (std::size_t i = 0; i < io_services_.size(); ++i) {
threads.emplace_back(
std::make_shared<std::thread>([](io_service_ptr svr) { svr->run(); }, io_services_[i]));
threads.emplace_back(std::make_shared<std::thread>(
[](io_service_ptr svr) { svr->run(); }, io_services_[i]));
}
for (std::size_t i = 0; i < threads.size(); ++i) threads[i]->join();
for (std::size_t i = 0; i < threads.size(); ++i)
threads[i]->join();
}
void stop() {
@@ -36,14 +38,15 @@ class io_service_pool : private asio::noncopyable {
}
}
boost::asio::io_service& get_io_service() {
boost::asio::io_service& io_service = *io_services_[next_io_service_];
boost::asio::io_service &get_io_service() {
boost::asio::io_service &io_service = *io_services_[next_io_service_];
++next_io_service_;
if (next_io_service_ == io_services_.size()) next_io_service_ = 0;
if (next_io_service_ == io_services_.size())
next_io_service_ = 0;
return io_service;
}
private:
private:
typedef std::shared_ptr<boost::asio::io_service> io_service_ptr;
typedef std::shared_ptr<boost::asio::io_service::work> work_ptr;
@@ -56,7 +59,7 @@ class io_service_pool : private asio::noncopyable {
/// The next io_service to use for a connection.
std::size_t next_io_service_;
};
} // namespace rpc_service
} // namespace rest_rpc
} // namespace rpc_service
} // namespace rest_rpc
#endif // REST_RPC_IO_SERVICE_POOL_H_
#endif // REST_RPC_IO_SERVICE_POOL_H_
include/rest_rpc/meta_util.hpp
+111 -98
View File
@@ -1,106 +1,119 @@
#ifndef REST_RPC_META_UTIL_HPP
#define REST_RPC_META_UTIL_HPP
#include <functional>
#include "cplusplus_14.h"
namespace rest_rpc{
namespace rest_rpc {
template <typename... Args, typename Func, std::size_t... Idx>
void for_each(const std::tuple<Args...>& t, Func&& f, std::index_sequence<Idx...>) {
(void)std::initializer_list<int> { (f(std::get<Idx>(t)), void(), 0)...};
}
template <typename... Args, typename Func, std::size_t... Idx>
void for_each_i(const std::tuple<Args...>& t, Func&& f, std::index_sequence<Idx...>) {
(void)std::initializer_list<int> { (f(std::get<Idx>(t), std::integral_constant<size_t, Idx>{}), void(), 0)...};
}
template<typename T>
struct function_traits;
template<typename Ret, typename Arg, typename... Args>
struct function_traits<Ret(Arg, Args...)>
{
public:
enum { arity = sizeof...(Args)+1 };
typedef Ret function_type(Arg, Args...);
typedef Ret return_type;
using stl_function_type = std::function<function_type>;
typedef Ret(*pointer)(Arg, Args...);
typedef std::tuple<Arg, Args...> tuple_type;
typedef std::tuple<std::remove_const_t<std::remove_reference_t<Arg>>, std::remove_const_t<std::remove_reference_t<Args>>...> bare_tuple_type;
using args_tuple = std::tuple<std::string, Arg, std::remove_const_t<std::remove_reference_t<Args>>...>;
using args_tuple_2nd = std::tuple<std::string, std::remove_const_t<std::remove_reference_t<Args>>...>;
};
template<typename Ret>
struct function_traits<Ret()> {
public:
enum { arity = 0 };
typedef Ret function_type();
typedef Ret return_type;
using stl_function_type = std::function<function_type>;
typedef Ret(*pointer)();
typedef std::tuple<> tuple_type;
typedef std::tuple<> bare_tuple_type;
using args_tuple = std::tuple<std::string>;
using args_tuple_2nd = std::tuple<std::string>;
};
template<typename Ret, typename... Args>
struct function_traits<Ret(*)(Args...)> : function_traits<Ret(Args...)>{};
template <typename Ret, typename... Args>
struct function_traits<std::function<Ret(Args...)>> : function_traits<Ret(Args...)>{};
template <typename ReturnType, typename ClassType, typename... Args>
struct function_traits<ReturnType(ClassType::*)(Args...)> : function_traits<ReturnType(Args...)>{};
template <typename ReturnType, typename ClassType, typename... Args>
struct function_traits<ReturnType(ClassType::*)(Args...) const> : function_traits<ReturnType(Args...)>{};
template<typename Callable>
struct function_traits : function_traits<decltype(&Callable::operator())>{};
template<typename T>
using remove_const_reference_t = std::remove_const_t<std::remove_reference_t<T>>;
template<size_t... Is>
auto make_tuple_from_sequence(std::index_sequence<Is...>)->decltype(std::make_tuple(Is...)) {
std::make_tuple(Is...);
}
template<size_t N>
constexpr auto make_tuple_from_sequence()->decltype(make_tuple_from_sequence(std::make_index_sequence<N>{})) {
return make_tuple_from_sequence(std::make_index_sequence<N>{});
}
namespace detail {
template <class Tuple, class F, std::size_t...Is>
void tuple_switch(const std::size_t i, Tuple&& t, F&& f, std::index_sequence<Is...>) {
(void)std::initializer_list<int> {
(i == Is && (
(void)std::forward<F>(f)(std::integral_constant<size_t, Is>{}), 0))...
};
}
} // namespace detail
template <class Tuple, class F>
inline void tuple_switch(const std::size_t i, Tuple&& t, F&& f) {
constexpr auto N =
std::tuple_size<std::remove_reference_t<Tuple>>::value;
detail::tuple_switch(i, std::forward<Tuple>(t), std::forward<F>(f),
std::make_index_sequence<N>{});
}
template<int N, typename... Args>
using nth_type_of = std::tuple_element_t<N, std::tuple<Args...>>;
template<typename... Args>
using last_type_of = nth_type_of<sizeof...(Args)-1, Args...>;
template <typename... Args, typename Func, std::size_t... Idx>
void for_each(const std::tuple<Args...> &t, Func &&f,
std::index_sequence<Idx...>) {
(void)std::initializer_list<int>{(f(std::get<Idx>(t)), void(), 0)...};
}
#endif //REST_RPC_META_UTIL_HPP
template <typename... Args, typename Func, std::size_t... Idx>
void for_each_i(const std::tuple<Args...> &t, Func &&f,
std::index_sequence<Idx...>) {
(void)std::initializer_list<int>{
(f(std::get<Idx>(t), std::integral_constant<size_t, Idx>{}), void(),
0)...};
}
template <typename T> struct function_traits;
template <typename Ret, typename Arg, typename... Args>
struct function_traits<Ret(Arg, Args...)> {
public:
enum { arity = sizeof...(Args) + 1 };
typedef Ret function_type(Arg, Args...);
typedef Ret return_type;
using stl_function_type = std::function<function_type>;
typedef Ret (*pointer)(Arg, Args...);
typedef std::tuple<Arg, Args...> tuple_type;
typedef std::tuple<std::remove_const_t<std::remove_reference_t<Arg>>,
std::remove_const_t<std::remove_reference_t<Args>>...>
bare_tuple_type;
using args_tuple =
std::tuple<std::string, Arg,
std::remove_const_t<std::remove_reference_t<Args>>...>;
using args_tuple_2nd =
std::tuple<std::string,
std::remove_const_t<std::remove_reference_t<Args>>...>;
};
template <typename Ret> struct function_traits<Ret()> {
public:
enum { arity = 0 };
typedef Ret function_type();
typedef Ret return_type;
using stl_function_type = std::function<function_type>;
typedef Ret (*pointer)();
typedef std::tuple<> tuple_type;
typedef std::tuple<> bare_tuple_type;
using args_tuple = std::tuple<std::string>;
using args_tuple_2nd = std::tuple<std::string>;
};
template <typename Ret, typename... Args>
struct function_traits<Ret (*)(Args...)> : function_traits<Ret(Args...)> {};
template <typename Ret, typename... Args>
struct function_traits<std::function<Ret(Args...)>>
: function_traits<Ret(Args...)> {};
template <typename ReturnType, typename ClassType, typename... Args>
struct function_traits<ReturnType (ClassType::*)(Args...)>
: function_traits<ReturnType(Args...)> {};
template <typename ReturnType, typename ClassType, typename... Args>
struct function_traits<ReturnType (ClassType::*)(Args...) const>
: function_traits<ReturnType(Args...)> {};
template <typename Callable>
struct function_traits : function_traits<decltype(&Callable::operator())> {};
template <typename T>
using remove_const_reference_t =
std::remove_const_t<std::remove_reference_t<T>>;
template <size_t... Is>
auto make_tuple_from_sequence(std::index_sequence<Is...>)
-> decltype(std::make_tuple(Is...)) {
std::make_tuple(Is...);
}
template <size_t N>
constexpr auto make_tuple_from_sequence()
-> decltype(make_tuple_from_sequence(std::make_index_sequence<N>{})) {
return make_tuple_from_sequence(std::make_index_sequence<N>{});
}
namespace detail {
template <class Tuple, class F, std::size_t... Is>
void tuple_switch(const std::size_t i, Tuple &&t, F &&f,
std::index_sequence<Is...>) {
(void)std::initializer_list<int>{
(i == Is &&
((void)std::forward<F>(f)(std::integral_constant<size_t, Is>{}), 0))...};
}
} // namespace detail
template <class Tuple, class F>
inline void tuple_switch(const std::size_t i, Tuple &&t, F &&f) {
constexpr auto N = std::tuple_size<std::remove_reference_t<Tuple>>::value;
detail::tuple_switch(i, std::forward<Tuple>(t), std::forward<F>(f),
std::make_index_sequence<N>{});
}
template <int N, typename... Args>
using nth_type_of = std::tuple_element_t<N, std::tuple<Args...>>;
template <typename... Args>
using last_type_of = nth_type_of<sizeof...(Args) - 1, Args...>;
} // namespace rest_rpc
#endif // REST_RPC_META_UTIL_HPP
include/rest_rpc/nonstd_any.hpp
+365 -405
View File
File diff suppressed because it is too large Load Diff
include/rest_rpc/router.h
+177 -157
View File
@@ -1,185 +1,205 @@
#ifndef REST_RPC_ROUTER_H_
#define REST_RPC_ROUTER_H_
#include <functional>
#include "use_asio.hpp"
#include "codec.h"
#include "meta_util.hpp"
#include "use_asio.hpp"
#include <functional>
namespace rest_rpc {
enum class ExecMode { sync, async };
const constexpr ExecMode Async = ExecMode::async;
enum class ExecMode { sync, async };
const constexpr ExecMode Async = ExecMode::async;
namespace rpc_service {
class connection;
namespace rpc_service {
class connection;
class router : asio::noncopyable {
public:
template<ExecMode model, typename Function>
void register_handler(std::string const& name, Function f) {
return register_nonmember_func<model>(name, std::move(f));
}
class router : asio::noncopyable {
public:
template <ExecMode model, typename Function>
void register_handler(std::string const &name, Function f) {
return register_nonmember_func<model>(name, std::move(f));
}
template<ExecMode model, typename Function, typename Self>
void register_handler(std::string const& name, const Function& f, Self* self) {
return register_member_func<model>(name, f, self);
}
template <ExecMode model, typename Function, typename Self>
void register_handler(std::string const &name, const Function &f,
Self *self) {
return register_member_func<model>(name, f, self);
}
void remove_handler(std::string const& name) { this->map_invokers_.erase(name); }
void remove_handler(std::string const &name) {
this->map_invokers_.erase(name);
}
template<typename T>
void route(const char* data, std::size_t size, std::weak_ptr<T> conn) {
auto conn_sp = conn.lock();
if (!conn_sp) {
return;
}
template <typename T>
void route(const char *data, std::size_t size, std::weak_ptr<T> conn) {
auto conn_sp = conn.lock();
if (!conn_sp) {
return;
}
auto req_id = conn_sp->request_id();
std::string result;
try {
msgpack_codec codec;
auto p = codec.unpack<std::tuple<std::string>>(data, size);
auto& func_name = std::get<0>(p);
auto it = map_invokers_.find(func_name);
if (it == map_invokers_.end()) {
result = codec.pack_args_str(result_code::FAIL, "unknown function: " + func_name);
conn_sp->response(req_id, std::move(result));
return;
}
auto req_id = conn_sp->request_id();
std::string result;
try {
msgpack_codec codec;
auto p = codec.unpack<std::tuple<std::string>>(data, size);
auto &func_name = std::get<0>(p);
auto it = map_invokers_.find(func_name);
if (it == map_invokers_.end()) {
result = codec.pack_args_str(result_code::FAIL,
"unknown function: " + func_name);
conn_sp->response(req_id, std::move(result));
return;
}
ExecMode model;
it->second(conn, data, size, result, model);
if (model == ExecMode::sync) {
if (result.size() >= MAX_BUF_LEN) {
result = codec.pack_args_str(result_code::FAIL, "the response result is out of range: more than 10M " + func_name);
}
conn_sp->response(req_id, std::move(result));
}
}
catch (const std::exception & ex) {
msgpack_codec codec;
result = codec.pack_args_str(result_code::FAIL, ex.what());
conn_sp->response(req_id, std::move(result));
}
}
ExecMode model;
it->second(conn, data, size, result, model);
if (model == ExecMode::sync) {
if (result.size() >= MAX_BUF_LEN) {
result = codec.pack_args_str(
result_code::FAIL,
"the response result is out of range: more than 10M " +
func_name);
}
conn_sp->response(req_id, std::move(result));
}
} catch (const std::exception &ex) {
msgpack_codec codec;
result = codec.pack_args_str(result_code::FAIL, ex.what());
conn_sp->response(req_id, std::move(result));
}
}
router() = default;
router() = default;
private:
router(const router&) = delete;
router(router&&) = delete;
private:
router(const router &) = delete;
router(router &&) = delete;
template<typename F, size_t... I, typename Arg, typename... Args>
static typename std::result_of<F(std::weak_ptr<connection>, Args...)>::type call_helper(
const F & f, const std::index_sequence<I...>&, std::tuple<Arg, Args...> tup, std::weak_ptr<connection> ptr) {
return f(ptr, std::move(std::get<I + 1>(tup))...);
}
template <typename F, size_t... I, typename Arg, typename... Args>
static typename std::result_of<F(std::weak_ptr<connection>, Args...)>::type
call_helper(const F &f, const std::index_sequence<I...> &,
std::tuple<Arg, Args...> tup, std::weak_ptr<connection> ptr) {
return f(ptr, std::move(std::get<I + 1>(tup))...);
}
template<typename F, typename Arg, typename... Args>
static
typename std::enable_if<std::is_void<typename std::result_of<F(std::weak_ptr<connection>, Args...)>::type>::value>::type
call(const F & f, std::weak_ptr<connection> ptr, std::string & result, std::tuple<Arg, Args...> tp) {
call_helper(f, std::make_index_sequence<sizeof...(Args)>{}, std::move(tp), ptr);
result = msgpack_codec::pack_args_str(result_code::OK);
}
template <typename F, typename Arg, typename... Args>
static typename std::enable_if<std::is_void<typename std::result_of<
F(std::weak_ptr<connection>, Args...)>::type>::value>::type
call(const F &f, std::weak_ptr<connection> ptr, std::string &result,
std::tuple<Arg, Args...> tp) {
call_helper(f, std::make_index_sequence<sizeof...(Args)>{}, std::move(tp),
ptr);
result = msgpack_codec::pack_args_str(result_code::OK);
}
template<typename F, typename Arg, typename... Args>
static
typename std::enable_if<!std::is_void<typename std::result_of<F(std::weak_ptr<connection>, Args...)>::type>::value>::type
call(const F & f, std::weak_ptr<connection> ptr, std::string & result, std::tuple<Arg, Args...> tp) {
auto r = call_helper(f, std::make_index_sequence<sizeof...(Args)>{}, std::move(tp), ptr);
msgpack_codec codec;
result = msgpack_codec::pack_args_str(result_code::OK, r);
}
template <typename F, typename Arg, typename... Args>
static typename std::enable_if<!std::is_void<typename std::result_of<
F(std::weak_ptr<connection>, Args...)>::type>::value>::type
call(const F &f, std::weak_ptr<connection> ptr, std::string &result,
std::tuple<Arg, Args...> tp) {
auto r = call_helper(f, std::make_index_sequence<sizeof...(Args)>{},
std::move(tp), ptr);
msgpack_codec codec;
result = msgpack_codec::pack_args_str(result_code::OK, r);
}
template<typename F, typename Self, size_t... Indexes, typename Arg, typename... Args>
static typename std::result_of<F(Self, std::weak_ptr<connection>, Args...)>::type call_member_helper(
const F & f, Self * self, const std::index_sequence<Indexes...>&,
std::tuple<Arg, Args...> tup, std::weak_ptr<connection> ptr = std::shared_ptr<connection>{ nullptr }) {
return (*self.*f)(ptr, std::move(std::get<Indexes + 1>(tup))...);
}
template <typename F, typename Self, size_t... Indexes, typename Arg,
typename... Args>
static
typename std::result_of<F(Self, std::weak_ptr<connection>, Args...)>::type
call_member_helper(const F &f, Self *self,
const std::index_sequence<Indexes...> &,
std::tuple<Arg, Args...> tup,
std::weak_ptr<connection> ptr =
std::shared_ptr<connection>{nullptr}) {
return (*self.*f)(ptr, std::move(std::get<Indexes + 1>(tup))...);
}
template<typename F, typename Self, typename Arg, typename... Args>
static typename std::enable_if<
std::is_void<typename std::result_of<F(Self, std::weak_ptr<connection>, Args...)>::type>::value>::type
call_member(const F & f, Self * self, std::weak_ptr<connection> ptr, std::string & result,
std::tuple<Arg, Args...> tp) {
call_member_helper(f, self, typename std::make_index_sequence<sizeof...(Args)>{}, std::move(tp), ptr);
result = msgpack_codec::pack_args_str(result_code::OK);
}
template <typename F, typename Self, typename Arg, typename... Args>
static typename std::enable_if<std::is_void<typename std::result_of<
F(Self, std::weak_ptr<connection>, Args...)>::type>::value>::type
call_member(const F &f, Self *self, std::weak_ptr<connection> ptr,
std::string &result, std::tuple<Arg, Args...> tp) {
call_member_helper(f, self,
typename std::make_index_sequence<sizeof...(Args)>{},
std::move(tp), ptr);
result = msgpack_codec::pack_args_str(result_code::OK);
}
template<typename F, typename Self, typename Arg, typename... Args>
static typename std::enable_if<
!std::is_void<typename std::result_of<F(Self, std::weak_ptr<connection>, Args...)>::type>::value>::type
call_member(const F & f, Self * self, std::weak_ptr<connection> ptr, std::string & result,
std::tuple<Arg, Args...> tp) {
auto r =
call_member_helper(f, self, typename std::make_index_sequence<sizeof...(Args)>{}, std::move(tp), ptr);
result = msgpack_codec::pack_args_str(result_code::OK, r);
}
template <typename F, typename Self, typename Arg, typename... Args>
static typename std::enable_if<!std::is_void<typename std::result_of<
F(Self, std::weak_ptr<connection>, Args...)>::type>::value>::type
call_member(const F &f, Self *self, std::weak_ptr<connection> ptr,
std::string &result, std::tuple<Arg, Args...> tp) {
auto r = call_member_helper(
f, self, typename std::make_index_sequence<sizeof...(Args)>{},
std::move(tp), ptr);
result = msgpack_codec::pack_args_str(result_code::OK, r);
}
template<typename Function, ExecMode mode = ExecMode::sync>
struct invoker {
template<ExecMode model>
static inline void apply(const Function& func, std::weak_ptr<connection> conn, const char* data, size_t size,
std::string& result, ExecMode& exe_model) {
using args_tuple = typename function_traits<Function>::args_tuple_2nd;
exe_model = ExecMode::sync;
msgpack_codec codec;
try {
auto tp = codec.unpack<args_tuple>(data, size);
call(func, conn, result, std::move(tp));
exe_model = model;
}
catch (std::invalid_argument & e) {
result = codec.pack_args_str(result_code::FAIL, e.what());
}
catch (const std::exception & e) {
result = codec.pack_args_str(result_code::FAIL, e.what());
}
}
template <typename Function, ExecMode mode = ExecMode::sync> struct invoker {
template <ExecMode model>
static inline void apply(const Function &func,
std::weak_ptr<connection> conn, const char *data,
size_t size, std::string &result,
ExecMode &exe_model) {
using args_tuple = typename function_traits<Function>::args_tuple_2nd;
exe_model = ExecMode::sync;
msgpack_codec codec;
try {
auto tp = codec.unpack<args_tuple>(data, size);
call(func, conn, result, std::move(tp));
exe_model = model;
} catch (std::invalid_argument &e) {
result = codec.pack_args_str(result_code::FAIL, e.what());
} catch (const std::exception &e) {
result = codec.pack_args_str(result_code::FAIL, e.what());
}
}
template<ExecMode model, typename Self>
static inline void apply_member(const Function& func, Self* self, std::weak_ptr<connection> conn,
const char* data, size_t size, std::string& result,
ExecMode& exe_model) {
using args_tuple = typename function_traits<Function>::args_tuple_2nd;
exe_model = ExecMode::sync;
msgpack_codec codec;
try {
auto tp = codec.unpack<args_tuple>(data, size);
call_member(func, self, conn, result, std::move(tp));
exe_model = model;
}
catch (std::invalid_argument & e) {
result = codec.pack_args_str(result_code::FAIL, e.what());
}
catch (const std::exception & e) {
result = codec.pack_args_str(result_code::FAIL, e.what());
}
}
};
template <ExecMode model, typename Self>
static inline void apply_member(const Function &func, Self *self,
std::weak_ptr<connection> conn,
const char *data, size_t size,
std::string &result, ExecMode &exe_model) {
using args_tuple = typename function_traits<Function>::args_tuple_2nd;
exe_model = ExecMode::sync;
msgpack_codec codec;
try {
auto tp = codec.unpack<args_tuple>(data, size);
call_member(func, self, conn, result, std::move(tp));
exe_model = model;
} catch (std::invalid_argument &e) {
result = codec.pack_args_str(result_code::FAIL, e.what());
} catch (const std::exception &e) {
result = codec.pack_args_str(result_code::FAIL, e.what());
}
}
};
template<ExecMode model, typename Function>
void register_nonmember_func(std::string const& name, Function f) {
this->map_invokers_[name] = { std::bind(&invoker<Function>::template apply<model>, std::move(f), std::placeholders::_1,
std::placeholders::_2, std::placeholders::_3,
std::placeholders::_4, std::placeholders::_5) };
}
template <ExecMode model, typename Function>
void register_nonmember_func(std::string const &name, Function f) {
this->map_invokers_[name] = {std::bind(
&invoker<Function>::template apply<model>, std::move(f),
std::placeholders::_1, std::placeholders::_2, std::placeholders::_3,
std::placeholders::_4, std::placeholders::_5)};
}
template<ExecMode model, typename Function, typename Self>
void register_member_func(const std::string& name, const Function& f, Self* self) {
this->map_invokers_[name] = { std::bind(&invoker<Function>::template apply_member<model, Self>,
f, self, std::placeholders::_1, std::placeholders::_2,
std::placeholders::_3, std::placeholders::_4,
std::placeholders::_5) };
}
template <ExecMode model, typename Function, typename Self>
void register_member_func(const std::string &name, const Function &f,
Self *self) {
this->map_invokers_[name] = {std::bind(
&invoker<Function>::template apply_member<model, Self>, f, self,
std::placeholders::_1, std::placeholders::_2, std::placeholders::_3,
std::placeholders::_4, std::placeholders::_5)};
}
std::unordered_map<std::string,
std::function<void(std::weak_ptr<connection>, const char*, size_t, std::string&, ExecMode& model)>>
map_invokers_;
};
} // namespace rpc_service
} // namespace rest_rpc
std::unordered_map<
std::string, std::function<void(std::weak_ptr<connection>, const char *,
size_t, std::string &, ExecMode &model)>>
map_invokers_;
};
} // namespace rpc_service
} // namespace rest_rpc
#endif // REST_RPC_ROUTER_H_
#endif // REST_RPC_ROUTER_H_
include/rest_rpc/rpc_client.hpp
+23 -21
View File
@@ -1,21 +1,21 @@
#pragma once
#include <iostream>
#include <string>
#include <deque>
#include <future>
#include <utility>
#include "use_asio.hpp"
#include "client_util.hpp"
#include "const_vars.h"
#include "meta_util.hpp"
#include "use_asio.hpp"
#include <deque>
#include <functional>
#include <future>
#include <iostream>
#include <string>
#include <utility>
using namespace rest_rpc::rpc_service;
namespace rest_rpc {
/**
* The type to indicate the language of the client.
* The type to indicate the language of the client.
*/
enum class client_language_t {
CPP = 0,
@@ -192,7 +192,7 @@ public:
// sync call
#if __cplusplus > 201402L
template <size_t TIMEOUT, typename T = void, typename... Args>
auto call(const std::string &rpc_name, Args &&... 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));
@@ -209,13 +209,13 @@ public:
}
template <typename T = void, typename... Args>
auto call(const std::string &rpc_name, Args &&... 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) {
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));
@@ -229,13 +229,13 @@ public:
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(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) {
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));
@@ -249,14 +249,14 @@ public:
template <typename T, typename... Args>
typename std::enable_if<!std::is_void<T>::value, T>::type
call(const std::string &rpc_name, Args &&... args) {
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) {
Args &&...args) {
auto p = std::make_shared<std::promise<req_result>>();
std::future<req_result> future = p->get_future();
@@ -275,9 +275,9 @@ public:
}
/**
* This internal_async_call is used for other language client.
* We use callback to handle the result is received, so we should not
* add the future to the future map.
* This internal_async_call is used for other language client.
* We use callback to handle the result is received, so we should not
* add the future to the future map.
*/
long internal_async_call(const std::string &encoded_func_name_and_args) {
auto p = std::make_shared<std::promise<req_result>>();
@@ -298,7 +298,7 @@ public:
void
async_call(const std::string &rpc_name,
std::function<void(boost::system::error_code, string_view)> cb,
Args &&... args) {
Args &&...args) {
if (!has_connected_) {
if (cb)
cb(boost::asio::error::make_error_code(
@@ -503,7 +503,8 @@ private:
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));
// 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) {
@@ -776,7 +777,8 @@ private:
ssl_context);
// verify peer TODO
#else
assert(is_ssl()); // please add definition CINATRA_ENABLE_SSL, not allowed coming in this branch
assert(is_ssl()); // please add definition CINATRA_ENABLE_SSL, not allowed
// coming in this branch
#endif
}
@@ -869,4 +871,4 @@ private:
client_language_t client_language_ = client_language_t::CPP;
std::function<void(long, const std::string &)> on_result_received_callback_;
};
}
} // namespace rest_rpc
include/rest_rpc/rpc_server.h
+186 -181
View File
@@ -1,226 +1,231 @@
#ifndef REST_RPC_RPC_SERVER_H_
#define REST_RPC_RPC_SERVER_H_
#include <thread>
#include <mutex>
#include <condition_variable>
#include "connection.h"
#include "io_service_pool.h"
#include "router.h"
#include <condition_variable>
#include <mutex>
#include <thread>
using boost::asio::ip::tcp;
namespace rest_rpc {
namespace rpc_service {
using rpc_conn = std::weak_ptr<connection>;
class rpc_server : private asio::noncopyable {
public:
rpc_server(unsigned short port, size_t size, size_t timeout_seconds = 15, size_t check_seconds = 10)
: io_service_pool_(size),
acceptor_(io_service_pool_.get_io_service(), tcp::endpoint(tcp::v4(), port)),
timeout_seconds_(timeout_seconds),
check_seconds_(check_seconds) {
do_accept();
check_thread_ = std::make_shared<std::thread>([this] { clean(); });
pub_sub_thread_ = std::make_shared<std::thread>([this] { clean_sub_pub(); });
}
namespace rpc_service {
using rpc_conn = std::weak_ptr<connection>;
class rpc_server : private asio::noncopyable {
public:
rpc_server(unsigned short port, size_t size, size_t timeout_seconds = 15,
size_t check_seconds = 10)
: io_service_pool_(size), acceptor_(io_service_pool_.get_io_service(),
tcp::endpoint(tcp::v4(), port)),
timeout_seconds_(timeout_seconds), check_seconds_(check_seconds) {
do_accept();
check_thread_ = std::make_shared<std::thread>([this] { clean(); });
pub_sub_thread_ =
std::make_shared<std::thread>([this] { clean_sub_pub(); });
}
rpc_server(unsigned short port, size_t size, ssl_configure ssl_conf, size_t timeout_seconds = 15, size_t check_seconds = 10) :
rpc_server(port, size, timeout_seconds, check_seconds) {
rpc_server(unsigned short port, size_t size, ssl_configure ssl_conf,
size_t timeout_seconds = 15, size_t check_seconds = 10)
: rpc_server(port, size, timeout_seconds, check_seconds) {
#ifdef CINATRA_ENABLE_SSL
ssl_conf_ = std::move(ssl_conf);
ssl_conf_ = std::move(ssl_conf);
#else
assert(false);//please add definition CINATRA_ENABLE_SSL, not allowed coming in this branch
assert(false); // please add definition CINATRA_ENABLE_SSL, not allowed
// coming in this branch
#endif
}
}
~rpc_server() {
{
std::unique_lock<std::mutex> lock(mtx_);
stop_check_ = true;
cv_.notify_all();
}
check_thread_->join();
~rpc_server() {
{
std::unique_lock<std::mutex> lock(mtx_);
stop_check_ = true;
cv_.notify_all();
}
check_thread_->join();
{
std::unique_lock<std::mutex> lock(sub_mtx_);
stop_check_pub_sub_ = true;
sub_cv_.notify_all();
}
pub_sub_thread_->join();
{
std::unique_lock<std::mutex> lock(sub_mtx_);
stop_check_pub_sub_ = true;
sub_cv_.notify_all();
}
pub_sub_thread_->join();
io_service_pool_.stop();
if(thd_){
thd_->join();
}
}
io_service_pool_.stop();
if (thd_) {
thd_->join();
}
}
void async_run() {
thd_ = std::make_shared<std::thread>([this] { io_service_pool_.run(); });
}
void async_run() {
thd_ = std::make_shared<std::thread>([this] { io_service_pool_.run(); });
}
void run() {
io_service_pool_.run();
}
void run() { io_service_pool_.run(); }
template<ExecMode model = ExecMode::sync, typename Function>
void register_handler(std::string const& name, const Function& f) {
router_.register_handler<model>(name, f);
}
template <ExecMode model = ExecMode::sync, typename Function>
void register_handler(std::string const &name, const Function &f) {
router_.register_handler<model>(name, f);
}
template<ExecMode model = ExecMode::sync, typename Function, typename Self>
void register_handler(std::string const& name, const Function& f, Self* self) {
router_.register_handler<model>(name, f, self);
}
template <ExecMode model = ExecMode::sync, typename Function, typename Self>
void register_handler(std::string const &name, const Function &f,
Self *self) {
router_.register_handler<model>(name, f, self);
}
void set_conn_timeout_callback(std::function<void(int64_t)> callback) {
conn_timeout_callback_ = std::move(callback);
}
void set_conn_timeout_callback(std::function<void(int64_t)> callback) {
conn_timeout_callback_ = std::move(callback);
}
template<typename T>
void publish(const std::string& key, T data) {
publish(key, "", std::move(data));
}
template <typename T> void publish(const std::string &key, T data) {
publish(key, "", std::move(data));
}
template<typename T>
void publish_by_token(const std::string& key, std::string token, T data) {
publish(key, std::move(token), std::move(data));
}
template <typename T>
void publish_by_token(const std::string &key, std::string token, T data) {
publish(key, std::move(token), std::move(data));
}
std::set<std::string> get_token_list() {
std::unique_lock<std::mutex> lock(sub_mtx_);
return token_list_;
}
std::set<std::string> get_token_list() {
std::unique_lock<std::mutex> lock(sub_mtx_);
return token_list_;
}
private:
void do_accept() {
conn_.reset(new connection(io_service_pool_.get_io_service(), timeout_seconds_, router_));
conn_->set_callback([this](std::string key, std::string token, std::weak_ptr<connection> conn) {
std::unique_lock<std::mutex> lock(sub_mtx_);
sub_map_.emplace(std::move(key) + token, conn);
if (!token.empty()) {
token_list_.emplace(std::move(token));
}
});
private:
void do_accept() {
conn_.reset(new connection(io_service_pool_.get_io_service(),
timeout_seconds_, router_));
conn_->set_callback([this](std::string key, std::string token,
std::weak_ptr<connection> conn) {
std::unique_lock<std::mutex> lock(sub_mtx_);
sub_map_.emplace(std::move(key) + token, conn);
if (!token.empty()) {
token_list_.emplace(std::move(token));
}
});
acceptor_.async_accept(conn_->socket(), [this](boost::system::error_code ec) {
if (ec) {
//LOG(INFO) << "acceptor error: " << ec.message();
}
else {
acceptor_.async_accept(conn_->socket(),
[this](boost::system::error_code ec) {
if (ec) {
// LOG(INFO) << "acceptor error: " <<
// ec.message();
} else {
#ifdef CINATRA_ENABLE_SSL
if (!ssl_conf_.cert_file.empty()) {
conn_->init_ssl_context(ssl_conf_);
}
if (!ssl_conf_.cert_file.empty()) {
conn_->init_ssl_context(ssl_conf_);
}
#endif
conn_->start();
std::unique_lock<std::mutex> lock(mtx_);
conn_->set_conn_id(conn_id_);
connections_.emplace(conn_id_++, conn_);
}
conn_->start();
std::unique_lock<std::mutex> lock(mtx_);
conn_->set_conn_id(conn_id_);
connections_.emplace(conn_id_++, conn_);
}
do_accept();
});
}
do_accept();
});
}
void clean() {
while (!stop_check_) {
std::unique_lock<std::mutex> lock(mtx_);
cv_.wait_for(lock, std::chrono::seconds(check_seconds_));
void clean() {
while (!stop_check_) {
std::unique_lock<std::mutex> lock(mtx_);
cv_.wait_for(lock, std::chrono::seconds(check_seconds_));
for (auto it = connections_.cbegin(); it != connections_.cend();) {
if (it->second->has_closed()) {
if (conn_timeout_callback_) {
conn_timeout_callback_(it->second->conn_id());
}
it = connections_.erase(it);
}
else {
++it;
}
}
}
}
for (auto it = connections_.cbegin(); it != connections_.cend();) {
if (it->second->has_closed()) {
if (conn_timeout_callback_) {
conn_timeout_callback_(it->second->conn_id());
}
it = connections_.erase(it);
} else {
++it;
}
}
}
}
void clean_sub_pub() {
while (!stop_check_pub_sub_) {
std::unique_lock<std::mutex> lock(sub_mtx_);
sub_cv_.wait_for(lock, std::chrono::seconds(10));
void clean_sub_pub() {
while (!stop_check_pub_sub_) {
std::unique_lock<std::mutex> lock(sub_mtx_);
sub_cv_.wait_for(lock, std::chrono::seconds(10));
for (auto it = sub_map_.cbegin(); it != sub_map_.cend();) {
auto conn = it->second.lock();
if (conn == nullptr || conn->has_closed()) {
it = sub_map_.erase(it);
}
else {
++it;
}
}
}
}
for (auto it = sub_map_.cbegin(); it != sub_map_.cend();) {
auto conn = it->second.lock();
if (conn == nullptr || conn->has_closed()) {
it = sub_map_.erase(it);
} else {
++it;
}
}
}
}
template<typename T>
void publish(std::string key, std::string token, T data) {
{
std::unique_lock<std::mutex> lock(sub_mtx_);
if (sub_map_.empty())
return;
}
template <typename T>
void publish(std::string key, std::string token, T data) {
{
std::unique_lock<std::mutex> lock(sub_mtx_);
if (sub_map_.empty())
return;
}
std::shared_ptr<std::string> shared_data = get_shared_data<T>(std::move(data));
std::unique_lock<std::mutex> lock(sub_mtx_);
auto range = sub_map_.equal_range(key + token);
for (auto it = range.first; it != range.second; ++it) {
auto conn = it->second.lock();
if (conn == nullptr || conn->has_closed()) {
continue;
}
std::shared_ptr<std::string> shared_data =
get_shared_data<T>(std::move(data));
std::unique_lock<std::mutex> lock(sub_mtx_);
auto range = sub_map_.equal_range(key + token);
for (auto it = range.first; it != range.second; ++it) {
auto conn = it->second.lock();
if (conn == nullptr || conn->has_closed()) {
continue;
}
conn->publish(key + token, *shared_data);
}
}
conn->publish(key + token, *shared_data);
}
}
template<typename T>
typename std::enable_if<std::is_assignable<std::string, T>::value, std::shared_ptr<std::string>>::type
get_shared_data(std::string data) {
return std::make_shared<std::string>(std::move(data));
}
template <typename T>
typename std::enable_if<std::is_assignable<std::string, T>::value,
std::shared_ptr<std::string>>::type
get_shared_data(std::string data) {
return std::make_shared<std::string>(std::move(data));
}
template<typename T>
typename std::enable_if<!std::is_assignable<std::string, T>::value, std::shared_ptr<std::string>>::type
get_shared_data(T data) {
msgpack_codec codec;
auto buf = codec.pack(std::move(data));
return std::make_shared<std::string>(buf.data(), buf.size());
}
template <typename T>
typename std::enable_if<!std::is_assignable<std::string, T>::value,
std::shared_ptr<std::string>>::type
get_shared_data(T data) {
msgpack_codec codec;
auto buf = codec.pack(std::move(data));
return std::make_shared<std::string>(buf.data(), buf.size());
}
io_service_pool io_service_pool_;
tcp::acceptor acceptor_;
std::shared_ptr<connection> conn_;
std::shared_ptr<std::thread> thd_;
std::size_t timeout_seconds_;
io_service_pool io_service_pool_;
tcp::acceptor acceptor_;
std::shared_ptr<connection> conn_;
std::shared_ptr<std::thread> thd_;
std::size_t timeout_seconds_;
std::unordered_map<int64_t, std::shared_ptr<connection>> connections_;
int64_t conn_id_ = 0;
std::mutex mtx_;
std::shared_ptr<std::thread> check_thread_;
size_t check_seconds_;
bool stop_check_ = false;
std::condition_variable cv_;
std::unordered_map<int64_t, std::shared_ptr<connection>> connections_;
int64_t conn_id_ = 0;
std::mutex mtx_;
std::shared_ptr<std::thread> check_thread_;
size_t check_seconds_;
bool stop_check_ = false;
std::condition_variable cv_;
std::function<void(int64_t)> conn_timeout_callback_;
std::unordered_multimap<std::string, std::weak_ptr<connection>> sub_map_;
std::set<std::string> token_list_;
std::mutex sub_mtx_;
std::condition_variable sub_cv_;
std::function<void(int64_t)> conn_timeout_callback_;
std::unordered_multimap<std::string, std::weak_ptr<connection>> sub_map_;
std::set<std::string> token_list_;
std::mutex sub_mtx_;
std::condition_variable sub_cv_;
std::shared_ptr<std::thread> pub_sub_thread_;
bool stop_check_pub_sub_ = false;
std::shared_ptr<std::thread> pub_sub_thread_;
bool stop_check_pub_sub_ = false;
ssl_configure ssl_conf_;
router router_;
};
} // namespace rpc_service
} // namespace rest_rpc
ssl_configure ssl_conf_;
router router_;
};
} // namespace rpc_service
} // namespace rest_rpc
#endif // REST_RPC_RPC_SERVER_H_
#endif // REST_RPC_RPC_SERVER_H_
include/rest_rpc/string_view.hpp
+881 -804
View File
File diff suppressed because it is too large Load Diff
include/rest_rpc/use_asio.hpp
+11 -12
View File
@@ -1,24 +1,23 @@
#pragma once
#if defined(ASIO_STANDALONE)
//MSVC : define environment path 'ASIO_STANDALONE_INCLUDE', e.g. 'E:\bdlibs\asio-1.10.6\include'
// MSVC : define environment path 'ASIO_STANDALONE_INCLUDE', e.g.
// 'E:\bdlibs\asio-1.10.6\include'
#include <asio.hpp>
#ifdef CINATRA_ENABLE_SSL
#include <asio/ssl.hpp>
#endif
#include <asio/steady_timer.hpp>
#include <asio/detail/noncopyable.hpp>
namespace boost
{
namespace asio
{
using namespace ::asio;
}
namespace system {
using ::std::error_code;
}
#include <asio/steady_timer.hpp>
namespace boost {
namespace asio {
using namespace ::asio;
}
namespace system {
using ::std::error_code;
}
} // namespace boost
#else
#include <boost/asio.hpp>
#ifdef CINATRA_ENABLE_SSL
@@ -47,7 +46,7 @@ using string_view = boost::string_view;
#ifdef CINATRA_ENABLE_SSL
#if __cplusplus > 201402L
#if defined (__GNUC__)
#if defined(__GNUC__)
#if __GNUC__ < 8
#include <experimental/filesystem>
namespace rpcfs = std::experimental::filesystem;