From c338820e4db43ad32c20ff429a038b06bcb980f8 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期四, 08 四月 2021 18:13:25 +0800
Subject: [PATCH] BIG change, join center,bus; now msg is head+body.

---
 src/socket.cpp |  161 ++++++++++++++++++++++++++---------------------------
 1 files changed, 78 insertions(+), 83 deletions(-)

diff --git a/src/socket.cpp b/src/socket.cpp
index 21928b8..f2b29f4 100644
--- a/src/socket.cpp
+++ b/src/socket.cpp
@@ -20,115 +20,110 @@
 #include "bh_util.h"
 #include "defs.h"
 #include "msg.h"
-#include <chrono>
 
 using namespace bhome_msg;
 using namespace bhome_shm;
-using namespace std::chrono_literals;
 
 namespace
 {
 
-int GetSocketDefaultLen(ShmSocket::Type type)
-{
-	switch (type) {
-	case ShmSocket::eSockRequest: return 12;
-	case ShmSocket::eSockReply: return 64;
-	case ShmSocket::eSockPublish: return 0;
-	case ShmSocket::eSockSubscribe: return 64;
-	default: return 0;
-	}
-}
-
 } // namespace
 
-ShmSocket::ShmSocket(Type type, bhome_shm::SharedMemory &shm) :
-    shm_(shm), type_(type), run_(false)
-{
-	int len = GetSocketDefaultLen(type);
-	if (len != 0) {
-		mq_.reset(new Queue(shm_, len));
-
-		auto RecvProc = [this]() {
-			while (run_) {
-				try {
-					std::unique_lock<std::mutex> lk(mutex_);
-					if (cv_recv_cb_.wait_for(lk, 100ms, [this]() { return HasRecvCB(); })) {
-						BHMsg msg;
-						if (mq_->Recv(msg, 100)) {
-							this->onRecv_(msg);
-						}
-					}
-				} catch (...) {
-				}
-			}
-		};
-		run_.store(true);
-		workers_.emplace_back(RecvProc);
-	}
-}
-
-ShmSocket::ShmSocket(Type type) :
-    ShmSocket(type, BHomeShm())
+ShmSocket::ShmSocket(Shm &shm, const MQId &id, const int len) :
+    shm_(shm), run_(false), mq_(id, shm, len)
 {
 }
+ShmSocket::ShmSocket(bhome_shm::SharedMemory &shm, const int len) :
+    shm_(shm), run_(false), mq_(shm, len) {}
 
 ShmSocket::~ShmSocket()
 {
-	Stop();
+	Stop(); //TODO should stop in sub class, incase thread access sub class data.
 }
 
-bool ShmSocket::Publish(const std::string &topic, const void *data, const size_t size, const int timeout_ms)
+bool ShmSocket::Start(int nworker, const RecvCB &onData, const IdleCB &onIdle)
 {
-	if (type_ != eSockPublish) {
-		return false;
-	}
-	assert(!mq_);
-	try {
-		MsgI imsg;
-		if (!imsg.MakeRC(shm_, MakePub(topic, data, size))) {
-			return false;
-		}
-		DEFER1(imsg.Release(shm_));
-		return Queue::Send(shm_, kBHBusQueueId, imsg, timeout_ms);
+	auto onRecv = [this, onData](ShmSocket &socket, MsgI &imsg, BHMsgHead &head) {
+		auto Find = [&](RecvCB &cb) {
+			std::lock_guard<std::mutex> lock(mutex());
+			const std::string &msgid = head.msg_id();
+			auto pos = async_cbs_.find(msgid);
+			if (pos != async_cbs_.end()) {
+				cb.swap(pos->second);
+				async_cbs_.erase(pos);
+				return true;
+			} else {
+				return false;
+			}
+		};
 
-	} catch (...) {
-		return false;
-	}
-}
+		RecvCB cb;
+		if (Find(cb)) {
+			cb(socket, imsg, head);
+		} else if (onData) {
+			onData(socket, imsg, head);
+		} // else ignored, or dropped
+	};
 
-bool ShmSocket::Subscribe(const std::vector<std::string> &topics, const int timeout_ms)
-{
-	if (type_ != eSockSubscribe) {
-		return false;
-	}
-	assert(mq_);
-	try {
-		return mq_->Send(kBHBusQueueId, MakeSub(mq_->Id(), topics), timeout_ms);
-	} catch (...) {
-		return false;
-	}
-}
-
-bool ShmSocket::SetRecvCallback(const RecvCB &onRecv)
-{
 	std::lock_guard<std::mutex> lock(mutex_);
-	onRecv_ = onRecv;
-	cv_recv_cb_.notify_one();
+	StopNoLock();
+	auto RecvProc = [this, onRecv, onIdle]() {
+		while (run_) {
+			try {
+				MsgI imsg;
+				if (mq().Recv(imsg, 10)) {
+					DEFER1(imsg.Release(shm()));
+					BHMsgHead head;
+					if (imsg.ParseHead(head)) {
+						onRecv(*this, imsg, head);
+					}
+				} else if (onIdle) {
+					onIdle(*this);
+				}
+			} catch (...) {
+			}
+		}
+	};
+
+	run_.store(true);
+	for (int i = 0; i < nworker; ++i) {
+		workers_.emplace_back(RecvProc);
+	}
 	return true;
 }
 
-bool ShmSocket::HasRecvCB()
+bool ShmSocket::Stop()
 {
-	return static_cast<bool>(onRecv_);
+	std::lock_guard<std::mutex> lock(mutex_);
+	return StopNoLock();
 }
 
-void ShmSocket::Stop()
+bool ShmSocket::StopNoLock()
 {
-	run_ = false;
-	for (auto &t : workers_) {
-		if (t.joinable()) {
-			t.join();
+	if (run_.exchange(false)) {
+		for (auto &w : workers_) {
+			if (w.joinable()) {
+				w.join();
+			}
 		}
+		workers_.clear();
+		return true;
 	}
-}
\ No newline at end of file
+	return false;
+}
+
+bool ShmSocket::SyncRecv(bhome_msg::MsgI &msg, bhome::msg::BHMsgHead &head, const int timeout_ms)
+{
+	std::lock_guard<std::mutex> lock(mutex_);
+	auto Recv = [&]() {
+		if (mq().Recv(msg, timeout_ms)) {
+			if (msg.ParseHead(head)) {
+				return true;
+			} else {
+				msg.Release(shm());
+			}
+		}
+		return false;
+	};
+	return !RunningNoLock() && Recv();
+}

--
Gitblit v1.8.0