fix response

include/rest_rpc/asio_util.hpp

@@ -1,11 +1,23 @@
 #pragma once
+#include "traits.h"
 #include "use_asio.hpp"
 
 namespace rest_rpc {
+template <typename T>
+constexpr inline bool is_awaitable_v =
+    util::is_specialization_v<std::remove_cvref_t<T>, asio::awaitable>;
+
+template <typename T, bool IsAwaitable = is_awaitable_v<T>> struct return_type {
+  using type = T;
+};
+
+template <typename T> struct return_type<T, true> {
+  using type = typename std::remove_cvref_t<T>::value_type;
+};
+
+template <typename T> using return_type_t = typename return_type<T>::type;
+
 template <typename Coro> inline auto async_start(auto executor, Coro &&coro) {
-  using R =
-      typename std::remove_cvref_t<std::invoke_result_t<Coro>>::value_type;
-  static_assert(std::is_void_v<R>);
   asio::co_spawn(executor, std::forward<Coro>(coro), asio::detached);
 }
 

include/rest_rpc/error_code.h

@@ -20,6 +20,7 @@ enum class rpc_errc : std::int8_t {
   protocol_error,
   has_response,
   duplicate_topic,
+  rpc_context_init_failed,
 };
 
 class rpc_error_category : public std::error_category {
@@ -60,6 +61,9 @@ public:
       return "has response, duplicate response is not allowed";
     case rpc_errc::duplicate_topic:
       return "duplicate topic";
+    case rpc_errc::rpc_context_init_failed:
+      return "the rpc context init failed, it must be created in rpc handler "
+             "io thread, otherwise will init failed";
     default:
       return "unknown error";
     }

include/rest_rpc/rpc_client.hpp

@@ -94,13 +94,15 @@ public:
   }
 
   template <auto func, typename... Args>
-  asio::awaitable<call_result<function_return_type_t<decltype(func)>>>
+  asio::awaitable<
+      call_result<return_type_t<function_return_type_t<decltype(func)>>>>
   call(Args &&...args) {
     return call_for<func>(std::chrono::seconds(5), std::forward<Args>(args)...);
   }
 
   template <auto func, typename... Args>
