From 4deeafbd502dc3c57dab8ad6ca601a38a9e7f074 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期二, 06 四月 2021 19:10:49 +0800
Subject: [PATCH] add uni center.
---
src/pubsub.cpp | 96 ++++++++++++++++++++++-------------------------
1 files changed, 45 insertions(+), 51 deletions(-)
diff --git a/src/pubsub.cpp b/src/pubsub.cpp
index e38c445..90688ec 100644
--- a/src/pubsub.cpp
+++ b/src/pubsub.cpp
@@ -16,67 +16,61 @@
* =====================================================================================
*/
#include "pubsub.h"
-#include <chrono>
-
-namespace bhome_shm {
+#include "bh_util.h"
+#include "defs.h"
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),
-run_(false)
+bool SocketPublish::Publish(const Topic &topic, const void *data, const size_t size, const int timeout_ms)
{
-}
-
-BusManager::~BusManager()
-{
- Stop();
+ try {
+ MsgI imsg;
+ if (!imsg.MakeRC(shm(), MakePub(topic, data, size))) {
+ return false;
+ }
+ DEFER1(imsg.Release(shm()));
+ return ShmMsgQueue::Send(shm(), kBHTopicBus, imsg, timeout_ms);
+ } catch (...) {
+ return false;
+ }
}
-bool BusManager::Start(const int nworker)
+bool SocketSubscribe::Subscribe(const std::vector<Topic> &topics, const int timeout_ms)
{
- 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);
- }
+ try {
+ return mq().Send(kBHTopicBus, MakeSub(mq().Id(), topics), timeout_ms);
+ } catch (...) {
+ return false;
+ }
}
-bool BusManager::Stop()
+bool SocketSubscribe::StartRecv(const TopicDataCB &tdcb, int nworker)
{
- std::lock_guard<std::mutex> guard(mutex_);
- StopNoLock();
+ auto AsyncRecvProc = [this, tdcb](BHMsg &msg) {
+ if (msg.type() == kMsgTypePublish) {
+ MsgPub d;
+ if (d.ParseFromString(msg.body())) {
+ tdcb(d.topic(), d.data());
+ }
+ } else {
+ // ignored, or dropped
+ }
+ };
+
+ return tdcb && Start(AsyncRecvProc, nworker);
}
-bool BusManager::StopNoLock()
+bool SocketSubscribe::RecvSub(Topic &topic, std::string &data, const int timeout_ms)
{
- if (run_.exchange(false)) {
- for (auto &w: workers_) {
- if (w.joinable()) {
- w.join();
- }
- }
- }
-}
-
-} // namespace bhome_shm
-
+ BHMsg msg;
+ if (SyncRecv(msg, timeout_ms) && msg.type() == kMsgTypePublish) {
+ MsgPub d;
+ if (d.ParseFromString(msg.body())) {
+ d.mutable_topic()->swap(topic);
+ d.mutable_data()->swap(data);
+ return true;
+ }
+ }
+ return false;
+}
\ No newline at end of file
--
Gitblit v1.8.0