From 1f3729698a131b3f701f67adb6a1258aa1235dce Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期二, 20 四月 2021 15:43:53 +0800
Subject: [PATCH] api server callback change tag to src; refactor.
---
src/topic_node.h | 38 +++++++++++++++++---------------------
1 files changed, 17 insertions(+), 21 deletions(-)
diff --git a/src/topic_node.h b/src/topic_node.h
index 87ad770..35cdde5 100644
--- a/src/topic_node.h
+++ b/src/topic_node.h
@@ -34,7 +34,6 @@
SharedMemory &shm() { return shm_; }
public:
- typedef std::function<void(std::string &proc_id, const void *data, const int len)> DataCB;
TopicNode(SharedMemory &shm);
~TopicNode();
@@ -44,8 +43,10 @@
bool Heartbeat(const int timeout_ms);
// topic rpc server
- typedef std::function<bool(const std::string &client_proc_id, const MsgRequestTopic &request, MsgRequestTopicReply &reply)> ServerCB;
- bool ServerStart(ServerCB const &cb, const int nworker = 2);
+ typedef std::function<bool(const std::string &client_proc_id, const MsgRequestTopic &request, MsgRequestTopicReply &reply)> ServerSyncCB;
+ typedef std::function<void(void *src_info, std::string &client_proc_id, MsgRequestTopic &request)> ServerAsyncCB;
+ bool ServerStart(ServerSyncCB const &cb, const int nworker = 2);
+ bool ServerStart(ServerAsyncCB const &cb, const int nworker = 2);
bool ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply, const int timeout_ms);
bool ServerRecvRequest(void *&src_info, std::string &proc_id, MsgRequestTopic &request, const int timeout_ms);
bool ServerSendReply(void *src_info, const MsgRequestTopicReply &reply);
@@ -65,7 +66,7 @@
bool Subscribe(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms);
bool RecvSub(std::string &proc_id, MsgPublish &pub, const int timeout_ms);
- void Start(ServerCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker = 2);
+ void Start(ServerAsyncCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker = 2);
void Stop();
private:
@@ -77,45 +78,40 @@
{
class Impl
{
- typedef std::unordered_map<Topic, Address> Store;
- Store store_;
+ typedef std::unordered_map<Topic, Address> Records;
+ Records records_;
public:
bool Find(const Topic &topic, Address &addr)
{
- auto pos = store_.find(topic);
- if (pos != store_.end()) {
+ auto pos = records_.find(topic);
+ if (pos != records_.end()) {
addr = pos->second;
return true;
} else {
return false;
}
}
- bool Update(const Topic &topic, const Address &addr)
+ bool Store(const Topic &topic, const Address &addr)
{
- store_[topic] = addr;
+ records_[topic] = addr;
return true;
}
};
Synced<Impl> impl_;
- // Impl &impl()
- // {
- // thread_local Impl impl;
- // return impl;
- // }
public:
bool Find(const Topic &topic, Address &addr) { return impl_->Find(topic, addr); }
- bool Update(const Topic &topic, const Address &addr) { return impl_->Update(topic, addr); }
+ bool Store(const Topic &topic, const Address &addr) { return impl_->Store(topic, addr); }
};
// some sockets may be the same one, using functions make it easy to change.
- auto &SockNode() { return sock_node_; }
- auto &SockPub() { return SockNode(); }
- auto &SockSub() { return sock_sub_; }
- auto &SockClient() { return sock_client_; }
- auto &SockServer() { return sock_server_; }
+ ShmSocket &SockNode() { return sock_node_; }
+ ShmSocket &SockPub() { return SockNode(); }
+ ShmSocket &SockSub() { return sock_sub_; }
+ ShmSocket &SockClient() { return sock_client_; }
+ ShmSocket &SockServer() { return sock_server_; }
bool IsRegistered() const { return registered_.load(); }
ShmSocket sock_node_;
--
Gitblit v1.8.0