-  asio::awaitable<call_result<function_return_type_t<decltype(func)>>>
+  asio::awaitable<
+      call_result<return_type_t<function_return_type_t<decltype(func)>>>>
   call_for(auto duration, Args &&...args) {
     using args_tuple = function_parameters_t<decltype(func)>;
     static_assert(std::is_constructible_v<args_tuple, Args...>,
@@ -108,7 +110,7 @@ public:
 
     rest_rpc_header header{};
     header.function_id = get_key<func>();
-    using R = function_return_type_t<decltype(func)>;
+    using R = return_type_t<function_return_type_t<decltype(func)>>;
     auto r = co_await (watchdog(duration) ||
                        call_impl<R>(header, std::forward<Args>(args)...));
     if (r.index() == 0) {

include/rest_rpc/rpc_connection.hpp

@@ -8,39 +8,42 @@
 namespace rest_rpc {
 class rpc_connection;
 
-class rpc_context {
+class tls_data {
 public:
-  static auto &context() {
-    thread_local rpc_context instance;
-    return instance;
-  }
-
-  rpc_context(rpc_context &&o) : conn_(std::move(o.conn_)), delay_(o.delay_) {}
-  rpc_context(const rpc_context &o) = delete;
-
   void set_connection(std::shared_ptr<rpc_connection> conn) { conn_ = conn; }
 
   bool delay() { return delay_; }
 
-  void set_delay(bool r) { context().delay_ = r; }
+  void set_delay(bool r) { delay_ = r; }
+
+  auto get_executor();
+  std::shared_ptr<rpc_connection> get_conn() { return conn_; }
+
+private:
+  std::shared_ptr<rpc_connection> conn_ = nullptr;
+  bool delay_ = false;
+};
+
+inline auto &get_context() {
+  static thread_local tls_data instance;
+  return instance;
+}
+
+class rpc_context {
+public:
+  rpc_context();
 
   auto get_executor();
 
   template <auto func, typename... Args>
   asio::awaitable<std::error_code> response_s(Args &&...args);
 
-  template <auto func, typename... Args>
-  std::error_code sync_response_s(Args &&...args);
-
   template <typename... Args>
   asio::awaitable<std::error_code> response(Args &&...args);
 
-  template <typename... Args> std::error_code sync_response(Args &&...args);
-
 private:
-  rpc_context() = default;
-  std::shared_ptr<rpc_connection> conn_;
-  bool delay_ = false;
+  asio::any_io_executor executor_;
+  std::shared_ptr<rpc_connection> conn_ = nullptr;
   bool has_response_ = false;
 };
 
@@ -99,11 +102,11 @@ public:
       }
 
       // route
-      rpc_context::context().set_connection(self);
+      get_context().set_connection(self);
       auto result = co_await router_.route(header.function_id, body_);
-      bool delay = rpc_context::context().delay();
+      bool delay = get_context().delay();
       if (delay) {
-        rpc_context::context().set_delay(false);
+        get_context().set_delay(false);
         continue;
       }
 
@@ -198,6 +201,21 @@ private:
   std::atomic<uint32_t> topic_id_;
 };
 
+auto tls_data::get_executor() {
+  if (!conn_) {
+    return asio::any_io_executor();
+  }
+  return conn_->get_executor();
+}
+
+auto rpc_context::get_executor() { return executor_; }
+
+rpc_context::rpc_context() {
+  executor_ = get_context().get_executor();
+  conn_ = get_context().get_conn();
+  get_context().set_delay(true);
+}
+
 // zero or one arguments
 template <auto func, typename... Args>
 asio::awaitable<std::error_code> rpc_context::response_s(Args &&...args) {
@@ -210,29 +228,19 @@ asio::awaitable<std::error_code> rpc_context::response_s(Args &&...args) {
   return response(std::forward<Args>(args)...);
 }
 
-template <auto func, typename... Args>
-std::error_code rpc_context::sync_response_s(Args &&...args) {
-  return sync_wait(conn_->get_executor(),
-                   response_s<func>(std::forward<Args>(args)...));
-}
-
 template <typename... Args>
 asio::awaitable<std::error_code> rpc_context::response(Args &&...args) {
   if (has_response_) {
     co_return make_error_code(rpc_errc::has_response);
   }
-
-  has_response_ = true;
+  if (!conn_) {
+    REST_LOG_ERROR << "rpc context init failed";
+    co_return make_error_code(rpc_errc::rpc_context_init_failed);
+  }
   rpc_service::msgpack_codec codec;
   rpc_result result(codec.pack_args(std::forward<Args>(args)...));
+  has_response_ = true;
   co_return co_await conn_->response(result);
 }
 
-template <typename... Args>
-std::error_code rpc_context::sync_response(Args &&...args) {
-  return sync_wait(conn_->get_executor(),
-                   response(std::forward<Args>(args)...));
-}
-
-auto rpc_context::get_executor() { return conn_->get_executor(); }
 } // namespace rest_rpc

include/rest_rpc/rpc_router.hpp

@@ -2,6 +2,7 @@
 #include "codec.h"
 #include "error_code.h"
 
+#include "asio_util.hpp"
 #include "util.hpp"
 #include <cstdint>
 #include <functional>
@@ -9,10 +10,6 @@
 #include <string_view>
 
 namespace rest_rpc {
-template <typename T>
-constexpr inline bool is_awaitable_v =
-    util::is_specialization_v<std::remove_cvref_t<T>, asio::awaitable>;
-
 template <typename T>
 constexpr inline bool is_void_v =
     std::is_same_v<T, void> || std::is_same_v<T, asio::awaitable<void>>;

tests/test_rest_rpc.cpp

@@ -69,9 +69,8 @@ template <auto func> asio::awaitable<void> response(auto ctx) {
 }
 
 std::string_view delay_response(std::string_view str) {
-  auto &ctx = rpc_context::context();
+  rpc_context ctx;
   // set_delay before response in another thread
-  ctx.set_delay(true);
 
   // std::thread thd([ctx = std::move(ctx)]() mutable {
   //   std::this_thread::sleep_for(std::chrono::seconds(2));
@@ -115,6 +114,66 @@ asio::awaitable<std::string> no_arg_coro1() {
   co_return "test";
 }
 
+std::string delay_response1(std::string_view str) {
+  rpc_context ctx; // right, created in io thread.
+  async_start(ctx.get_executor(), ctx.response(std::string(str)));
+
+  return "";
+}
+
+asio::awaitable<std::string> delay_response2(std::string_view str) {
+  auto coro = []() -> asio::awaitable<void> {
+    rpc_context ctx; // rpc_context init will be failed, it should be defined in
+                     // io thread.
+    auto ret = co_await ctx.response("test");
+    REST_LOG_INFO << ret.message();
+    CHECK(ret);
+  };
+  sync_wait(get_global_executor(), coro());
+
+  rpc_context ctx;
+  co_await ctx.response(str);
+  auto ret = co_await ctx.response(str);
+  CHECK(ret);
+
+  co_return "";
+}
+
+std::string delay_response3(std::string str) {
+  rpc_context ctx;
+  std::thread thd([ctx = std::move(ctx), str]() mutable {
+    sync_wait(ctx.get_executor(), ctx.response(std::move(str)));
+  });
+  thd.detach();
+
+  return "";
+}
+
+TEST_CASE("test delay response") {
+  using T = return_type_t<int>;
+  rpc_server server("127.0.0.1:9005");
+  server.register_handler<delay_response1>();
+  server.register_handler<delay_response2>();
+  server.register_handler<delay_response3>();
+  server.async_start();
+  rpc_client client;
+  sync_wait(client.get_executor(), client.connect("127.0.0.1:9005"));
+  auto result =
+      sync_wait(client.get_executor(), client.call<delay_response1>("test"));
+  CHECK(result.value == "test");
+  auto result1 =
+      sync_wait(client.get_executor(), client.call<delay_response2>("test"));
+  CHECK(result1.value == "test");
+  result1 =
+      sync_wait(client.get_executor(), client.call<delay_response2>("test"));
+  CHECK(result1.value == "test");
+  auto result2 = sync_wait(
+      client.get_executor(),
+      client.call_for<delay_response3>(std::chrono::minutes(2), "test"));
+  CHECK(result2.value == "test");
+  server.stop();
+}
+
 // TODO: client pool
 asio::awaitable<void> test_router() {
   rpc_router router;