From 4e5cb7960ce4e7e66d5190be67426aeca8b55c3d Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期五, 09 四月 2021 18:45:08 +0800
Subject: [PATCH] add heartbeat, not tested yet.

---
 src/socket.cpp |   77 +++++++++++++++++++-------------------
 1 files changed, 38 insertions(+), 39 deletions(-)

diff --git a/src/socket.cpp b/src/socket.cpp
index 73681f1..116175d 100644
--- a/src/socket.cpp
+++ b/src/socket.cpp
@@ -29,53 +29,51 @@
 
 } // namespace
 
-ShmSocket::ShmSocket(Shm &shm, const void *id, const int len) :
-    shm_(shm), run_(false)
+ShmSocket::ShmSocket(Shm &shm, const MQId &id, const int len) :
+    shm_(shm), run_(false), mq_(id, shm, len)
 {
-	if (id && len > 0) {
-		mq_.reset(new Queue(*static_cast<const MQId *>(id), shm, len));
-	}
 }
 ShmSocket::ShmSocket(bhome_shm::SharedMemory &shm, const int len) :
-    shm_(shm), run_(false)
-{
-	if (len > 0) {
-		mq_.reset(new Queue(shm_, len));
-	}
-}
+    shm_(shm), run_(false), mq_(shm, len) {}
 
 ShmSocket::~ShmSocket()
 {
 	Stop(); //TODO should stop in sub class, incase thread access sub class data.
 }
 
-bool ShmSocket::Start(const RecvCB &onData, int nworker)
+bool ShmSocket::Start(int nworker, const RecvCB &onData, const IdleCB &onIdle)
 {
-	if (!mq_) {
-		return false;
-	}
+	auto onRecvWithPerMsgCB = [this, onData](ShmSocket &socket, MsgI &imsg, BHMsgHead &head) {
+		RecvCB cb;
+		if (async_cbs_->Find(head.msg_id(), cb)) {
+			cb(socket, imsg, head);
+		} else if (onData) {
+			onData(socket, imsg, head);
+		} // else ignored, or dropped
+	};
 
-	std::lock_guard<std::mutex> lock(mutex_);
-	StopNoLock();
-	auto RecvProc = [this, onData]() {
-		while (run_) {
-			try {
-				MsgI imsg;
-				DEFER1(imsg.Release(shm_));
-				if (mq_->Recv(imsg, 100)) {
-					BHMsg msg;
-					if (imsg.Unpack(msg)) {
-						onData(*this, imsg, msg);
-					}
+	auto recvLoopBody = [this, onRecvWithPerMsgCB, onIdle]() {
+		try {
+			MsgI imsg;
+			if (mq().Recv(imsg, 10)) {
+				DEFER1(imsg.Release(shm()));
+				BHMsgHead head;
+				if (imsg.ParseHead(head)) {
+					onRecvWithPerMsgCB(*this, imsg, head);
 				}
-			} catch (...) {
+			} else if (onIdle) {
+				onIdle(*this);
 			}
+		} catch (...) {
 		}
 	};
 
+	std::lock_guard<std::mutex> lock(mutex_);
+	StopNoLock();
+
 	run_.store(true);
 	for (int i = 0; i < nworker; ++i) {
-		workers_.emplace_back(RecvProc);
+		workers_.emplace_back([this, recvLoopBody]() { while (run_) { recvLoopBody(); } });
 	}
 	return true;
 }
@@ -100,17 +98,18 @@
 	return false;
 }
 
-bool ShmSocket::SyncSend(const void *id, const bhome_msg::BHMsg &msg, const int timeout_ms)
-{
-	return mq_->Send(*static_cast<const MQId *>(id), msg, timeout_ms);
-}
-
-bool ShmSocket::SyncRecv(bhome_msg::BHMsg &msg, const int timeout_ms)
+bool ShmSocket::SyncRecv(bhome_msg::MsgI &msg, bhome::msg::BHMsgHead &head, const int timeout_ms)
 {
 	std::lock_guard<std::mutex> lock(mutex_);
-	if (!mq_ || RunningNoLock()) {
+	auto Recv = [&]() {
+		if (mq().Recv(msg, timeout_ms)) {
+			if (msg.ParseHead(head)) {
+				return true;
+			} else {
+				msg.Release(shm());
+			}
+		}
 		return false;
-	} else {
-		return mq_->Recv(msg, timeout_ms);
-	}
+	};
+	return !RunningNoLock() && Recv();
 }

--
Gitblit v1.8.0