diff --git a/.gitignore b/.gitignore index 63b31b2..0823d05 100644 --- a/.gitignore +++ b/.gitignore @@ -52,4 +52,8 @@ build/ **.user x64 -.cache \ No newline at end of file +.cache +# for clion +cmake-build-debug +cmake-build-release +cmake-build-debug-coverage \ No newline at end of file diff --git a/CMakeLists.txt b/CMakeLists.txt index 1d3991d..4b3a43b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -11,4 +11,7 @@ include(cmake/develop.cmake) if (BUILD_UNIT_TESTS) add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/tests) -endif () \ No newline at end of file +endif () +if (BUILD_EXAMPLES) + add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/examples) +endif () diff --git a/README.md b/README.md index 8c35c41..9929adc 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ | macOS Monterey 12 (AppleClang 14.0.0.14000029) | ![win](https://github.com/qicosmos/rest_rpc/actions/workflows/mac.yml/badge.svg?branch=master) | | Windows Server 2022 (MSVC 19.33.31630.0) | ![win](https://github.com/qicosmos/rest_rpc/actions/workflows/windows.yml/badge.svg?branch=master) | -c++11, high performance, cross platform, easy to use rpc framework. +c++20, high performance, cross platform, easy to use rpc framework. It's so easy to love RPC. @@ -15,7 +15,7 @@ Modern C++开发的RPC库就是这么简单好用！ # rest_rpc简介 -rest_rpc是一个高性能、易用、跨平台、header only的c++11 rpc库，它的目标是让tcp通信变得非常简单易用，即使不懂网络通信的人也可以直接使用它。它依赖header-only的standalone [asio](https://github.com/chriskohlhoff/asio)(tag:asio-1-36-0) +rest_rpc是一个高性能、易用、跨平台、header only的基于c++20 协程的rpc 库，它的目标是让tcp通信变得非常简单易用，即使不懂网络通信的人也可以直接使用它。它依赖header-only的standalone [asio](https://github.com/chriskohlhoff/asio)(tag:asio-1-36-0) 可以快速上手，使用者只需要关注自己的业务逻辑即可。 @@ -36,29 +36,36 @@ rest_rpc为用户提供了非常简单易用的接口，几行代码就可以实 ``` //服务端注册加法rpc服务 -struct dummy{ - int add(rpc_conn conn, int a, int b) { return a + b; } -}; +int add(rpc_conn conn, int a, int b) { return a + b; } int main(){ - rpc_server server(9000, std::thread::hardware_concurrency()); + rpc_server server("127.0.0.1:9004", std::thread::hardware_concurrency()); - dummy d; - server.register_handler("add", &dummy::add, &d); - - server.run(); + server.register_handler(); + + server.start(); } ``` ``` //客户端调用加法的rpc服务 int main(){ - rpc_client client("127.0.0.1", 9000); - client.connect(); - - int result = client.call("add", 1, 2); - - client.run(); + auto rpc_call = []() -> asio::awaitable { + rpc_client client; + auto ec = co_await client.connect("127.0.0.1:9004"); + if(ec) { + REST_LOG_ERROR << ec0.message(); + co_return; + } + + auto r = co_await client.call(1, 2); + if(r.ec==rpc_errc::ok) { + REST_LOG_INFO << "call result: " << r.value; + assert(r.value == 3); + } + }; + + sync_wait(get_global_executor(), rpc_call()); } ``` @@ -69,133 +76,134 @@ int main(){ //1.先定义person对象 struct person { - int id; - std::string name; - int age; - - MSGPACK_DEFINE(id, name, age); + int id; + std::string name; + int age; }; //2.提供并服务 -person get_person(rpc_conn conn) { - return { 1, "tom", 20 }; +person get_person(person p) { + p.name = "jack"; + return p; } int main(){ - //... - server.register_handler("get_person", get_person); + rpc_server server("127.0.0.1:9004", std::thread::hardware_concurrency()); + server.register_handler(); + server.start(); } ``` ``` //客户端调用获取person对象的rpc服务 int main(){ - rpc_client client("127.0.0.1", 9000); - client.connect(); - - person result = client.call("get_person"); - std::cout << result.name << std::endl; - - client.run(); -} -``` - -## 酷 - -异步？ - -同步？ - -future? - -callback？ - -当初为了提供什么样的接口在社区群里还争论了一番，有人希望提供callback接口，有人希望提供future接口，最后我 -决定都提供，专治强迫症患者:) - -现在想要的这些接口都给你提供了，你想用什么类型的接口就用什么类型的接口，够酷吧，让我们来看看怎么用这些接口吧： - -``` -//服务端提供echo服务 -std::string echo(rpc_conn conn, const std::string& src) { - return src; -} - -server.register_handler("echo", echo); -``` - -客户端同步接口 - -``` -auto result = client.call("echo", "hello"); -``` - -客户端异步回调接口 - -``` -client.async_call("echo", [](asio::error_code ec, R data){ - std::cout << "echo " << data << '\n'; -}); -``` - -## async_call接口说明 -有两个重载的async_call接口，一个是返回future的接口，一个是带超时的异步接口。 - -返回future的async_call接口： -``` -std::future future = client.async_call("echo", "purecpp"); -``` - -带超时的异步回调接口： -``` -async_call("some_rpc_service_name", callback, service_args...); -``` - -如果不显式设置超时时间的话，则会用默认的5s超时. -``` -async_call("some_rpc_service_name", callback, args...); -``` - -``` -client.async_call("echo", [](asio::error_code ec, std::string result) { - if (ec) { - std::cout << ec.message() <<" "<< data << "\n"; - return; + auto rpc_call = []() -> asio::awaitable { + rpc_client client; + auto ec = co_await client.connect("127.0.0.1:9004"); + if(ec) { + REST_LOG_ERROR << ec0.message(); + co_return; } - - std::cout << result << " async\n"; -}, "purecpp"); -``` - -客户端异步future接口 - -``` -auto future = client->async_call("echo", "hello"); -auto status = future.wait_for(std::chrono::seconds(2)); -if (status == std::future_status::timeout) { - std::cout << "timeout\n"; -} -else if (status == std::future_status::ready) { - auto str = future.get().as(); - std::cout << "echo " << str << '\n'; + + person p{1, "tom", 20}; + auto r = co_await client.call(p); + if(r.ec==rpc_errc::ok) { + REST_LOG_INFO << "call result: " << r.value.name; + assert(r.value.name == "jack"); + } + }; + + sync_wait(get_global_executor(), rpc_call()); } ``` -除了上面的这些很棒的接口之外，更酷的是rest_rpc还支持了订阅发布的功能，这是目前很多rpc库做不到的。 +## 发布订阅 +以订阅某个topic为例： -服务端订阅发布的例子在这里： +server 端代码： +```cpp +void publish() { + rpc_server server("127.0.0.1:9004", 4); + server.async_start(); + + REST_LOG_INFO << "will pubish, waiting for input"; + auto pub = [&]() -> asio::awaitable { + std::string str; + while (true) { + std::cin >> str; + if(str == "quit") { + break; + } + + co_await server.publish("topic1", str);// 向客户端发布一个string，你也可以发布一个对象，内部会自动序列化 + } + }; + + sync_wait(get_global_executor(), pub()); +} -https://github.com/qicosmos/rest_rpc/blob/master/examples/server/main.cpp#L121 -https://github.com/qicosmos/rest_rpc/blob/master/examples/client/main.cpp#L383 +client 端代码： +```cpp +void subscribe() { + REST_LOG_INFO << "will subscribe, waiting for publish"; + auto sub = [&]() -> asio::awaitable { + rpc_client client; + co_await client.connect("127.0.0.1:9004"); + while (true) { + // 订阅topic1，会自动将结果反序列化为std::string, 如果publish是一个person对象，则subscribe参数填person，内部会自动反序列化 + auto [ec, result] = co_await client.subscribe("topic1"); + if (ec != rpc_errc::ok) { + REST_LOG_ERROR << "subscribe failed: " << make_error_code(ec).message(); + break; + } + + REST_LOG_INFO << result; + } + }; + + sync_wait(get_global_executor(), sub()); +} +``` ## 快 rest_rpc是目前最快的rpc库，具体和grpc和brpc做了性能对比测试，rest_rpc性能是最高的，远超grpc。 -性能测试的结果在这里： +性能测试代码在这里： -https://github.com/qicosmos/rest_rpc/blob/master/doc/%E5%8D%95%E6%9C%BA%E4%B8%8Arest_rpc%E5%92%8Cbrpc%E6%80%A7%E8%83%BD%E6%B5%8B%E8%AF%95.md +https://github.com/qicosmos/rest_rpc/tree/master/tests/bench.cpp +## 使用自己的序列化库 +rest_rpc 默认使用yalantinglibs的struct_pack 去做系列化/反序列化的，它的性能非常好。 + +rest_rpc 也支持用户使用自己的序列化库，只需要去实现一个序列化和一个反序列化函数。 +```cpp + namespace user_codec { + // adl lookup in user_codec namespace + template + std::string serialize(rest_adl_tag, Args &&...args) { + msgpack::sbuffer buffer(2 * 1024); + if constexpr (sizeof...(Args) > 1) { + msgpack::pack(buffer, std::forward_as_tuple(std::forward(args)...)); + } else { + msgpack::pack(buffer, std::forward(args)...); + } + + return std::string(buffer.data(), buffer.size()); + } + + template T deserialize(rest_adl_tag, std::string_view data) { + try { + static msgpack::unpacked msg; + msgpack::unpack(msg, data.data(), data.size()); + return msg.get().as(); + } catch (...) { + return T{}; + } + } + } // namespace user_codec +``` +实现这两个函数之后rest_rpc 将会使用自定义的序列化/反序列化函数了。 # rest_rpc的更多用法 @@ -203,9 +211,6 @@ https://github.com/qicosmos/rest_rpc/blob/master/doc/%E5%8D%95%E6%9C%BA%E4%B8%8A https://github.com/qicosmos/rest_rpc/tree/master/examples -# future - -make an IDL tool to genrate the client code. ## 社区和群 purecpp.cn diff --git a/doc/单机上rest_rpc和brpc性能测试.md b/doc/单机上rest_rpc和brpc性能测试.md deleted file mode 100644 index 590c2dc..0000000 --- a/doc/单机上rest_rpc和brpc性能测试.md +++ /dev/null @@ -1,330 +0,0 @@ -单机上[rest_rpc](https://github.com/qicosmos/rest_rpc "rest_rpc")和[brpc](https://github.com/apache/incubator-brpc "brpc")性能测试 - -# 测试环境 - - | 软硬件环境 | 参数 | - | -------- | :-----: | - | OS | 18.04.1-Ubuntu | - | CPU | 6Core，Intel(R) Core(TM) i5-9400F CPU @ 2.90GHz | - | 内存 | 16G | - | g++版本 | g++ (Ubuntu 8.2.0-1ubuntu2~18.04) 8.2.0 | - | 机器类别 | 实体机 | - -# 测试方法 - -1. server与client均为单进程并且部署在同一台实体机； -2. server开启多线程处理，线程数是CPU核数； -3. client是开启单线程与多线程分别测试性能； -4. client循环向server发送一定字节的字符串，比如由‘A’构成的1K字符串，server收到后不做业务处理，立即返回给client，client收到server返回的字符串也不做业务处理； -5. 当循环结束后，client统计延时； -6. server每隔1s会向标准输出设备打印出收到的请求数； -7. 采用rest_rpc与brpc默认的编译方式（cmake -DCMAKE_BUILD_TYPE=Release）。 -8. rest_rpc采用高效模式与brpc进行对比测试。 - -# 测试代码编写及分析 - -## rest_rpc测试代码 - -1. 下载源码：https://github.com/qicosmos/rest_rpc -2. 解压后进入examples/server目录，修改main.cpp，增加函数： -``` -std::string echo(connection* conn, const std::string& orignal) { - g_qps.increase(); - return orignal; -} -``` -3. echo函数将客户端从conn连接传过来的字符串original原封不动得返回，并调用increase函数进行请求计数。 -在main函数增加 - -``` -server.register_handler("echo", echo); -``` - -用于向server注册消息处理函数。 - -4. examples/client目录，修改main.cpp，client为单线程模式： - -在test_performance1()函数中进行请求发送： -``` -void test_performance1() { - rpc_client client("127.0.0.1", 9000); - bool r = client.connect(); - if (!r) { - std::cout << "connect timeout" << std::endl; - return; - } - string str(1024, 'A'); - auto begin = high_resolution_clock::now(); - for (size_t i = 0; i < LOOP; i++) { - auto future = client.async_call("echo", str); - auto status = future.wait_for(std::chrono::seconds(2)); - if (status == std::future_status::deferred) { - std::cout << "deferred\n"; - } - else if (status == std::future_status::timeout) { - std::cout << "timeout\n"; - } - else if (status == std::future_status::ready) { - } - } - auto end = high_resolution_clock::now(); - cout << "elapse time = " << duration_cast(end - begin).count() << "s" << endl; - std::cout << "finish\n"; -} -``` - -client为多线程模式： -``` -void test_performance1() { - vectorclients; - auto begin = high_resolution_clock::now(); - for (int i = 0;i < THREADNUM;i++) - { - clients.emplace_back(thread(client_thread)); - } - - for (auto &it : clients) - { - it.join(); - } - auto end = high_resolution_clock::now(); - cout << "elapse time = " << duration_cast(end - begin).count() << "s" << endl; - cout << "finish\n"; -} -``` - -定义循环次数为： static const int LOOP = 100000; -定义线程数：const static int THREADNUM = 6; -编写client_thread： - -## brpc测试代码 - -1. 下载源码：https://github.com/apache/incubator-brpc.git -2. 按官网编译指令下载相应包后进行编译 - -进入目录examples/echo_c++，修改server.cpp，在 -``` -virtual void Echo(google::protobuf::RpcController* cntl_base, - const EchoRequest* request, EchoResponse* response, google::protobuf::Closure* done) -注释相应LOG日志： - virtual void Echo(google::protobuf::RpcController* cntl_base, - const EchoRequest* request, - EchoResponse* response, - google::protobuf::Closure* done) { - // This object helps you to call done->Run() in RAII style. If you need - // to process the request asynchronously, pass done_guard.release(). - brpc::ClosureGuard done_guard(done); - - brpc::Controller* cntl = - static_cast(cntl_base); - - // The purpose of following logs is to help you to understand - // how clients interact with servers more intuitively. You should - // remove these logs in performance-sensitive servers. - /*LOG(INFO) << "Received request[log_id=" << cntl->log_id() - << "] from " << cntl->remote_side() - << " to " << cntl->local_side() - << ": " << request->message() - << " (attached=" << cntl->request_attachment() << ")";*/ - - // Fill response. - response->set_message(request->message()); - g_qps.increase(); - // You can compress the response by setting Controller, but be aware - // that compression may be costly, evaluate before turning on. - // cntl->set_response_compress_type(brpc::COMPRESS_TYPE_GZIP); - - if (FLAGS_echo_attachment) { - // Set attachment which is wired to network directly instead of - // being serialized into protobuf messages. - cntl->response_attachment().append(cntl->request_attachment()); - } -} -``` -代码行response->set_message(request->message());将将客户端的请求原封不动发回客户端。 - -3. 修改client.cpp，单线程模式： -``` -const static size_t BUF_SIZE = 1024; -const static int LOOP = 1000000; -char buf[BUF_SIZE] = "" -修改main.函数：： -int main(int argc, char* argv[]) { - memset(buf, 'A', BUF_SIZE); - // Parse gflags. We recommend you to use gflags as well. - GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true); - - // A Channel represents a communication line to a Server. Notice that - // Channel is thread-safe and can be shared by all threads in your program. - brpc::Channel channel; - - // Initialize the channel, NULL means using default options. - brpc::ChannelOptions options; - - options.protocol = FLAGS_protocol; - options.connection_type = FLAGS_connection_type; - options.timeout_ms = FLAGS_timeout_ms/*milliseconds*/; - options.max_retry = FLAGS_max_retry; - if (channel.Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &options) != 0) { - LOG(ERROR) << "Fail to initialize channel"; - return -1; - } - - // Normally, you should not call a Channel directly, but instead construct - // a stub Service wrapping it. stub can be shared by all threads as well. - example::EchoService_Stub stub(&channel); - - // Send a request and wait for the response every 1 second. - //int log_id = 0; - - example::EchoRequest request; - example::EchoResponse response; - request.set_message(buf); - auto begin = high_resolution_clock::now(); - for (int i = 0; i < LOOP; i++) - { - brpc::Controller cntl; - cntl.request_attachment().append(FLAGS_attachment); - stub.Echo(&cntl, &request, &response, NULL); - if (!cntl.Failed()) { - /*LOG(INFO) << "Received response from " << cntl.remote_side() - << " to " << cntl.local_side() - << ": " << response.message() << " (attached=" - << cntl.response_attachment() << ")" - << " latency=" << cntl.latency_us() << "us";*/ - } - else { - LOG(WARNING) << cntl.ErrorText(); - } - } - auto end = high_resolution_clock::now(); - cout << "elapse time = " << duration_cast(end - begin).count() << "s" << endl; - std::cout << "finish\n"; - //LOG(INFO) << "EchoClient is going to quit"; - return 0; -} -``` - -去除日志，并统计循环发送延迟，在循环体内只是向server发送报文，并未执行任何业务。 - -多线程模式 - -``` -#include -#include -#include -#include -#include -#include -#include -#include "echo.pb.h" -using namespace std; -using namespace chrono; -DEFINE_string(attachment, "", "Carry this along with requests"); -DEFINE_string(protocol, "baidu_std", "Protocol type. Defined in src/brpc/options.proto"); -DEFINE_string(connection_type, "", "Connection type. Available values: single, pooled, short"); -//DEFINE_string(server, "0.0.0.0:8000", "IP Address of server"); -DEFINE_string(server, "127.0.0.1:9000", "IP Address of server"); -DEFINE_string(load_balancer, "", "The algorithm for load balancing"); -DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds"); -DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)"); -DEFINE_int32(interval_ms, 1000, "Milliseconds between consecutive requests"); -//const static size_t BUF_SIZE = 1024 * 8 * 2 * 4; -const static size_t BUF_SIZE = 1024; -const static int LOOP = 100000; -const static int THREADNUM = 20; -char buf[BUF_SIZE] = ""; -void client_thread() -{ - brpc::Channel channel; - brpc::ChannelOptions options; - - options.protocol = FLAGS_protocol; - options.connection_type = FLAGS_connection_type; - options.timeout_ms = FLAGS_timeout_ms/*milliseconds*/; - options.max_retry = FLAGS_max_retry; - if (channel.Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &options) != 0) { - LOG(ERROR) << "Fail to initialize channel"; - return; - } - example::EchoService_Stub stub(&channel); - example::EchoRequest request; - example::EchoResponse response; - request.set_message(buf); - for (int i = 0; i < LOOP; i++) - { - brpc::Controller cntl; - cntl.request_attachment().append(FLAGS_attachment); - stub.Echo(&cntl, &request, &response, NULL); - if (!cntl.Failed()) { - } - else { - LOG(WARNING) << cntl.ErrorText(); - } - } -} -int main(int argc, char* argv[]) { - memset(buf, 'A', BUF_SIZE); - GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true); - vectorclients; - auto begin = high_resolution_clock::now(); - for (int i = 0;i < THREADNUM;i++) - { - clients.emplace_back(thread(client_thread)); - } - for (auto &it : clients) - { - it.join(); - } - auto end = high_resolution_clock::now(); - cout << "elapse time = " << duration_cast(end - begin).count() << "s" << endl; - cout << "finish\n"; -} -``` - -# 测试结果 - -## brpc和grpc的[测试结果](https://github.com/apache/incubator-brpc/blob/master/docs/cn/benchmark.md "测试结果") - - ![alt](https://github.com/apache/incubator-brpc/blob/master/docs/images/qps_vs_threadnum.png "多线程") - -## rest_rpc和brpc的测试结果 - -本次测试使用相同环境与方法对rest_rpc与brpc进行性能测试，最终测试数据取多次测试数据的平均值，测试数据如下： - -**client使用单进程单线程测试模式** - - | 消息字节 | QPS(rest_rpc) | QPS(brpc) | - | -------- | :-----: | :-----: | - | 1K | 52631 | 41667 | - | 8K | 42863 | 34483 | - | 16K | 37037 | 29412 | - | 64K | 16949 | 8982 | - - ![alt](https://github.com/qicosmos/rest_rpc/blob/master/doc/%E5%8D%95%E7%BA%BF%E7%A8%8B.png "单线程") - -从相关数据可以看出，rest_rpc QPS要高于brpc，特别是当消息字节达到64K时，rest QPS几乎是brpc的2倍。 - -**client多线程测试1K消息(rest_rpc采用高效模式)** - - | 线程数 | QPS(rest_rpc) | QPS(brpc) | - | -------- | :-----: | :-----: | - | 6 | 200000 | 103448 | - | 10 | 250000 | 166667 | - | 20 | 285714 | 240963 | - - ![alt](https://github.com/qicosmos/rest_rpc/blob/master/doc/%E5%A4%9A%E7%BA%BF%E7%A8%8B1k.png "多线程1k") - -**client多线程测试64K消息(rest_rpc采用高效模式)** - - | 线程数 | QPS(rest_rpc) | QPS(brpc) | - | -------- | :-----: | :-----: | - | 6 | 60000 | 34883 | - | 10 | 76923 | 40000 | - | 20 | 59405 | 35242 | - - ![alt](https://github.com/qicosmos/rest_rpc/blob/master/doc/%E5%A4%9A%E7%BA%BF%E7%A8%8B64k.png "多线程64k") - - - - diff --git a/doc/单线程.png b/doc/单线程.png deleted file mode 100644 index 5233d7e..0000000 Binary files a/doc/单线程.png and /dev/null differ diff --git a/doc/多线程1k.png b/doc/多线程1k.png deleted file mode 100644 index b506331..0000000 Binary files a/doc/多线程1k.png and /dev/null differ diff --git a/doc/多线程64k.png b/doc/多线程64k.png deleted file mode 100644 index c6eba97..0000000 Binary files a/doc/多线程64k.png and /dev/null differ diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt new file mode 100644 index 0000000..da2f04e --- /dev/null +++ b/examples/CMakeLists.txt @@ -0,0 +1,10 @@ +project(example) +set(project_name test_rest_rpc) + +include_directories( + "../thirdparty/asio" + "../thirdparty" + ) + +add_executable(server server.cpp) +add_executable(client client.cpp) diff --git a/examples/client.cpp b/examples/client.cpp new file mode 100644 index 0000000..ebd98e0 --- /dev/null +++ b/examples/client.cpp @@ -0,0 +1,59 @@ +#include +using namespace rest_rpc; + +// Only need a declaration. +std::string_view echo(std::string_view str); +int add(int a, int b); + +void basic_usage() { + auto rpc_call = []() -> asio::awaitable { + rpc_client client; + auto ec0 = co_await client.connect("127.0.0.1:9004"); + if(ec0) { + REST_LOG_ERROR << ec0.message(); + co_return; + } + + auto r = co_await client.call("test"); + if(r.ec==rpc_errc::ok) { + REST_LOG_INFO << "call result: " << r.value; + assert(r.value == "test"); + } + + auto r1 = co_await client.call(1, 2); + if(r1.ec==rpc_errc::ok) { + REST_LOG_INFO << "call result: " << r1.value; + assert(r1.value == 3); + } + }; + + sync_wait(get_global_executor(), rpc_call()); +} + +void subscribe() { + REST_LOG_INFO << "will subscribe, waiting for publish"; + auto sub = [&]() -> asio::awaitable { + rpc_client client; + co_await client.connect("127.0.0.1:9004"); + while (true) { + auto [ec, result] = co_await client.subscribe("topic1"); + if (ec != rpc_errc::ok) { + REST_LOG_ERROR << "subscribe failed: " << make_error_code(ec).message(); + break; + } + + if (result == "close") { + co_return; + } + + REST_LOG_INFO << result; + } + }; + + sync_wait(get_global_executor(), sub()); +} + +int main() { + basic_usage(); + subscribe(); +} \ No newline at end of file diff --git a/examples/server.cpp b/examples/server.cpp new file mode 100644 index 0000000..9f88639 --- /dev/null +++ b/examples/server.cpp @@ -0,0 +1,97 @@ +#include +using namespace rest_rpc; + +std::string_view echo(std::string_view str) { + return str; +} + +struct dummy{ + int add(int a, int b) { return a + b; } +}; +int add(int a, int b) { return a + b; } + +struct person { + int id; + std::string name; + int age; +}; + +person get_person(person p) { + p.name = "jack"; + return p; +} + +void basict_usage() { + rpc_server server("127.0.0.1:9004", 4); + server.register_handler(); + + dummy d{}; + server.register_handler<&dummy::add>(&d); + + server.register_handler(); + + auto ec = server.async_start(); + if(ec) { + REST_LOG_ERROR << ec.message(); + return; + } + + auto rpc_call = []() -> asio::awaitable { + rpc_client client; + auto ec0 = co_await client.connect("127.0.0.1:9004"); + if(ec0) { + REST_LOG_ERROR << ec0.message(); + co_return; + } + + auto r = co_await client.call("test"); + if(r.ec==rpc_errc::ok) { + REST_LOG_INFO << "call result: " << r.value; + assert(r.value == "test"); + } + + auto r1 = co_await client.call<&dummy::add>(1, 2); + if(r1.ec==rpc_errc::ok) { + REST_LOG_INFO << "call result: " << r1.value; + assert(r1.value == 3); + } + + person p{1, "tom", 20}; + auto r2 = co_await client.call(p); + if(r2.ec==rpc_errc::ok) { + REST_LOG_INFO << "call result: " << r2.value.name; + assert(r2.value.name == "jack"); + } + }; + + sync_wait(get_global_executor(), rpc_call()); +} + +void publish() { + rpc_server server("127.0.0.1:9004", 4); + server.register_handler(); + server.register_handler(); + server.async_start(); + + REST_LOG_INFO << "will pubish, waiting for input"; + auto pub = [&]() -> asio::awaitable { + std::string str; + while (true) { + std::cin >> str; + if(str == "quit") { + break; + } + + co_await server.publish("topic1", str); + } + }; + + sync_wait(get_global_executor(), pub()); + server.stop(); +} + +int main() { + basict_usage(); + + publish(); +} \ No newline at end of file diff --git a/include/rest_rpc.hpp b/include/rest_rpc.hpp index cd998b6..2392021 100644 --- a/include/rest_rpc.hpp +++ b/include/rest_rpc.hpp @@ -1,2 +1,2 @@ -#include "rest_rpc/rest_rpc_client.hpp" -#include "rest_rpc/rest_rpc_server.h" \ No newline at end of file +#include "rest_rpc/rpc_client.hpp" +#include "rest_rpc/rpc_server.hpp" \ No newline at end of file diff --git a/include/rest_rpc/rpc_client.hpp b/include/rest_rpc/rpc_client.hpp index 73c6d9d..e319d3d 100644 --- a/include/rest_rpc/rpc_client.hpp +++ b/include/rest_rpc/rpc_client.hpp @@ -20,9 +20,7 @@ template struct call_result { R value; }; -template <> struct call_result { - rpc_errc ec; -}; +template <> struct call_result { rpc_errc ec; }; class rpc_client { public: @@ -218,12 +216,14 @@ private: socket_->impl_, asio::buffer(&resp_header, sizeof(rest_rpc_header)), asio::as_tuple(asio::use_awaitable)); if (ec) { - result.ec = rpc_errc::write_error; + result.ec = rpc_errc::read_error; close_socket(*socket_); + comple_all(); co_return result; } if (resp_header.magic != 39) { result.ec = rpc_errc::protocol_error; + comple_all(); co_return result; } @@ -240,6 +240,7 @@ private: REST_LOG_WARNING << "read body error: " << ec.message(); result.ec = rpc_errc::read_error; close_socket(*socket_); + comple_all(); co_return result; } result.ec = (rpc_errc)socket_->body_[0]; @@ -257,6 +258,12 @@ private: co_return std::move(result); } + void comple_all() { + for (auto &pair : socket_->sub_ops_) { + pair.second.complete(false); + } + } + asio::awaitable watchdog(auto duration) { asio::steady_timer timer(socket_->get_executor()); timer.expires_after(duration); diff --git a/include/rest_rpc/use_asio.hpp b/include/rest_rpc/use_asio.hpp index 16ffd6b..6f03553 100644 --- a/include/rest_rpc/use_asio.hpp +++ b/include/rest_rpc/use_asio.hpp @@ -18,23 +18,3 @@ using ssl_socket = asio::ssl::stream; #endif #include - -#ifdef CINATRA_ENABLE_SSL -#if __cplusplus > 201402L -#if defined(__GNUC__) -#if __GNUC__ < 8 -#include -namespace rpcfs = std::experimental::filesystem; -#else -#include -namespace rpcfs = std::filesystem; -#endif -#else -#include -namespace rpcfs = boost::filesystem; -#endif -#else -#include -namespace rpcfs = boost::filesystem; -#endif -#endif