From 8cbb55f3066f71f5a4328193414d4555c87e96be Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期二, 30 三月 2021 11:41:30 +0800
Subject: [PATCH] refactor.
---
src/pubsub.cpp | 123 +++++++++++++++++++++++++++++++++++++++++
1 files changed, 123 insertions(+), 0 deletions(-)
diff --git a/src/pubsub.cpp b/src/pubsub.cpp
index b592113..add968c 100644
--- a/src/pubsub.cpp
+++ b/src/pubsub.cpp
@@ -16,9 +16,132 @@
* =====================================================================================
*/
#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)
+{
+}
+BusManager::~BusManager()
+{
+ Stop();
+}
+
+bool BusManager::Start(const 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);
+ }
+ }
+ };
+
+ run_.store(true);
+ const int n = std::min(nworker, kMaxWorker);
+ for (int i = 0; i < n; ++i) {
+ workers_.emplace_back(Worker);
+ }
+ return true;
+}
+
+bool BusManager::Stop()
+{
+ 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
--
Gitblit v1.8.0