diff --git a/include/rest_rpc/error_code.h b/include/rest_rpc/error_code.h
index 997375e..51e83cd 100644
--- a/include/rest_rpc/error_code.h
+++ b/include/rest_rpc/error_code.h
@@ -19,6 +19,7 @@ enum class rpc_errc : std::int8_t {
   request_timeout,
   protocol_error,
   has_response,
+  duplicate_topic,
 };
 
 class rpc_error_category : public std::error_category {
@@ -57,6 +58,8 @@ public:
       return "protocol error";
     case rpc_errc::has_response:
       return "has response, duplicate response is not allowed";
+    case rpc_errc::duplicate_topic:
+      return "duplicate topic";
     default:
       return "unknown error";
     }
diff --git a/include/rest_rpc/rest_rpc_protocol.hpp b/include/rest_rpc/rest_rpc_protocol.hpp
index 5ce75e6..7966834 100644
--- a/include/rest_rpc/rest_rpc_protocol.hpp
+++ b/include/rest_rpc/rest_rpc_protocol.hpp
@@ -6,7 +6,7 @@
 #include <arpa/inet.h>
 #endif
 
-#if  defined(__APPLE__) || defined(_WIN32)
+#if defined(__APPLE__) || defined(_WIN32)
 #else
 inline uint64_t htonll(uint64_t value) {
   return ((uint64_t)htonl(value & 0xFFFFFFFF) << 32) | htonl(value >> 32);
@@ -20,7 +20,7 @@ inline uint64_t ntohll(uint64_t value) {
 namespace rest_rpc {
 inline constexpr uint8_t REST_MAGIC_NUM = 39;
 struct rest_rpc_header {
-  uint8_t magic;
+  uint8_t magic = REST_MAGIC_NUM;
   uint8_t version;
   uint8_t serialize_type;
   uint8_t msg_type;
diff --git a/include/rest_rpc/rpc_client.hpp b/include/rest_rpc/rpc_client.hpp
index f5c18b9..2593838 100644
--- a/include/rest_rpc/rpc_client.hpp
+++ b/include/rest_rpc/rpc_client.hpp
@@ -104,26 +104,56 @@ public:
     static_assert(std::is_constructible_v<args_tuple, Args...>,
                   "called rpc function and arguments are not match");
 
+    rest_rpc_header header{};
+    header.function_id = get_key<func>();
     using R = function_return_type_t<decltype(func)>;
     auto r = co_await (watchdog(duration) ||
-                       call_impl<func>(std::forward<Args>(args)...));
+                       call_impl<R>(header, std::forward<Args>(args)...));
     if (r.index() == 0) {
       co_return call_result<R>{rpc_errc::request_timeout};
     }
     co_return std::get<1>(r);
   }
 
+  template <typename R = void>
+  asio::awaitable<call_result<R>> subscribe(std::string_view topic) {
+    uint32_t topic_id = MD5::MD5Hash32(topic.data(), topic.size()); // topic id
+    bool b = false;
+    call_result<R> ret{};
+    auto it = sub_ops_.find(topic_id);
+    if (it == sub_ops_.end()) {
+      rest_rpc_header header{};
+      header.msg_type = 1; // pub/sub
+
+      header.function_id = topic_id;
+      auto [it, r] = sub_ops_.emplace(topic_id, sub_operation{});
+      if (!r) {
+        REST_LOG_ERROR << "subscribe duplicate topic";
+        co_return call_result<R>{rpc_errc::duplicate_topic};
+      }
+
+      std::tie(b, ret) = co_await (
+          asio::async_compose<decltype(asio::use_awaitable), void(bool)>(
+              std::ref(it->second), asio::use_awaitable) &&
+          call_impl<R>(header));
+    } else {
+      std::tie(b, ret) = co_await (
+          asio::async_compose<decltype(asio::use_awaitable), void(bool)>(
+              std::ref(it->second), asio::use_awaitable) &&
+          wait_response<R>());
+    }
+
+    co_return std::move(ret);
+  }
+
   void enable_tcp_no_delay(bool r) { tcp_no_delay_ = r; }
 
   void enable_cross_ending(bool r) { cross_ending_ = r; }
 
 private:
-  template <auto func, typename... Args>
-  asio::awaitable<call_result<function_return_type_t<decltype(func)>>>
-  call_impl(Args &&...args) {
-    rest_rpc_header header{};
-    header.magic = 39;
-    header.function_id = get_key<func>();
+  template <typename R, typename... Args>
+  asio::awaitable<call_result<R>> call_impl(rest_rpc_header &header,
+                                            Args &&...args) {
     rpc_service::msgpack_codec codec;
     auto buf = codec.pack_args(std::forward<Args>(args)...);
     header.body_len = buf.size();
@@ -138,7 +168,6 @@ private:
       buffers.push_back(asio::buffer(buf.data(), buf.size()));
     }
 
-    using R = function_return_type_t<decltype(func)>;
     call_result<R> result{};
     std::error_code ec;
     size_t size;
@@ -149,6 +178,13 @@ private:
       co_return result;
     }
 
+    co_return co_await wait_response<R>();
+  }
+
+  template <typename R> asio::awaitable<call_result<R>> wait_response() {
+    call_result<R> result{};
+    std::error_code ec;
+    size_t size;
     rest_rpc_header resp_header;
     std::tie(ec, size) = co_await asio::async_read(
         socket_, asio::buffer(&resp_header, sizeof(rest_rpc_header)),
@@ -176,16 +212,19 @@ private:
       co_return result;
     }
     result.ec = (rpc_errc)body_[0];
-    if (resp_header.body_len > 0) {
-      auto view = std::string_view(body_.data() + 1, resp_header.body_len - 1);
-      REST_LOG_INFO << view;
-    }
     if constexpr (!std::is_void_v<R>) {
+      rpc_service::msgpack_codec codec;
       result.value = codec.unpack<R>(
           std::string_view(body_.data() + 1, resp_header.body_len - 1));
     }
 
-    co_return result;
+    if (resp_header.msg_type == 1) { // pubsub
+      if (auto it = sub_ops_.find(resp_header.function_id);
+          it != sub_ops_.end()) {
+        it->second.complete(true);
+      }
+    }
+    co_return std::move(result);
   }
 
   asio::awaitable<std::error_code> watchdog(auto duration) {
@@ -195,9 +234,27 @@ private:
     co_return ec;
   }
 
+  class sub_operation {
+  public:
+    template <typename Self> void operator()(Self &&self) {
+      using SelfType = std::decay_t<Self>;
+      auto shared_self = std::make_shared<SelfType>(std::move(self));
+
+      complete_handler_ = [shared_self](bool r) mutable {
+        shared_self->complete(r);
+      };
+    }
+
+    void complete(bool r) { complete_handler_(r); }
+
+  private:
+    std::function<void(bool)> complete_handler_;
+  };
+
   tcp_socket socket_;
   std::string body_;
   bool tcp_no_delay_ = true;
   bool cross_ending_ = false;
+  std::unordered_map<uint32_t, sub_operation> sub_ops_;
 };
 } // namespace rest_rpc
