From 34cd75f77d0ca94dbdba4e6cc9451fe4d33e78b3 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期三, 19 五月 2021 19:14:13 +0800
Subject: [PATCH] add api BHQueryProcs.
---
src/topic_node.cpp | 199 ++++++++++++++++++++++++++++---------------------
1 files changed, 112 insertions(+), 87 deletions(-)
diff --git a/src/topic_node.cpp b/src/topic_node.cpp
index 51a0ab7..8bbb929 100644
--- a/src/topic_node.cpp
+++ b/src/topic_node.cpp
@@ -70,69 +70,53 @@
}
LOG_DEBUG() << "Node Init, id " << ssn_id_;
auto NodeInit = [&]() {
- auto SendInitCmd = [&]() {
- int64_t init_cmd = ssn_id_ << 4 | EncodeCmd(eCmdNodeInit);
- auto end_time = steady_clock::now() + 3s;
- bool r = false;
- do {
- r = ShmMsgQueue::TrySend(shm(), BHTopicCenterAddress(), init_cmd);
- } while (!r && steady_clock::now() < end_time);
- return r;
- };
- if (SendInitCmd()) {
- LOG_DEBUG() << "node send init ok";
- auto end_time = steady_clock::now() + 3s;
- do {
- try {
- //TODO recv offset, avoid query.
- for (int i = eSockStart; i < eSockEnd; ++i) {
- sockets_.emplace_back(new ShmSocket(shm_, false, ssn_id_ + i, kMqLen));
- }
- break;
- } catch (...) {
- sockets_.clear();
- std::this_thread::sleep_for(100ms);
- }
- } while (steady_clock::now() < end_time);
+ int64_t init_request = ssn_id_ << 4 | EncodeCmd(eCmdNodeInit);
+ int64_t reply = 0;
+ if (BHNodeInit(init_request, reply) && DecodeCmd(reply) == eCmdNodeInitReply) {
+ int64_t abs_addr = reply >> 4;
+ sockets_.emplace_back(new ShmSocket(abs_addr, shm_, ssn_id_));
+ LOG_DEBUG() << "node init ok";
+ } else {
+ LOG_ERROR() << "Node Init Error";
}
};
if (sockets_.empty()) {
NodeInit();
}
if (!sockets_.empty()) {
- auto onNodeCmd = [this](ShmSocket &socket, int64_t &val) {
- LOG_DEBUG() << "node recv cmd: " << DecodeCmd(val);
- switch (DecodeCmd(val)) {
- case eCmdNodeInitReply: {
- MsgI msg(val >> 4);
- DEFER1(msg.Release());
+ auto onMsg = [this](ShmSocket &socket, MsgI &imsg, BHMsgHead &head) {
+ LOG_DEBUG() << "node recv type: " << head.type();
+ switch (head.type()) {
+ case kMsgTypeProcInit: {
+ // reuse msg to send proc init.
MsgProcInit body;
+ body.set_extra_mq_num(eSockEnd - eSockStart - 1);
auto head = InitMsgHead(GetType(body), info_.proc_id(), ssn_id_);
AddRoute(head, socket);
- if (msg.Fill(head, body)) {
- socket.Send(BHTopicCenterAddress(), msg);
+ if (imsg.Fill(head, body)) {
+ socket.Send(BHTopicCenterAddress(), imsg);
}
} break;
- default:
- break;
- }
- return true;
- };
-
- // recv msgs to avoid memory leak.
- auto onMsg = [this](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) {
- LOG_DEBUG() << "node recv type: " << head.type();
- if (head.type() == kMsgTypeProcInitReply) {
+ case kMsgTypeProcInitReply: {
LOG_DEBUG() << "got proc init reply";
MsgProcInitReply reply;
- if (imsg.ParseBody(reply)) {
+ if (imsg.ParseBody(reply) && IsSuccess(reply.errmsg().errcode())) {
+ for (auto &addr : reply.extra_mqs()) {
+ LOG_DEBUG() << "add socket " << addr.abs_addr() << ", id:" << addr.mq_id();
+ sockets_.emplace_back(new ShmSocket(addr.abs_addr(), shm(), addr.mq_id()));
+ }
SetProcIndex(reply.proc_index());
this->state_ = eStateUnregistered;
+
+ ServerStart(ServerAsyncCB(), 1);
+ SubscribeStartWorker(SubDataCB(), 1);
}
+ } break;
+ default: break;
}
return true;
};
- SockNode().Start(1, onMsg, onNodeCmd);
+ SockNode().Start(1, onMsg);
return true;
}
return false;
@@ -174,10 +158,10 @@
}
auto end_time = steady_clock::now() + milliseconds(timeout_ms);
- while (state_ != eStateUnregistered && steady_clock::now() < end_time) {
+ while (!Valid() && steady_clock::now() < end_time) {
std::this_thread::yield();
}
- if (state_ != eStateUnregistered) {
+ if (!Valid()) {
SetLastError(eError, kErrMsgNotInit);
return false;
}
@@ -308,6 +292,25 @@
reply.ParseBody(reply_body));
}
+bool TopicNode::QueryProcs(BHAddress &dest, MsgQueryProc &query, MsgQueryProcReply &reply_body, const int timeout_ms)
+{
+ if (!IsOnline()) {
+ SetLastError(eNotRegistered, kErrMsgNotRegistered);
+ return false;
+ }
+ auto &sock = SockNode();
+
+ BHMsgHead head(InitMsgHead(GetType(query), proc_id(), ssn()));
+ AddRoute(head, sock);
+
+ MsgI reply;
+ DEFER1(reply.Release());
+ BHMsgHead reply_head;
+ return (sock.SendAndRecv(BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms) &&
+ reply_head.type() == kMsgTypeQueryProcReply &&
+ reply.ParseBody(reply_body));
+}
+
bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
{
if (!IsOnline()) {
@@ -360,26 +363,32 @@
bool TopicNode::ServerStart(const ServerAsyncCB &acb, int nworker)
{
- auto onRecv = [this, acb](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) {
- if (head.type() != kMsgTypeRequestTopic || head.route_size() == 0) { return; }
- MsgRequestTopic req;
- if (!imsg.ParseBody(req)) { return; }
+ if (acb) {
+ auto onRecv = [this, acb](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) {
+ if (head.type() != kMsgTypeRequestTopic || head.route_size() == 0) { return; }
+ MsgRequestTopic req;
+ if (!imsg.ParseBody(req)) { return; }
- try {
- SrcInfo *p = new SrcInfo;
- if (!p) {
- throw std::runtime_error("no memory.");
+ try {
+ SrcInfo *p = new SrcInfo;
+ if (!p) {
+ throw std::runtime_error("no memory.");
+ }
+ p->route.assign(head.route().begin(), head.route().end());
+ p->msg_id = head.msg_id();
+ acb(p, *head.mutable_proc_id(), req);
+ } catch (std::exception &e) {
+ LOG_ERROR() << "error server handle msg:" << e.what();
}
- p->route.assign(head.route().begin(), head.route().end());
- p->msg_id = head.msg_id();
- acb(p, *head.mutable_proc_id(), req);
- } catch (std::exception &e) {
- LOG_ERROR() << "error server handle msg:" << e.what();
- }
- };
+ };
- auto &sock = SockServer();
- return acb && sock.Start(onRecv, nworker);
+ return SockServer().Start(onRecv, nworker);
+ } else {
+ auto onRequest = [this](ShmSocket &socket, MsgI &msg, BHMsgHead &head) {
+ server_buffer_->Write(std::move(head), msg.body());
+ };
+ return SockServer().Start(onRequest, nworker);
+ }
}
bool TopicNode::ServerRecvRequest(void *&src_info, std::string &proc_id, MsgRequestTopic &request, const int timeout_ms)
@@ -388,13 +397,19 @@
SetLastError(eNotRegistered, kErrMsgNotRegistered);
return false;
}
-
- auto &sock = SockServer();
-
- MsgI imsg;
BHMsgHead head;
- if (sock.SyncRecv(imsg, head, timeout_ms) && head.type() == kMsgTypeRequestTopic) {
- if (imsg.ParseBody(request)) {
+ std::string body;
+ auto end_time = steady_clock::now() + milliseconds(timeout_ms);
+ while (!server_buffer_->Read(head, body)) {
+ if (steady_clock::now() < end_time) {
+ robust::QuickSleep();
+ } else {
+ return false;
+ }
+ }
+
+ if (head.type() == kMsgTypeRequestTopic) {
+ if (request.ParseFromString(body)) {
head.mutable_proc_id()->swap(proc_id);
try {
SrcInfo *p = new SrcInfo;
@@ -633,20 +648,24 @@
bool TopicNode::SubscribeStartWorker(const SubDataCB &tdcb, int nworker)
{
- auto &sock = SockSub();
-
- auto AsyncRecvProc = [this, tdcb](ShmSocket &, MsgI &imsg, BHMsgHead &head) {
- if (head.type() == kMsgTypePublish) {
- MsgPublish pub;
- if (imsg.ParseBody(pub)) {
- tdcb(head.proc_id(), pub);
+ if (tdcb) {
+ auto AsyncRecvProc = [this, tdcb](ShmSocket &, MsgI &imsg, BHMsgHead &head) {
+ if (head.type() == kMsgTypePublish) {
+ MsgPublish pub;
+ if (imsg.ParseBody(pub)) {
+ tdcb(head.proc_id(), pub);
+ }
+ } else {
+ // ignored, or dropped
}
- } else {
- // ignored, or dropped
- }
- };
-
- return tdcb && sock.Start(AsyncRecvProc, nworker);
+ };
+ return SockSub().Start(AsyncRecvProc, nworker);
+ } else {
+ auto onSub = [this](ShmSocket &socket, MsgI &msg, BHMsgHead &head) {
+ sub_buffer_->Write(std::move(head), msg.body());
+ };
+ return SockSub().Start(onSub, nworker);
+ }
}
bool TopicNode::RecvSub(std::string &proc_id, MsgPublish &pub, const int timeout_ms)
@@ -656,13 +675,19 @@
return false;
}
- auto &sock = SockSub();
- MsgI msg;
- DEFER1(msg.Release(););
BHMsgHead head;
+ std::string body;
+ auto end_time = steady_clock::now() + milliseconds(timeout_ms);
+ while (!sub_buffer_->Read(head, body)) {
+ if (steady_clock::now() < end_time) {
+ robust::QuickSleep();
+ } else {
+ return false;
+ }
+ }
//TODO error msg.
- if (sock.SyncRecv(msg, head, timeout_ms) && head.type() == kMsgTypePublish) {
- if (msg.ParseBody(pub)) {
+ if (head.type() == kMsgTypePublish) {
+ if (pub.ParseFromString(body)) {
head.mutable_proc_id()->swap(proc_id);
return true;
}
--
Gitblit v1.8.0