From f51636c193d032723c47343e39ff8296db350200 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期一, 29 三月 2021 18:04:00 +0800
Subject: [PATCH] change msg to use protobuf, add more msg type.
---
src/pubsub.cpp | 61 ++++++++++++++++++++++++++++++
1 files changed, 61 insertions(+), 0 deletions(-)
diff --git a/src/pubsub.cpp b/src/pubsub.cpp
index b592113..ee2614a 100644
--- a/src/pubsub.cpp
+++ b/src/pubsub.cpp
@@ -16,9 +16,70 @@
* =====================================================================================
*/
#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;
+ BHMsg 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);
+ }
+ 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;
+}
+
} // namespace bhome_shm
--
Gitblit v1.8.0