\ No newline at end of file
diff --git a/include/rest_rpc/rpc_connection.hpp b/include/rest_rpc/rpc_connection.hpp
index c3b5ed7..0dff924 100644
--- a/include/rest_rpc/rpc_connection.hpp
+++ b/include/rest_rpc/rpc_connection.hpp
@@ -76,6 +76,11 @@ public:
         break;
       }
 
+      if (header.msg_type == 1) { // pub sub
+        topic_id_ = header.function_id;
+        continue;
+      }
+
       detail::resize(body_, header.body_len);
 
       if (header.body_len > 0) {
@@ -105,9 +110,14 @@ public:
     }
   }
 
-  asio::awaitable<std::error_code> response(const rpc_result &result) {
+  asio::awaitable<std::error_code> response(const rpc_result &result,
+                                            uint32_t func_id = 0) {
     rest_rpc_header resp_header{};
     resp_header.magic = 39;
+    if (func_id != 0) {
+      resp_header.msg_type = 1;
+      resp_header.function_id = func_id;
+    }
     resp_header.body_len = result.size() + 1;
     if (cross_ending_) {
       prepare_for_send(resp_header);
@@ -131,6 +141,7 @@ public:
 
   uint64_t id() const { return conn_id_; }
   auto get_executor() { return socket_.get_executor(); }
+  uint32_t topic_id() const { return topic_id_; }
 
   void
   set_quit_callback(std::function<void(const uint64_t &conn_id)> callback) {
@@ -178,6 +189,7 @@ private:
   bool checkout_timeout_ = false;
   rpc_router &router_;
   bool cross_ending_;
+  uint32_t topic_id_;
 };
 
 // zero or one arguments
diff --git a/include/rest_rpc/rpc_server.hpp b/include/rest_rpc/rpc_server.hpp
index a8a61c8..0462314 100644
--- a/include/rest_rpc/rpc_server.hpp
+++ b/include/rest_rpc/rpc_server.hpp
@@ -102,6 +102,27 @@ public:
     return conns_.size();
   }
 
+  auto get_connections() {
+    std::scoped_lock lock(*conn_mtx_);
+    return conns_;
+  }
+
+  template <typename T>
+  asio::awaitable<void> publish(std::string_view topic, T &&t) {
+    auto id = MD5::MD5Hash32(topic.data(), topic.size());
+    auto conns = get_connections();
+    for (auto &[_, conn] : conns) {
+      if (conn->topic_id() == id) {
+        co_await conn->response(
+            rpc_service::msgpack_codec::pack_args(std::forward<T>(t)), id);
+      }
+    }
+  }
+
+  template <typename T> void sync_publish(std::string_view topic, T &&t) {
+    sync_wait(get_global_executor(), publish(topic, std::forward<T>(t)));
+  }
+
 private:
   std::error_code listen() {
     using asio::ip::tcp;