From 4deeafbd502dc3c57dab8ad6ca601a38a9e7f074 Mon Sep 17 00:00:00 2001 From: lichao <lichao@aiotlink.com> Date: 星期二, 06 四月 2021 19:10:49 +0800 Subject: [PATCH] add uni center. --- src/pubsub.cpp | 58 +++++++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 files changed, 55 insertions(+), 3 deletions(-) diff --git a/src/pubsub.cpp b/src/pubsub.cpp index b592113..90688ec 100644 --- a/src/pubsub.cpp +++ b/src/pubsub.cpp @@ -16,9 +16,61 @@ * ===================================================================================== */ #include "pubsub.h" +#include "bh_util.h" +#include "defs.h" -namespace bhome_shm { +using namespace std::chrono_literals; +using namespace bhome_msg; - -} // namespace bhome_shm +bool SocketPublish::Publish(const Topic &topic, const void *data, const size_t size, const int timeout_ms) +{ + try { + MsgI imsg; + if (!imsg.MakeRC(shm(), MakePub(topic, data, size))) { + return false; + } + DEFER1(imsg.Release(shm())); + return ShmMsgQueue::Send(shm(), kBHTopicBus, imsg, timeout_ms); + } catch (...) { + return false; + } +} +bool SocketSubscribe::Subscribe(const std::vector<Topic> &topics, const int timeout_ms) +{ + try { + return mq().Send(kBHTopicBus, MakeSub(mq().Id(), topics), timeout_ms); + } catch (...) { + return false; + } +} + +bool SocketSubscribe::StartRecv(const TopicDataCB &tdcb, int nworker) +{ + auto AsyncRecvProc = [this, tdcb](BHMsg &msg) { + if (msg.type() == kMsgTypePublish) { + MsgPub d; + if (d.ParseFromString(msg.body())) { + tdcb(d.topic(), d.data()); + } + } else { + // ignored, or dropped + } + }; + + return tdcb && Start(AsyncRecvProc, nworker); +} + +bool SocketSubscribe::RecvSub(Topic &topic, std::string &data, const int timeout_ms) +{ + BHMsg msg; + if (SyncRecv(msg, timeout_ms) && msg.type() == kMsgTypePublish) { + MsgPub d; + if (d.ParseFromString(msg.body())) { + d.mutable_topic()->swap(topic); + d.mutable_data()->swap(data); + return true; + } + } + return false; +} \ No newline at end of file -- Gitblit v1.8.0