| | |
| | | using namespace std::chrono; |
| | | using namespace std::chrono_literals; |
| | | |
| | | const char *const kErrMsgNotInit = "BHome node NOT initialized."; |
| | | const char *const kErrMsgNotRegistered = "BHome node NOT registered."; |
| | | |
| | | 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; |
| | | std::string msg_id; |
| | | }; |
| | | |
| | | const int kMqLen = 700; |
| | | |
| | | } // namespace |
| | | |
| | | TopicNode::TopicNode(SharedMemory &shm) : |
| | | shm_(shm), sock_node_(shm), sock_client_(shm), sock_server_(shm), sock_sub_(shm), registered_(false), registered_ever_(false) |
| | | shm_(shm), state_(eStateUnregistered) |
| | | { |
| | | // recv msgs to avoid memory leak. |
| | | auto default_ignore_msg = [](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { return true; }; |
| | | SockNode().Start(default_ignore_msg); |
| | | Init(); |
| | | } |
| | | |
| | | TopicNode::~TopicNode() |
| | | { |
| | | LOG_DEBUG() << "~TopicNode()"; |
| | | Stop(); |
| | | SockNode().Stop(); |
| | | if (!registered_ever_) { |
| | | SockNode().Remove(); |
| | | SockClient().Remove(); |
| | | SockServer().Remove(); |
| | | SockSub().Remove(); |
| | | } |
| | | |
| | | bool TopicNode::Init() |
| | | { |
| | | std::lock_guard<std::mutex> lk(mutex_); |
| | | |
| | | if (Valid()) { |
| | | return true; |
| | | } |
| | | |
| | | if (ssn_id_ == 0) { |
| | | ssn_id_ = ShmMsgQueue::NewId(); |
| | | } |
| | | LOG_DEBUG() << "Node Init, id " << ssn_id_; |
| | | MsgI msg; |
| | | msg.OffsetRef() = ssn_id_; |
| | | if (ShmMsgQueue::TrySend(shm(), BHInitAddress(), msg)) { |
| | | |
| | | auto end_time = steady_clock::now() + 3s; |
| | | do { |
| | | try { |
| | | 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); |
| | | if (!sockets_.empty()) { |
| | | // recv msgs to avoid memory leak. |
| | | auto default_ignore_msg = [](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { return true; }; |
| | | SockNode().Start(default_ignore_msg); |
| | | return true; |
| | | } |
| | | } |
| | | return false; |
| | | } |
| | | |
| | | void TopicNode::Start(ServerAsyncCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker) |
| | | { |
| | | if (!Init()) { |
| | | SetLastError(eError, kErrMsgNotInit); |
| | | return; |
| | | } |
| | | if (nworker < 1) { |
| | | nworker = 1; |
| | | } else if (nworker > 16) { |
| | | nworker = 16; |
| | | } |
| | | |
| | | SockNode().Start(); |
| | | ServerStart(server_cb, nworker); |
| | | SubscribeStartWorker(sub_cb, nworker); |
| | | ClientStartWorker(client_cb, nworker); |
| | | } |
| | | void TopicNode::Stop() |
| | | { |
| | | SockSub().Stop(); |
| | | SockServer().Stop(); |
| | | SockClient().Stop(); |
| | | LOG_DEBUG() << "Node Stopping"; |
| | | for (auto &p : sockets_) { p->Stop(); } |
| | | LOG_INFO() << "Node Stopped"; |
| | | } |
| | | |
| | | bool TopicNode::Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms) |
| | | { |
| | | if (!Init()) { |
| | | SetLastError(eError, kErrMsgNotInit); |
| | | return false; |
| | | } |
| | | |
| | | info_ = proc; |
| | | |
| | | 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()); |
| | | AddId(SockSub().id()); |
| | | AddId(SockPub().id()); |
| | | |
| | | auto head(InitMsgHead(GetType(body), body.proc().proc_id())); |
| | | auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn())); |
| | | AddRoute(head, sock.id()); |
| | | |
| | | auto CheckResult = [this](MsgI &msg, BHMsgHead &head, MsgCommonReply &rbody) { |
| | |
| | | msg.ParseBody(rbody) && |
| | | IsSuccess(rbody.errmsg().errcode()); |
| | | if (ok) { |
| | | registered_ever_.store(true); |
| | | state(eStateOnline); |
| | | } else { |
| | | state_cas(eStateOnline, eStateOffline); |
| | | } |
| | | registered_.store(ok); |
| | | }; |
| | | |
| | | if (timeout_ms == 0) { |
| | |
| | | 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(shm_);); |
| | | 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); |
| | | } |
| | | return IsRegistered(); |
| | | return IsOnline(); |
| | | } |
| | | } |
| | | bool TopicNode::Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms) |
| | | { |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | | info_.Clear(); |
| | | state_cas(eStateOnline, eStateOffline); |
| | | |
| | | auto &sock = SockNode(); |
| | | MsgUnregister body; |
| | | body.mutable_proc()->Swap(&proc); |
| | | |
| | | auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn())); |
| | | AddRoute(head, sock.id()); |
| | | |
| | | auto CheckResult = [this](MsgI &msg, BHMsgHead &head, MsgCommonReply &rbody) { |
| | | bool r = head.type() == kMsgTypeCommonReply && |
| | | msg.ParseBody(rbody) && |
| | | IsSuccess(rbody.errmsg().errcode()); |
| | | return r; |
| | | }; |
| | | |
| | | if (timeout_ms == 0) { |
| | | auto onResult = [this, CheckResult](ShmSocket &socket, MsgI &imsg, BHMsgHead &head) { |
| | | MsgCommonReply body; |
| | | CheckResult(imsg, head, body); |
| | | }; |
| | | 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); |
| | | return r && CheckResult(reply, reply_head, reply_body); |
| | | } |
| | | } |
| | | |
| | | bool TopicNode::Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | |
| | | MsgHeartbeat body; |
| | | body.mutable_proc()->Swap(&proc); |
| | | |
| | | auto head(InitMsgHead(GetType(body), body.proc().proc_id())); |
| | | auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn())); |
| | | 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(shm_);); |
| | | 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())); |
| | | } |
| | |
| | | return Heartbeat(proc, reply_body, timeout_ms); |
| | | } |
| | | |
| | | bool TopicNode::QueryTopicAddress(BHAddress &dest, MsgQueryTopic &query, MsgQueryTopicReply &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.id()); |
| | | |
| | | MsgI reply; |
| | | DEFER1(reply.Release()); |
| | | BHMsgHead reply_head; |
| | | return (sock.SendAndRecv(BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms) && |
| | | reply_head.type() == kMsgTypeQueryTopicReply && |
| | | reply.ParseBody(reply_body)); |
| | | } |
| | | |
| | | bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | |
| | | MsgRegisterRPC body; |
| | | body.mutable_topics()->Swap(&topics); |
| | | |
| | | auto head(InitMsgHead(GetType(body), proc_id())); |
| | | auto head(InitMsgHead(GetType(body), proc_id(), ssn())); |
| | | 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(shm_);); |
| | | 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; |
| | |
| | | |
| | | MsgRequestTopicReply reply_body; |
| | | if (rcb(head.proc_id(), req, reply_body)) { |
| | | BHMsgHead reply_head(InitMsgHead(GetType(reply_body), proc_id(), head.msg_id())); |
| | | BHMsgHead reply_head(InitMsgHead(GetType(reply_body), proc_id(), ssn(), head.msg_id())); |
| | | |
| | | 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); |
| | | } |
| | | }; |
| | | |
| | |
| | | |
| | | bool TopicNode::ServerRecvRequest(void *&src_info, std::string &proc_id, MsgRequestTopic &request, const int timeout_ms) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | |
| | | |
| | | bool TopicNode::ServerSendReply(void *src_info, const MsgRequestTopicReply &body) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | |
| | | if (!p || p->route.empty()) { |
| | | return false; |
| | | } |
| | | BHMsgHead head(InitMsgHead(GetType(body), proc_id(), p->msg_id)); |
| | | BHMsgHead head(InitMsgHead(GetType(body), proc_id(), ssn(), p->msg_id)); |
| | | 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) |
| | |
| | | return SockClient().Start(onData, nworker); |
| | | } |
| | | |
| | | bool TopicNode::ClientAsyncRequest(const MsgRequestTopic &req, std::string &out_msg_id, const RequestResultCB &cb) |
| | | bool TopicNode::ClientAsyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &req, std::string &out_msg_id, const RequestResultCB &cb) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | |
| | | |
| | | auto SendTo = [this, msg_id](const BHAddress &addr, const MsgRequestTopic &req, const RequestResultCB &cb) { |
| | | auto &sock = SockClient(); |
| | | BHMsgHead head(InitMsgHead(GetType(req), proc_id(), msg_id)); |
| | | BHMsgHead head(InitMsgHead(GetType(req), proc_id(), ssn(), msg_id)); |
| | | AddRoute(head, sock.id()); |
| | | head.set_topic(req.topic()); |
| | | |
| | |
| | | } |
| | | } |
| | | }; |
| | | 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); |
| | | } |
| | | }; |
| | | |
| | | try { |
| | | BHAddress addr; |
| | | #if 1 |
| | | return (ClientQueryRPCTopic(req.topic(), addr, 3000)) && SendTo(addr, req, cb); |
| | | #else |
| | | if (topic_query_cache_.Pick(req.topic(), addr)) { |
| | | return SendTo(addr, req, cb); |
| | | } |
| | | |
| | | auto &sock = SockClient(); |
| | | MsgQueryTopic query; |
| | | query.set_topic(req.topic()); |
| | | BHMsgHead head(InitMsgHead(GetType(query), proc_id())); |
| | | AddRoute(head, sock.id()); |
| | | |
| | | auto onQueryResult = [this, SendTo, req, cb](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { |
| | | MsgQueryTopicReply rep; |
| | | if (head.type() == kMsgTypeQueryTopicReply && imsg.ParseBody(rep)) { |
| | | auto &addr = rep.address(); |
| | | if (!addr.mq_id().empty()) { |
| | | topic_query_cache_.Store(req.topic(), addr); |
| | | SendTo(addr, req, cb); |
| | | } |
| | | } |
| | | }; |
| | | return sock.Send(&BHTopicCenterAddress(), head, query, std::move(onQueryResult)); |
| | | #endif |
| | | |
| | | } catch (...) { |
| | | SetLastError(eError, "internal error."); |
| | | return false; |
| | | } |
| | | } |
| | | |
| | | bool TopicNode::ClientSyncRequest(const MsgRequestTopic &request, std::string &out_proc_id, MsgRequestTopicReply &out_reply, const int timeout_ms) |
| | | bool TopicNode::ClientSyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &request, std::string &out_proc_id, MsgRequestTopicReply &out_reply, const int timeout_ms) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | |
| | | |
| | | BHAddress addr; |
| | | if (ClientQueryRPCTopic(request.topic(), addr, timeout_ms)) { |
| | | BHMsgHead head(InitMsgHead(GetType(request), proc_id())); |
| | | LOG_TRACE() << "node: " << SockNode().id() << ", topic dest: " << addr.mq_id(); |
| | | BHMsgHead head(InitMsgHead(GetType(request), proc_id(), ssn())); |
| | | AddRoute(head, sock.id()); |
| | | head.set_topic(request.topic()); |
| | | |
| | | MsgI reply_msg; |
| | | DEFER1(reply_msg.Release(shm_);); |
| | | 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); |
| | | return true; |
| | | } |
| | | } else { |
| | | SetLastError(eNotFound, "remote not found."); |
| | | } |
| | | } catch (...) { |
| | | SetLastError(eError, __func__ + std::string(" internal errer.")); |
| | |
| | | return false; |
| | | } |
| | | |
| | | int TopicNode::QueryRPCTopics(const Topic &topic, std::vector<NodeAddress> &addr, const int timeout_ms) |
| | | { |
| | | int n = 0; |
| | | MsgQueryTopic query; |
| | | query.set_topic(topic); |
| | | MsgQueryTopicReply rep; |
| | | BHAddress dest; // empty means local. |
| | | if (QueryTopicAddress(dest, query, rep, timeout_ms)) { |
| | | auto &ls = rep.node_address(); |
| | | n = ls.size(); |
| | | for (auto &na : ls) { |
| | | addr.push_back(na); |
| | | } |
| | | } |
| | | return n; |
| | | } |
| | | |
| | | bool TopicNode::ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | | auto &sock = SockClient(); |
| | | |
| | | if (topic_query_cache_.Find(topic, addr)) { |
| | | return true; |
| | | } |
| | | |
| | | MsgQueryTopic query; |
| | | query.set_topic(topic); |
| | | BHMsgHead head(InitMsgHead(GetType(query), proc_id())); |
| | | AddRoute(head, sock.id()); |
| | | |
| | | MsgI reply; |
| | | DEFER1(reply.Release(shm_)); |
| | | BHMsgHead reply_head; |
| | | |
| | | if (sock.SendAndRecv(&BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms)) { |
| | | if (reply_head.type() == kMsgTypeQueryTopicReply) { |
| | | MsgQueryTopicReply rep; |
| | | if (reply.ParseBody(rep)) { |
| | | addr = rep.address(); |
| | | if (addr.mq_id().empty()) { |
| | | return false; |
| | | } else { |
| | | topic_query_cache_.Store(topic, addr); |
| | | return true; |
| | | } |
| | | } |
| | | std::vector<NodeAddress> lst; |
| | | if (QueryRPCTopics(topic, lst, timeout_ms)) { |
| | | addr = lst.front().addr(); |
| | | if (addr.mq_id() != 0) { |
| | | topic_query_cache_.Store(topic, addr); |
| | | return true; |
| | | } |
| | | } else { |
| | | } |
| | | SetLastError(eNotFound, "remote not found."); |
| | | return false; |
| | | } |
| | | |
| | |
| | | |
| | | bool TopicNode::Publish(const MsgPublish &pub, const int timeout_ms) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | | try { |
| | | auto &sock = SockPub(); |
| | | BHMsgHead head(InitMsgHead(GetType(pub), proc_id())); |
| | | BHMsgHead head(InitMsgHead(GetType(pub), proc_id(), ssn())); |
| | | 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(shm());); |
| | | 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()); |
| | |
| | | |
| | | bool TopicNode::Subscribe(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | |
| | | MsgSubscribe sub; |
| | | sub.mutable_topics()->Swap(&topics); |
| | | |
| | | BHMsgHead head(InitMsgHead(GetType(sub), proc_id())); |
| | | BHMsgHead head(InitMsgHead(GetType(sub), proc_id(), ssn())); |
| | | 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(shm());); |
| | | 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()); |
| | |
| | | |
| | | bool TopicNode::RecvSub(std::string &proc_id, MsgPublish &pub, const int timeout_ms) |
| | | { |
| | | if (!IsRegistered()) { |
| | | SetLastError(eNotRegistered, "Not Registered."); |
| | | if (!IsOnline()) { |
| | | SetLastError(eNotRegistered, kErrMsgNotRegistered); |
| | | return false; |
| | | } |
| | | |
| | | auto &sock = SockSub(); |
| | | MsgI msg; |
| | | DEFER1(msg.Release(shm());); |
| | | DEFER1(msg.Release();); |
| | | BHMsgHead head; |
| | | //TODO error msg. |
| | | if (sock.SyncRecv(msg, head, timeout_ms) && head.type() == kMsgTypePublish) { |