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 |   83 +++++++++++++++++++++++++++++++++++++----
 1 files changed, 74 insertions(+), 9 deletions(-)

diff --git a/src/pubsub.cpp b/src/pubsub.cpp
index e38c445..add968c 100644
--- a/src/pubsub.cpp
+++ b/src/pubsub.cpp
@@ -17,15 +17,18 @@
  */
 #include "pubsub.h"
 #include <chrono>
+#include "bh_util.h"
+#include "defs.h"
 
 namespace bhome_shm {
 
 using namespace std::chrono_literals;
-const MQId kBusQueueId = boost::uuids::string_generator()("01234567-89ab-cdef-8349-1234567890ff");
 const int kMaxWorker = 16;
+using namespace bhome_msg;
 
 BusManager::BusManager(SharedMemory &shm):
-busq_(kBusQueueId, shm, 1000),
+shm_(shm),
+busq_(kBHBusQueueId, shm, 1000),
 run_(false)
 {
 }
@@ -42,15 +45,12 @@
     // start
     auto Worker = [&](){
         while (this->run_) {
-            std::this_thread::sleep_for(100ms);
             BusManager &self = *this;
-            Msg msg;
+            BHMsg msg;
             const int timeout_ms = 100;
-            if (!self.busq_.Recv(msg, timeout_ms)) {
-                continue;
+            if (self.busq_.Recv(msg, timeout_ms)) {
+                self.OnMsg(msg);
             }
-            // handle msg;
-            // type: subscribe(topic), publish(topic, data)
         }
     };
 
@@ -59,12 +59,13 @@
     for (int i = 0; i < n; ++i) {
         workers_.emplace_back(Worker);
     }
+    return true;
 }
 
 bool BusManager::Stop()
 {
     std::lock_guard<std::mutex> guard(mutex_);
-    StopNoLock();
+    return StopNoLock();
 }
 
 bool BusManager::StopNoLock()
@@ -75,7 +76,71 @@
                 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