From 6aa7e4c37a70709e7348bd16407c5983a563ed76 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期一, 29 三月 2021 21:11:34 +0800
Subject: [PATCH] test pub/sub msg; fix update last_time;

---
 src/pubsub.cpp |   72 +++++++++++++++++++++++++++++++++--
 1 files changed, 67 insertions(+), 5 deletions(-)

diff --git a/src/pubsub.cpp b/src/pubsub.cpp
index ee2614a..686c7d5 100644
--- a/src/pubsub.cpp
+++ b/src/pubsub.cpp
@@ -17,14 +17,17 @@
  */
 #include "pubsub.h"
 #include <chrono>
+#include "bh_util.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):
+shm_(shm),
 busq_(kBusQueueId, shm, 1000),
 run_(false)
 {
@@ -42,15 +45,12 @@
     // 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;
+            if (self.busq_.Recv(msg, timeout_ms)) {
+                self.OnMsg(msg);
             }
-            // handle msg;
-            // type: subscribe(topic), publish(topic, data)
         }
     };
 
@@ -81,5 +81,67 @@
     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