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.h | 43 ++++++++++++++++++++++++++++++++++++++-----
1 files changed, 38 insertions(+), 5 deletions(-)
diff --git a/src/pubsub.h b/src/pubsub.h
index 0d7f4f0..3c3d4ad 100644
--- a/src/pubsub.h
+++ b/src/pubsub.h
@@ -18,12 +18,45 @@
#ifndef PUBSUB_4KGRA997
#define PUBSUB_4KGRA997
-#include "shm.h"
+#include "defs.h"
+#include "socket.h"
+#include <string>
-namespace bhome_shm {
+class SocketPublish
+{
+ typedef ShmSocket Socket;
+ Socket::Shm &shm_;
+ Socket::Shm &shm() { return shm_; }
-bool Subscribe(const std::string &topic);
-
-} // namespace bhome_shm
+public:
+ SocketPublish(Socket::Shm &shm) :
+ shm_(shm) {}
+ SocketPublish() :
+ SocketPublish(BHomeShm()) {}
+ bool Publish(const Topic &topic, const void *data, const size_t size, const int timeout_ms);
+ bool Publish(const Topic &topic, const std::string &data, const int timeout_ms)
+ {
+ return Publish(topic, data.data(), data.size(), timeout_ms);
+ }
+};
+
+// socket subscribe
+class SocketSubscribe : private ShmSocket
+{
+ typedef ShmSocket Socket;
+
+public:
+ SocketSubscribe(Socket::Shm &shm) :
+ Socket(shm, 64) {}
+ SocketSubscribe() :
+ SocketSubscribe(BHomeShm()) {}
+ ~SocketSubscribe() { Stop(); }
+
+ typedef std::function<void(const Topic &topic, const std::string &data)> TopicDataCB;
+ bool StartRecv(const TopicDataCB &tdcb, int nworker = 2);
+ bool Stop() { return Socket::Stop(); }
+ bool Subscribe(const std::vector<Topic> &topics, const int timeout_ms);
+ bool RecvSub(Topic &topic, std::string &data, const int timeout_ms);
+};
#endif // end of include guard: PUBSUB_4KGRA997
--
Gitblit v1.8.0