diff --git a/examples/client/main.cpp b/examples/client/main.cpp index c091415..7474d6d 100644 --- a/examples/client/main.cpp +++ b/examples/client/main.cpp @@ -200,6 +200,25 @@ void test_download() { } } +void test_echo() { + rpc_client client("127.0.0.1", 9000); + bool r = client.connect(); + if (!r) { + std::cout << "connect timeout" << std::endl; + return; + } + + { + auto result = client.call("echo", "test"); + std::cout << result << std::endl; + } + + { + auto result = client.call("async_echo", "test"); + std::cout << result << std::endl; + } +} + void test_sync_client() { test_add(); test_translate(); @@ -297,6 +316,7 @@ void test_connect(){ } int main() { + test_echo(); test_sync_client(); test_async_client(); diff --git a/examples/server/main.cpp b/examples/server/main.cpp index c074a69..69543ba 100644 --- a/examples/server/main.cpp +++ b/examples/server/main.cpp @@ -6,10 +6,10 @@ using namespace rpc_service; #include "qps.h" struct dummy{ - int add(connection* conn, int a, int b) { return a + b; } + int add(rpc_conn conn, int a, int b) { return a + b; } }; -std::string translate(connection* conn, const std::string& orignal) { +std::string translate(rpc_conn conn, const std::string& orignal) { std::string temp = orignal; for (auto& c : temp) { c = std::toupper(c); @@ -17,7 +17,7 @@ std::string translate(connection* conn, const std::string& orignal) { return temp; } -void hello(connection* conn, const std::string& str) { +void hello(rpc_conn conn, const std::string& str) { std::cout << "hello " << str << std::endl; } @@ -29,21 +29,21 @@ struct person { MSGPACK_DEFINE(id, name, age); }; -std::string get_person_name(connection* conn, const person& p) { +std::string get_person_name(rpc_conn conn, const person& p) { return p.name; } -person get_person(connection* conn) { +person get_person(rpc_conn conn) { return { 1, "tom", 20 }; } -void upload(connection* conn, const std::string& filename, const std::string& content) { +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()); } -std::string download(connection* conn, const std::string& filename) { +std::string download(rpc_conn conn, const std::string& filename) { std::ifstream file(filename, std::ios::binary); if (!file) { return ""; @@ -62,14 +62,25 @@ std::string download(connection* conn, const std::string& filename) { qps g_qps; -std::string get_name(connection* conn, const person& p) { +std::string get_name(rpc_conn conn, const person& p) { g_qps.increase(); return p.name; } -//if you want to use conneciton to response immediately, use ExecMode::async model -void echo(connection* conn, const std::string& src) { - conn->response(src); +//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) { + std::thread thd([conn, src] { + std::this_thread::sleep_for(std::chrono::seconds(1)); + auto conn_sp = conn.lock(); + if (conn_sp) { + conn_sp->pack_and_response(std::move(src)); + } + }); + thd.detach(); +} + +std::string echo(rpc_conn conn, const std::string& src) { + return src; } int main() { @@ -84,7 +95,8 @@ int main() { server.register_handler("upload", upload); server.register_handler("download", download); server.register_handler("get_name", get_name); - server.register_handler("echo", echo); + server.register_handler("async_echo", async_echo); + server.register_handler("echo", echo); server.run(); diff --git a/include/connection.h b/include/connection.h index f5da70a..6d9a48b 100644 --- a/include/connection.h +++ b/include/connection.h @@ -57,6 +57,11 @@ class connection : public std::enable_shared_from_this, private asio }); } + void pack_and_response(std::string data) { + auto result = msgpack_codec::pack_args_str(result_code::OK, std::move(data)); + response(std::move(result)); + } + void set_conn_id(int64_t id) { conn_id_ = id; } int64_t conn_id() const { return conn_id_; } @@ -110,7 +115,7 @@ class connection : public std::enable_shared_from_this, private asio if (!ec) { router& _router = router::get(); - _router.route(body_.data(), length, this); + _router.route(body_.data(), length, this->shared_from_this()); } else { //LOG(INFO) << ec.message(); } diff --git a/include/router.h b/include/router.h index 473fda6..62f8c00 100644 --- a/include/router.h +++ b/include/router.h @@ -30,13 +30,13 @@ class router : asio::noncopyable { void remove_handler(std::string const& name) { this->map_invokers_.erase(name); } - void set_callback(const std::function& callback) { - callback_to_server_ = callback; - } - template - void route(const char* data, std::size_t size, T conn) { + void route(const char* data, std::size_t size, std::weak_ptr conn) { + auto conn_sp = conn.lock(); + if (!conn_sp) { + return; + } + std::string result; try { msgpack_codec codec; @@ -45,24 +45,22 @@ class router : asio::noncopyable { auto it = map_invokers_.find(func_name); if (it == map_invokers_.end()) { result = codec.pack_args_str(result_code::FAIL, "unknown function: " + func_name); - callback_to_server_(func_name, std::move(result), conn, true); + conn_sp->response(std::move(result)); return; } ExecMode model; it->second(conn, data, size, result, model); - if (model == ExecMode::sync && callback_to_server_) { + 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); - callback_to_server_(func_name, std::move(result), conn, true); - return; } - callback_to_server_(func_name, std::move(result), conn, false); + conn_sp->response(std::move(result)); } } catch (const std::exception& ex) { msgpack_codec codec; result = codec.pack_args_str(result_code::FAIL, ex.what()); - callback_to_server_("", std::move(result), conn, true); + conn_sp->response(std::move(result)); } } @@ -73,39 +71,39 @@ class router : asio::noncopyable { router(router&&) = delete; template - static typename std::result_of::type call_helper( - const F& f, const std::index_sequence&, const std::tuple& tup, connection* ptr) { + static typename std::result_of, Args...)>::type call_helper( + const F& f, const std::index_sequence&, const std::tuple& tup, std::weak_ptr ptr) { return f(ptr, std::get(tup)...); } template static - typename std::enable_if::type>::value>::type - call(const F& f, connection* ptr, std::string& result, std::tuple& tp) { + typename std::enable_if, Args...)>::type>::value>::type + call(const F& f, std::weak_ptr ptr, std::string& result, std::tuple& tp) { call_helper(f, std::make_index_sequence{}, tp, ptr); result = msgpack_codec::pack_args_str(result_code::OK); } template static - typename std::enable_if::type>::value>::type - call(const F& f, connection* ptr, std::string& result, const std::tuple& tp) { + typename std::enable_if, Args...)>::type>::value>::type + call(const F& f, std::weak_ptr ptr, std::string& result, const std::tuple& tp) { auto r = call_helper(f, std::make_index_sequence{}, tp, ptr); msgpack_codec codec; result = msgpack_codec::pack_args_str(result_code::OK, r); } template - static typename std::result_of::type call_member_helper( + static typename std::result_of, Args...)>::type call_member_helper( const F& f, Self* self, const std::index_sequence&, - const std::tuple& tup, connection* ptr = 0) { + const std::tuple& tup, std::weak_ptr ptr = 0) { return (*self.*f)(ptr, std::get(tup)...); } template static typename std::enable_if< - std::is_void::type>::value>::type - call_member(const F& f, Self* self, connection* ptr, std::string& result, + std::is_void, Args...)>::type>::value>::type + call_member(const F& f, Self* self, std::weak_ptr ptr, std::string& result, const std::tuple& tp) { call_member_helper(f, self, typename std::make_index_sequence{}, tp, ptr); result = msgpack_codec::pack_args_str(result_code::OK); @@ -113,8 +111,8 @@ class router : asio::noncopyable { template static typename std::enable_if< - !std::is_void::type>::value>::type - call_member(const F& f, Self* self, connection* ptr, std::string& result, + !std::is_void, Args...)>::type>::value>::type + call_member(const F& f, Self* self, std::weak_ptr ptr, std::string& result, const std::tuple& tp) { auto r = call_member_helper(f, self, typename std::make_index_sequence{}, tp, ptr); @@ -124,7 +122,7 @@ class router : asio::noncopyable { template struct invoker { template - static inline void apply(const Function& func, connection* conn, const char* data, size_t size, + static inline void apply(const Function& func, std::weak_ptr conn, const char* data, size_t size, std::string& result, ExecMode& exe_model) { using args_tuple = typename function_traits::args_tuple_2nd; exe_model = ExecMode::sync; @@ -141,7 +139,7 @@ class router : asio::noncopyable { } template - static inline void apply_member(const Function& func, Self* self, connection* conn, + static inline void apply_member(const Function& func, Self* self, std::weak_ptr conn, const char* data, size_t size, std::string& result, ExecMode& exe_model) { using args_tuple = typename function_traits::args_tuple_2nd; @@ -175,10 +173,8 @@ class router : asio::noncopyable { } std::unordered_map> + std::function, const char*, size_t, std::string&, ExecMode& model)>> map_invokers_; - std::function - callback_to_server_; }; } // namespace rpc_service } // namespace rest_rpc diff --git a/include/rpc_client.hpp b/include/rpc_client.hpp index 73a3d7e..fffd19a 100644 --- a/include/rpc_client.hpp +++ b/include/rpc_client.hpp @@ -36,7 +36,7 @@ namespace rest_rpc { string_view data_; }; - const constexpr size_t DEFAULT_TIMEOUT = 1000; //milliseconds + const constexpr size_t DEFAULT_TIMEOUT = 5000; //milliseconds class rpc_client : private asio::noncopyable { public: diff --git a/include/rpc_server.h b/include/rpc_server.h index 71ef1b3..9675468 100644 --- a/include/rpc_server.h +++ b/include/rpc_server.h @@ -11,6 +11,7 @@ using boost::asio::ip::tcp; namespace rest_rpc { namespace rpc_service { +using rpc_conn = std::weak_ptr; class rpc_server : private asio::noncopyable { public: rpc_server(short port, size_t size, size_t timeout_seconds = 15, size_t check_seconds = 10) @@ -18,8 +19,6 @@ class rpc_server : private asio::noncopyable { acceptor_(io_service_pool_.get_io_service(), tcp::endpoint(tcp::v4(), port)), timeout_seconds_(timeout_seconds), check_seconds_(check_seconds) { - router::get().set_callback(std::bind(&rpc_server::callback, this, std::placeholders::_1, - std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)); do_accept(); check_thread_ = std::make_shared([this] { clean(); }); } @@ -45,12 +44,6 @@ class rpc_server : private asio::noncopyable { router::get().register_handler(name, f, self); } - void response(int64_t conn_id, std::string&& result) { - std::unique_lock lock(mtx_); - auto it = connections_.find(conn_id); - if (it != connections_.end()) { it->second->response(std::move(result)); } - } - private: void do_accept() { conn_.reset(new connection(io_service_pool_.get_io_service(), timeout_seconds_)); @@ -60,7 +53,7 @@ class rpc_server : private asio::noncopyable { } else { conn_->start(); std::unique_lock lock(mtx_); - conn_->set_conn_id(conn_id_); + conn_->set_conn_id(conn_id_); connections_.emplace(conn_id_++, conn_); } @@ -83,11 +76,6 @@ class rpc_server : private asio::noncopyable { } } - void callback(const std::string& topic, std::string&& result, connection* conn, - bool has_error = false) { - response(conn->conn_id(), std::move(result)); - } - io_service_pool io_service_pool_; tcp::acceptor acceptor_; std::shared_ptr conn_;