From 1fbfef2a51db4a3bac9d8a5b87af94a40a913b7a Mon Sep 17 00:00:00 2001 From: lichao <lichao@aiotlink.com> Date: 星期日, 25 四月 2021 15:33:40 +0800 Subject: [PATCH] change mqid from uuid to uint64. --- src/topic_node.cpp | 44 ++++++++++++++++++++++---------------------- 1 files changed, 22 insertions(+), 22 deletions(-) diff --git a/src/topic_node.cpp b/src/topic_node.cpp index 00db773..9398318 100644 --- a/src/topic_node.cpp +++ b/src/topic_node.cpp @@ -25,7 +25,7 @@ namespace { -inline void AddRoute(BHMsgHead &head, const MQId &id) { head.add_route()->set_mq_id(&id, sizeof(id)); } +inline void AddRoute(BHMsgHead &head, const MQId id) { head.add_route()->set_mq_id(id); } struct SrcInfo { std::vector<BHAddress> route; @@ -82,7 +82,7 @@ auto &sock = SockNode(); MsgRegister body; body.mutable_proc()->Swap(&proc); - auto AddId = [&](const MQId &id) { body.add_addrs()->set_mq_id(&id, sizeof(id)); }; + auto AddId = [&](const MQId id) { body.add_addrs()->set_mq_id(id); }; AddId(SockNode().id()); AddId(SockServer().id()); AddId(SockClient().id()); @@ -108,12 +108,12 @@ MsgCommonReply body; CheckResult(imsg, head, body); }; - return sock.Send(&BHTopicCenterAddress(), head, body, onResult); + return sock.Send(BHTopicCenterAddress(), head, body, onResult); } else { MsgI reply; DEFER1(reply.Release();); BHMsgHead reply_head; - bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); + bool r = sock.SendAndRecv(BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); if (r) { CheckResult(reply, reply_head, reply_body); } @@ -144,12 +144,12 @@ MsgCommonReply body; CheckResult(imsg, head, body); }; - return sock.Send(&BHTopicCenterAddress(), head, body, onResult); + return sock.Send(BHTopicCenterAddress(), head, body, onResult); } else { MsgI reply; DEFER1(reply.Release();); BHMsgHead reply_head; - bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); + bool r = sock.SendAndRecv(BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); return r && CheckResult(reply, reply_head, reply_body); } } @@ -169,12 +169,12 @@ AddRoute(head, sock.id()); if (timeout_ms == 0) { - return sock.Send(&BHTopicCenterAddress(), head, body); + return sock.Send(BHTopicCenterAddress(), head, body); } else { MsgI reply; DEFER1(reply.Release();); BHMsgHead reply_head; - bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); + bool r = sock.SendAndRecv(BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); r = r && reply_head.type() == kMsgTypeCommonReply && reply.ParseBody(reply_body); return (r && IsSuccess(reply_body.errmsg().errcode())); } @@ -201,7 +201,7 @@ MsgI reply; DEFER1(reply.Release()); BHMsgHead reply_head; - return (sock.SendAndRecv(&BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms) && + return (sock.SendAndRecv(BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms) && reply_head.type() == kMsgTypeQueryTopicReply && reply.ParseBody(reply_body)); } @@ -221,12 +221,12 @@ AddRoute(head, sock.id()); if (timeout_ms == 0) { - return sock.Send(&BHTopicCenterAddress(), head, body); + return sock.Send(BHTopicCenterAddress(), head, body); } else { MsgI reply; DEFER1(reply.Release();); BHMsgHead reply_head; - bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); + bool r = sock.SendAndRecv(BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); r = r && reply_head.type() == kMsgTypeCommonReply; r = r && reply.ParseBody(reply_body); return r; @@ -247,8 +247,8 @@ for (int i = 0; i < head.route_size() - 1; ++i) { reply_head.add_route()->Swap(head.mutable_route(i)); } - auto &remote = head.route().rbegin()->mq_id(); - sock.Send(remote.data(), reply_head, reply_body); + auto remote = head.route().rbegin()->mq_id(); + sock.Send(remote, reply_head, reply_body); } }; @@ -315,7 +315,7 @@ for (unsigned i = 0; i < p->route.size() - 1; ++i) { head.add_route()->Swap(&p->route[i]); } - return sock.Send(p->route.back().mq_id().data(), head, body); + return sock.Send(p->route.back().mq_id(), head, body); } bool TopicNode::ClientStartWorker(RequestResultCB const &cb, const int nworker) @@ -361,9 +361,9 @@ } } }; - return sock.Send(addr.mq_id().data(), head, req, onRecv); + return sock.Send(addr.mq_id(), head, req, onRecv); } else { - return sock.Send(addr.mq_id().data(), head, req); + return sock.Send(addr.mq_id(), head, req); } }; @@ -396,7 +396,7 @@ DEFER1(reply_msg.Release();); BHMsgHead reply_head; - if (sock.SendAndRecv(addr.mq_id().data(), head, request, reply_msg, reply_head, timeout_ms) && + if (sock.SendAndRecv(addr.mq_id(), head, request, reply_msg, reply_head, timeout_ms) && reply_head.type() == kMsgTypeRequestTopicReply && reply_msg.ParseBody(out_reply)) { reply_head.mutable_proc_id()->swap(out_proc_id); @@ -441,7 +441,7 @@ std::vector<NodeAddress> lst; if (QueryRPCTopics(topic, lst, timeout_ms)) { addr = lst.front().addr(); - if (!addr.mq_id().empty()) { + if (addr.mq_id() != 0) { topic_query_cache_.Store(topic, addr); return true; } @@ -464,13 +464,13 @@ AddRoute(head, sock.id()); if (timeout_ms == 0) { - return sock.Send(&BHTopicBusAddress(), head, pub); + return sock.Send(BHTopicBusAddress(), head, pub); } else { MsgI reply; DEFER1(reply.Release();); BHMsgHead reply_head; MsgCommonReply reply_body; - return sock.SendAndRecv(&BHTopicBusAddress(), head, pub, reply, reply_head, timeout_ms) && + return sock.SendAndRecv(BHTopicBusAddress(), head, pub, reply, reply_head, timeout_ms) && reply_head.type() == kMsgTypeCommonReply && reply.ParseBody(reply_body) && IsSuccess(reply_body.errmsg().errcode()); @@ -497,12 +497,12 @@ BHMsgHead head(InitMsgHead(GetType(sub), proc_id())); AddRoute(head, sock.id()); if (timeout_ms == 0) { - return sock.Send(&BHTopicBusAddress(), head, sub); + return sock.Send(BHTopicBusAddress(), head, sub); } else { MsgI reply; DEFER1(reply.Release();); BHMsgHead reply_head; - return sock.SendAndRecv(&BHTopicBusAddress(), head, sub, reply, reply_head, timeout_ms) && + return sock.SendAndRecv(BHTopicBusAddress(), head, sub, reply, reply_head, timeout_ms) && reply_head.type() == kMsgTypeCommonReply && reply.ParseBody(reply_body) && IsSuccess(reply_body.errmsg().errcode()); -- Gitblit v1.8.0