From 5657dca25451cfb63a90a3908db0c464fe3f343d Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期一, 29 三月 2021 14:16:49 +0800
Subject: [PATCH] add protobuf; refactor.
---
src/pubsub.cpp | 58 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 58 insertions(+), 0 deletions(-)
diff --git a/src/pubsub.cpp b/src/pubsub.cpp
index b592113..e38c445 100644
--- a/src/pubsub.cpp
+++ b/src/pubsub.cpp
@@ -16,9 +16,67 @@
* =====================================================================================
*/
#include "pubsub.h"
+#include <chrono>
namespace bhome_shm {
+using namespace std::chrono_literals;
+const MQId kBusQueueId = boost::uuids::string_generator()("01234567-89ab-cdef-8349-1234567890ff");
+const int kMaxWorker = 16;
+
+BusManager::BusManager(SharedMemory &shm):
+busq_(kBusQueueId, 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_) {
+ std::this_thread::sleep_for(100ms);
+ BusManager &self = *this;
+ Msg msg;
+ const int timeout_ms = 100;
+ if (!self.busq_.Recv(msg, timeout_ms)) {
+ continue;
+ }
+ // handle msg;
+ // type: subscribe(topic), publish(topic, data)
+ }
+ };
+
+ run_.store(true);
+ const int n = std::min(nworker, kMaxWorker);
+ for (int i = 0; i < n; ++i) {
+ workers_.emplace_back(Worker);
+ }
+}
+
+bool BusManager::Stop()
+{
+ std::lock_guard<std::mutex> guard(mutex_);
+ StopNoLock();
+}
+
+bool BusManager::StopNoLock()
+{
+ if (run_.exchange(false)) {
+ for (auto &w: workers_) {
+ if (w.joinable()) {
+ w.join();
+ }
+ }
+ }
+}
+
} // namespace bhome_shm
--
Gitblit v1.8.0