From b55ffe89f4b237be5f79232cfddfe22bfdb87c64 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期四, 01 四月 2021 13:23:48 +0800
Subject: [PATCH] make req/rep,sub/pub sockets sub class;
---
src/socket.h | 75 ++++++++++++++++++-------------------
1 files changed, 37 insertions(+), 38 deletions(-)
diff --git a/src/socket.h b/src/socket.h
index e65ac83..20da7c0 100644
--- a/src/socket.h
+++ b/src/socket.h
@@ -20,55 +20,54 @@
#define SOCKET_GWTJHBPO
#include "shm_queue.h"
-#include <vector>
-#include <thread>
-#include <memory>
-#include <functional>
-#include <mutex>
-#include <condition_variable>
#include <atomic>
+#include <boost/noncopyable.hpp>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <vector>
-class ShmSocket
+class ShmSocket : private boost::noncopyable
{
- typedef bhome_shm::ShmMsgQueue Queue;
+protected:
+ typedef bhome_shm::ShmMsgQueue Queue;
+
public:
- enum Type {
- eSockRequest,
- eSockReply,
- eSockSubscribe,
- eSockPublish,
- };
- typedef std::function<void (bhome_msg::BHMsg &msg)> RecvCB;
+ typedef bhome_shm::SharedMemory Shm;
+ typedef std::function<void(bhome_msg::BHMsg &msg)> RecvCB;
+ typedef std::function<void(bhome_msg::MsgI &imsg)> RecvRawCB;
- ShmSocket(Type type);
- ShmSocket(Type type, bhome_shm::SharedMemory &shm);
- ~ShmSocket();
+ ShmSocket(Shm &shm, const int len = 12);
+ ~ShmSocket();
- // bool Request(const std::string &topic, const void *data, const size_t size, onReply);
- bool RequestAndWait() { return false; } // call Request, and wait onReply notify cv
+ // start recv.
+ bool Start(const RecvCB &onData, int nworker = 1);
+ bool StartRaw(const RecvRawCB &onData, int nworker = 1);
+ bool Stop();
+ size_t Pending() const { return mq_ ? mq_->Pending() : 0; }
- // bool HandleRequest(onData);
- bool ReadRequest(); // exclude with HandleRequest
- bool SendReply(); // exclude with HandleRequest
+protected:
+ ShmSocket(Shm &shm, const void *id, const int len);
+ Shm &shm() { return shm_; }
+ const Shm &shm() const { return shm_; }
+ Queue &mq() { return *mq_; } // programmer should make sure that mq_ is valid.
+ const Queue &mq() const { return *mq_; }
+ std::mutex &mutex() { return mutex_; }
- bool Publish(const std::string &topic, const void *data, const size_t size, const int timeout_ms);
- bool Subscribe(const std::vector<std::string> &topics, const int timeout_ms);
- bool RecvSub(std::string &topic, std::string &data, const int timeout_ms);
- bool SetRecvCallback(const RecvCB &onRecv);
+ bool SyncSend(const void *id, const bhome_msg::BHMsg &msg, const int timeout_ms);
+ bool SyncRecv(bhome_msg::BHMsg &msg, const int timeout_ms);
+
private:
- bool HasRecvCB();
- void Stop();
+ bool StopNoLock();
+ bool RunningNoLock() { return !workers_.empty(); }
- bhome_shm::SharedMemory &shm_;
- Type type_;
- std::vector<std::thread> workers_;
- std::mutex mutex_;
- std::condition_variable cv_recv_cb_;
- std::atomic<bool> run_;
- RecvCB onRecv_;
+ Shm &shm_;
+ std::vector<std::thread> workers_;
+ std::mutex mutex_;
+ std::atomic<bool> run_;
- std::unique_ptr<Queue> mq_;
+ std::unique_ptr<Queue> mq_;
};
-
#endif // end of include guard: SOCKET_GWTJHBPO
--
Gitblit v1.8.0