[cross language]Add py and go example for http and rpc (#949)

Author: qicosmos

README.md

@@ -543,6 +543,28 @@ options:
 ./benchmark_client # [threads = hardware counts] [client_pre_thread = 20] [pipeline_size = 1] [host = 127.0.0.1] [port = 9000] [test_data_path = ./test_data/echo_test] [test_seconds = 30] [warm_up_seconds = 5]
 ```
 
+## cross language support
+### how to use coro_http with python
+See `yalantinglibs/src/coro_http/examples/py_example`
+
+#### build and run py_example
+- Set option `ENABLE_pybind11`(in `yalantinglibs/src/coro_http/examples/CMakeLists.txt`) `ON` and then build `py_example`.
+
+- go to `yalantinglibs/src/coro_http/examples/py_example`
+
+- python3 test.py
+
+### how to use coro_rpc with golang
+See `yalantinglibs/src/coro_rpc/examples/basic_example/go_example`
+
+#### build and run go_example
+- Set option `ENABLE_go`(in `yalantinglibs/src/coro_rpc/examples/base_examples/CMakeLists.txt`) `ON` and then build `coro_rpc`.
+
+- go to `yalantinglibs/src/coro_rpc/examples/base_examples/go_example`
+
+- go build test_rpc.go
+
+- ./test_rpc
 
 ## How to generate document
 

src/coro_http/examples/CMakeLists.txt

@@ -35,4 +35,29 @@ if (CMAKE_CXX_COMPILER_ID STREQUAL "GNU" AND CMAKE_SYSTEM_NAME MATCHES "Windows"
     target_link_libraries(coro_http_example wsock32 ws2_32)
     target_link_libraries(coro_http_load_balancer wsock32 ws2_32)
     target_link_libraries(coro_chat_room wsock32 ws2_32)
+endif()
+
+option(ENABLE_pybind11 "Enable pybind11 " OFF)
+if(ENABLE_pybind11)
+  cmake_minimum_required(VERSION 3.14)
+  project(MyPybind11Project)
+
+  include(FetchContent)
+
+  FetchContent_Declare(
+    pybind11
+    GIT_REPOSITORY https://github.com/pybind/pybind11.git
+    GIT_TAG v2.13.6
+  )
+
+  FetchContent_MakeAvailable(pybind11)
+
+  include_directories(${CMAKE_CURRENT_BINARY_DIR}/_deps/pybind11-src/include)
+
+  pybind11_add_module(py_example py_example/py_example.cpp)
+  set_target_properties(py_example PROPERTIES
+    LIBRARY_OUTPUT_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/py_example
+    ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/py_example
+    RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/py_example
+  )
 endif()

src/coro_http/examples/py_example/py_example.cpp

@@ -0,0 +1,44 @@
+#include <pybind11/pybind11.h>
+
+#include <string>
+#include <ylt/coro_http/coro_http_client.hpp>
+#include <ylt/coro_io/client_pool.hpp>
+
+namespace py = pybind11;
+
+class caller {
+ public:
+  caller(py::function callback) : callback_(std::move(callback)) {}
+  void async_get(std::string url) {
+    static auto pool =
+        coro_io::client_pool<coro_http::coro_http_client>::create(url);
+    pool->send_request([this, url](coro_http::coro_http_client &client)
+                           -> async_simple::coro::Lazy<void> {
+          auto result = co_await client.async_get(url);
+
+          // PyObject *py_str = PyUnicode_FromStringAndSize(
+          //     result.resp_body.data(), result.resp_body.size()); // zero copy
+          // py::gil_scoped_acquire acquire;
+          // callback_(py::reinterpret_steal<py::object>(py_str));
+
+          py::gil_scoped_acquire acquire;
+          callback_(result.resp_body);
+        })
+        .start([](auto &&) {
+        });
+  }
+
+ private:
+  py::function callback_;
+};
+
+PYBIND11_MODULE(py_example, m) {
+  m.def("hello", [] {
+    return std::string("hello");
+  });
+
+  py::class_<caller>(m, "caller")
+      .def(py::init<py::function>())
+      .def("async_get", &caller::async_get,
+           py::call_guard<py::gil_scoped_release>());
+}

src/coro_http/examples/py_example/test.py

@@ -0,0 +1,22 @@
+import asyncio
+import py_example
+
+async def async_get(url):
+    loop = asyncio.get_running_loop()
+    future = loop.create_future()    
+    def cpp_callback(message):
+        def set_result():
+            future.set_result(message)
+
+        loop.call_soon_threadsafe(set_result)
+
+    caller = py_example.caller(cpp_callback)
+    caller.async_get(url)
+    result = await future
+    print(result)
+
+async def main():
+    await async_get("http://taobao.com")
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/coro_rpc/examples/base_examples/CMakeLists.txt

@@ -36,6 +36,14 @@ else()
   endif()
 endif()
 
+option(ENABLE_go "Enable go " OFF)
+if(ENABLE_go)
+  add_library(coro_rpc coro_rpc_lib/coro_rpc.cpp)
+  set_target_properties(coro_rpc PROPERTIES
+    ARCHIVE_OUTPUT_DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}/go_example"
+  )
+endif()
+
 add_executable(coro_rpc_example_load_balancer load_balancer.cpp)
 add_executable(coro_rpc_example_client_pool client_pool.cpp)
 add_executable(coro_rpc_example_client_pools client_pools.cpp)

src/coro_rpc/examples/base_examples/coro_rpc_lib/coro_rpc.cpp

@@ -0,0 +1,148 @@
+#include "coro_rpc.h"
+
+#include <ylt/coro_io/load_balancer.hpp>
+#include <ylt/coro_rpc/coro_rpc_client.hpp>
+#include <ylt/coro_rpc/coro_rpc_server.hpp>
+
+inline char *create_copy_cstr(std::string msg) {
+  char *buf = (char *)malloc(msg.size() + 1);
+  memcpy(buf, msg.data(), msg.size());
+  buf[msg.size()] = '\0';
+  return buf;
+}
+
+inline void set_rpc_result(rpc_result &ret, coro_rpc::rpc_error ec) {
+  ret.ec = ec.val();
+  ret.err_msg = create_copy_cstr(ec.msg);
+}
+
+// rpc server
+inline void ylt_load_service(coro_rpc::context<void> context, uint64_t req_id) {
+  auto ctx_ptr = std::make_unique<coro_rpc::context<void>>(std::move(context));
+  load_service(ctx_ptr.release(), req_id);
+}
+
+void response_error(void *ctx, uint16_t err_code, const char *err_msg) {
+  std::unique_ptr<coro_rpc::context<void>> context(
+      (coro_rpc::context<void> *)ctx);
+
+  context->response_error(coro_rpc::err_code(err_code), err_msg);
+}
+
+void *response_msg(void *ctx, char *msg, uint64_t size) {
+  std::unique_ptr<coro_rpc::context<void>> context(
+      (coro_rpc::context<void> *)ctx);
+  // wait for response finish;
+  auto promise = std::make_unique<std::promise<std::error_code>>();
+  context->get_context_info()->set_complete_handler(
+      [p = promise.get()](const std::error_code &ec, std::size_t) {
+        p->set_value(ec);
+      });
+
+  context->get_context_info()->set_response_attachment(
+      std::string_view(msg, size));
+  context->response_msg();
+  return promise.release();
+}
+
+rpc_result wait_response_finish(void *p) {
+  auto promis = (std::promise<std::error_code> *)p;
+  auto ec = promis->get_future().get();
+  delete promis;
+  if (!ec) {
+    return {};
+  }
+
+  char *buf = create_copy_cstr(ec.message());
+  return rpc_result{0, buf, 0};
+}
+
+void *start_rpc_server(char *addr, int parallel) {
+  auto server = std::make_unique<coro_rpc::coro_rpc_server>(
+      parallel, addr, std::chrono::seconds(600));
+  server->register_handler<ylt_load_service>();
+  auto res = server->async_start();
+  if (res.hasResult()) {
+    ELOG_ERROR << "start server failed";
+    return nullptr;
+  }
+
+  return server.release();
+}
+
+void stop_rpc_server(void *server) {
+  if (server != nullptr) {
+    auto s = (coro_rpc::coro_rpc_server *)server;
+    s->stop();
+    delete s;
+  }
+}
+
+// rpc client
+void *create_client_pool(char *addr, int req_timeout_sec) {
+  std::vector<std::string_view> hosts{std::string_view(addr)};
+  auto ld = coro_io::load_balancer<coro_rpc::coro_rpc_client>::create(
+      hosts, {.pool_config{.client_config = {
+                               .request_timeout_duration =
+                                   std::chrono::seconds{req_timeout_sec}}}});
+  auto ld_ptr = std::make_unique<decltype(ld)>(std::move(ld));
+  return ld_ptr.release();
+}
+
+void free_client_pool(void *pool) {
+  auto ptr = (coro_io::load_balancer<coro_rpc::coro_rpc_client> *)pool;
+  delete ptr;
+}
+
+rpc_result load(void *pool, uint64_t req_id, char *dest, uint64_t dest_len) {
+  using namespace async_simple::coro;
+  rpc_result ret{};
+  if (dest == nullptr || dest_len == 0) {
+    set_rpc_result(ret,
+                   coro_rpc::rpc_error(coro_rpc::errc::invalid_rpc_arguments));
+    return ret;
+  }
+
+  auto ld = (coro_io::load_balancer<coro_rpc::coro_rpc_client> *)pool;
+  auto lazy = [&]() -> Lazy<void> {
+    auto result =
+        co_await ld->send_request([&](coro_rpc::coro_rpc_client &client,
+                                      std::string_view hostname) -> Lazy<void> {
+          client.set_resp_attachment_buf(std::span<char>(dest, dest_len));
+          auto result = co_await client.call<ylt_load_service>(req_id);
+          if (!result) {
+            set_rpc_result(ret, result.error());
+            co_return;
+          }
+
+          if (!client.is_resp_attachment_in_external_buf()) {
+            set_rpc_result(
+                ret, coro_rpc::rpc_error(coro_rpc::errc::message_too_large));
+            co_return;
+          }
+          ret.len = client.get_resp_attachment().size();
+        });
+  };
+
+  syncAwait(lazy());
+
+  return ret;
+}
+
+// log
+void init_rpc_log(char *log_filename, int log_level, uint64_t max_file_size,
+                  uint64_t max_file_num, bool async) {
+  if (log_filename == nullptr) {
+    easylog::set_min_severity((easylog::Severity)log_level);
+    ELOG_INFO << "init log, log_level: " << log_level;
+    return;
+  }
+
+  std::string filename(log_filename);
+  easylog::init_log((easylog::Severity)log_level, filename, async, true,
+                    max_file_size, max_file_num, false);
+  ELOG_INFO << "init log " << log_level << ", " << filename << ","
+            << max_file_size << "," << max_file_num;
+}
+
+void flush_rpc_log() { easylog::flush(); }

src/coro_rpc/examples/base_examples/coro_rpc_lib/coro_rpc.h

@@ -0,0 +1,48 @@
+#pragma once
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdlib.h>
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct {
+  int ec;
+  char *err_msg;
+  uint64_t len;
+} rpc_result;
+
+// ip:port
+extern void load_service(void *ctx, uint64_t req_id);
+extern void *response_msg(void *ctx, char *msg, uint64_t size);
+extern void response_error(void *ctx, uint16_t err_code, const char *err_msg);
+extern rpc_result wait_response_finish(void *p);
+extern void *start_rpc_server(char *addr, int parallel);
+extern void stop_rpc_server(void *server);
+
+extern void *create_client_pool(char *addr, int req_timeout_sec);
+extern void free_client_pool(void *pool);
+extern rpc_result load(void *pool, uint64_t req_id, char *dest,
+                       uint64_t dest_len);
+
+/*
+enum log_level {
+  NONE = 0,
+  TRACE,
+  DEBUG,
+  INFO,
+  WARN,
+  WARNING = WARN,
+  ERROR,
+  CRITICAL,
+  FATAL = CRITICAL,
+};
+*/
+extern void init_rpc_log(char *log_filename, int log_level,
+                         uint64_t max_file_size, uint64_t max_file_num,
+                         bool async);
+extern void flush_rpc_log();
+
+#ifdef __cplusplus
+}
+#endif