From c338820e4db43ad32c20ff429a038b06bcb980f8 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期四, 08 四月 2021 18:13:25 +0800
Subject: [PATCH] BIG change, join center,bus; now msg is head+body.
---
src/pubsub.cpp | 171 +++++++++++++++++++-------------------------------------
1 files changed, 58 insertions(+), 113 deletions(-)
diff --git a/src/pubsub.cpp b/src/pubsub.cpp
index add968c..471c63c 100644
--- a/src/pubsub.cpp
+++ b/src/pubsub.cpp
@@ -16,132 +16,77 @@
* =====================================================================================
*/
#include "pubsub.h"
-#include <chrono>
#include "bh_util.h"
#include "defs.h"
-namespace bhome_shm {
-
using namespace std::chrono_literals;
-const int kMaxWorker = 16;
using namespace bhome_msg;
-BusManager::BusManager(SharedMemory &shm):
-shm_(shm),
-busq_(kBHBusQueueId, shm, 1000),
-run_(false)
+bool SocketPublish::Publish(const std::string &proc_id, const Topic &topic, const void *data, const size_t size, const int timeout_ms)
{
+ try {
+ MsgPublish pub;
+ pub.set_topic(topic);
+ pub.set_data(data, size);
+ BHMsgHead head(InitMsgHead(GetType(pub), proc_id));
+ MsgI imsg;
+ if (imsg.MakeRC(shm(), head, pub)) {
+ DEFER1(imsg.Release(shm()));
+ return ShmMsgQueue::Send(shm(), BHTopicBusAddress(), imsg, timeout_ms);
+ }
+ } catch (...) {
+ }
+ return false;
}
-
-BusManager::~BusManager()
+namespace
{
- Stop();
+inline void AddRoute(BHMsgHead &head, const MQId &id) { head.add_route()->set_mq_id(&id, sizeof(id)); }
+
+} // namespace
+bool SocketSubscribe::Subscribe(const std::string &proc_id, const std::vector<Topic> &topics, const int timeout_ms)
+{
+ try {
+ MsgSubscribe sub;
+ for (auto &topic : topics) {
+ sub.add_topics(topic);
+ }
+ BHMsgHead head(InitMsgHead(GetType(sub), proc_id));
+ AddRoute(head, mq().Id());
+
+ return Send(&BHTopicBusAddress(), head, sub, timeout_ms);
+ } catch (...) {
+ return false;
+ }
}
-bool BusManager::Start(const int nworker)
+bool SocketSubscribe::StartRecv(const TopicDataCB &tdcb, int nworker)
{
- std::lock_guard<std::mutex> guard(mutex_);
- StopNoLock();
- // start
- auto Worker = [&](){
- while (this->run_) {
- BusManager &self = *this;
- BHMsg msg;
- const int timeout_ms = 100;
- if (self.busq_.Recv(msg, timeout_ms)) {
- self.OnMsg(msg);
- }
- }
- };
+ auto AsyncRecvProc = [this, tdcb](ShmSocket &, MsgI &imsg, BHMsgHead &head) {
+ if (head.type() == kMsgTypePublish) {
+ MsgPublish pub;
+ if (imsg.ParseBody(pub)) {
+ tdcb(head.proc_id(), pub.topic(), pub.data());
+ }
+ } else {
+ // ignored, or dropped
+ }
+ };
- run_.store(true);
- const int n = std::min(nworker, kMaxWorker);
- for (int i = 0; i < n; ++i) {
- workers_.emplace_back(Worker);
- }
- return true;
+ return tdcb && Start(AsyncRecvProc, nworker);
}
-bool BusManager::Stop()
+bool SocketSubscribe::RecvSub(std::string &proc_id, Topic &topic, std::string &data, const int timeout_ms)
{
- std::lock_guard<std::mutex> guard(mutex_);
- return StopNoLock();
-}
-
-bool BusManager::StopNoLock()
-{
- if (run_.exchange(false)) {
- for (auto &w: workers_) {
- if (w.joinable()) {
- w.join();
- }
- }
- return true;
- }
- return false;
-}
-
-void BusManager::OnMsg(const BHMsg &msg)
-{
- auto OnSubChange = [&](auto &&update) {
- DataSub sub;
- if (!msg.route().empty() && sub.ParseFromString(msg.body()) && !sub.topics().empty()) {
- assert(sizeof(MQId) == msg.route(0).mq_id().size());
- MQId client;
- memcpy(&client, msg.route(0).mq_id().data(), sizeof(client));
-
- std::lock_guard<std::mutex> guard(mutex_);
- auto &topics = sub.topics();
- for (auto &topic : topics) {
- try {
- update(topic, client);
- } catch(...) {
- //TODO log error
- }
- }
- }
- };
-
- auto Sub1 = [this](const std::string &topic, const MQId &id) {
- records_[topic].insert(id);
- };
-
- auto Unsub1 = [this](const std::string &topic, const MQId &id) {
- auto pos = records_.find(topic);
- if (pos != records_.end()) {
- if (pos->second.erase(id) && pos->second.empty()) {
- records_.erase(pos);
- }
- }
- };
-
- auto OnPublish = [&]() {
- DataPub pub;
- MsgI pubmsg;
- if (!pub.ParseFromString(msg.body()) || !pubmsg.MakeRC(shm_, msg)) {
- return;
- }
- DEFER1(pubmsg.Release(shm_));
-
- std::lock_guard<std::mutex> guard(mutex_);
- auto pos = records_.find(pub.topic());
- if (pos != records_.end() && !pos->second.empty()) {
- auto &clients = pos->second;
- for (auto &cli : clients) {
- busq_.Send(cli, pubmsg, 100);
- }
- } else {
- // printf("invalid topic: %s\n", pub.topic().c_str());
- }
- };
-
- switch (msg.type()) {
- case kMsgTypeSubscribe: OnSubChange(Sub1); break;
- case kMsgTypeUnsubscribe: OnSubChange(Unsub1); break;
- case kMsgTypePublish : OnPublish(); break;
- default: break;
- }
-}
-
-} // namespace bhome_shm
-
+ MsgI msg;
+ BHMsgHead head;
+ if (SyncRecv(msg, head, timeout_ms) && head.type() == kMsgTypePublish) {
+ MsgPublish pub;
+ if (msg.ParseBody(pub)) {
+ head.mutable_proc_id()->swap(proc_id);
+ pub.mutable_topic()->swap(topic);
+ pub.mutable_data()->swap(data);
+ return true;
+ }
+ }
+ return false;
+}
\ No newline at end of file
--
Gitblit v1.8.0