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 |  173 ++++++++++++++++++++++++++++-----------------------------
 1 files changed, 84 insertions(+), 89 deletions(-)

diff --git a/src/socket.cpp b/src/socket.cpp
index 1c28bfa..f2b29f4 100644
--- a/src/socket.cpp
+++ b/src/socket.cpp
@@ -17,118 +17,113 @@
  */
 
 #include "socket.h"
-#include <chrono>
-#include "msg.h"
-#include "defs.h"
 #include "bh_util.h"
+#include "defs.h"
+#include "msg.h"
 
 using namespace bhome_msg;
 using namespace bhome_shm;
-using namespace std::chrono_literals;
 
-namespace {
-
-int GetSocketDefaultLen(ShmSocket::Type type) 
+namespace
 {
-    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);
-        
-    } catch (...) {
-        return false;
-    }
+	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;
+			}
+		};
+
+		RecvCB cb;
+		if (Find(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, 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::Subscribe(const std::vector<std::string> &topics, const int timeout_ms)
+bool ShmSocket::Stop()
 {
-    if (type_ != eSockSubscribe) {
-        return false;
-    }
-    assert(mq_);
-    try {
-        return mq_->Send(kBHBusQueueId, MakeSub(mq_->Id(), topics), timeout_ms);
-    } catch (...) {
-        return false;
-    }
+	std::lock_guard<std::mutex> lock(mutex_);
+	return StopNoLock();
 }
 
-bool ShmSocket::SetRecvCallback(const RecvCB &onRecv)
+bool ShmSocket::StopNoLock()
 {
-    std::lock_guard<std::mutex> lock(mutex_);
-    onRecv_ = onRecv;
-    cv_recv_cb_.notify_one();
-    return true;
+	if (run_.exchange(false)) {
+		for (auto &w : workers_) {
+			if (w.joinable()) {
+				w.join();
+			}
+		}
+		workers_.clear();
+		return true;
+	}
+	return false;
 }
 
-bool ShmSocket::HasRecvCB()
+bool ShmSocket::SyncRecv(bhome_msg::MsgI &msg, bhome::msg::BHMsgHead &head, const int timeout_ms)
 {
-    return static_cast<bool>(onRecv_);
+	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();
 }
-
-void ShmSocket::Stop()
-{
-    run_ = false;
-    for (auto &t : workers_) {
-        if (t.joinable()) {
-            t.join();
-        }
-    }
-}
\ No newline at end of file

--
Gitblit v1.8.0