From 3f8ae6cf4f03be83f16846af32d73dd89b937c40 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期二, 30 三月 2021 15:45:54 +0800
Subject: [PATCH] add dot separated topic partial match; refactor.
---
src/pubsub.cpp | 64 +++++++++++++++++++++++++------
1 files changed, 51 insertions(+), 13 deletions(-)
diff --git a/src/pubsub.cpp b/src/pubsub.cpp
index add968c..a0dc4e9 100644
--- a/src/pubsub.cpp
+++ b/src/pubsub.cpp
@@ -28,7 +28,7 @@
BusManager::BusManager(SharedMemory &shm):
shm_(shm),
-busq_(kBHBusQueueId, shm, 1000),
+busq_(kBHBusQueueId, shm, 16),
run_(false)
{
}
@@ -46,7 +46,7 @@
auto Worker = [&](){
while (this->run_) {
BusManager &self = *this;
- BHMsg msg;
+ MsgI msg;
const int timeout_ms = 100;
if (self.busq_.Recv(msg, timeout_ms)) {
self.OnMsg(msg);
@@ -81,8 +81,15 @@
return false;
}
-void BusManager::OnMsg(const BHMsg &msg)
+void BusManager::OnMsg(MsgI &imsg)
{
+ DEFER1(imsg.Release(shm_));
+
+ BHMsg msg;
+ if (!imsg.Unpack(msg)) {
+ return;
+ }
+
auto OnSubChange = [&](auto &&update) {
DataSub sub;
if (!msg.route().empty() && sub.ParseFromString(msg.body()) && !sub.topics().empty()) {
@@ -117,21 +124,52 @@
auto OnPublish = [&]() {
DataPub pub;
- MsgI pubmsg;
- if (!pub.ParseFromString(msg.body()) || !pubmsg.MakeRC(shm_, msg)) {
+ if (!pub.ParseFromString(msg.body())) {
return;
}
- DEFER1(pubmsg.Release(shm_));
+ auto FindClients = [&](const std::string &topic){
+ Clients dests;
+ std::lock_guard<std::mutex> guard(mutex_);
+ auto Find1 = [&](const std::string &t) {
+ auto pos = records_.find(topic);
+ if (pos != records_.end() && !pos->second.empty()) {
+ auto &clients = pos->second;
+ for (auto &cli : clients) {
+ dests.insert(cli);
+ }
+ }
+ };
+ Find1(topic);
- std::lock_guard<std::mutex> guard(mutex_);
- auto pos = records_.find(pub.topic());
- if (pos != records_.end() && !pos->second.empty()) {
- auto &clients = pos->second;
- for (auto &cli : clients) {
- busq_.Send(cli, pubmsg, 100);
+ //TODO check and adjust topic on client side sub/pub.
+ size_t pos = 0;
+ while (true) {
+ pos = topic.find(kTopicSep, pos);
+ if (pos == topic.npos || ++pos == topic.size()) {
+ // Find1(std::string()); // sub all.
+ break;
+ } else {
+ Find1(topic.substr(0, pos));
+ }
}
+ return dests;
+ };
+
+ auto Dispatch = [&](auto &&send1) {
+ const Clients &clients(FindClients(pub.topic()));
+ for (auto &cli : clients) {
+ send1(cli);
+ }
+ };
+
+ if (imsg.IsCounted()) {
+ Dispatch([&](const MQId &cli) { busq_.Send(cli, imsg, 100); });
} else {
- // printf("invalid topic: %s\n", pub.topic().c_str());
+ MsgI pubmsg;
+ if (!pubmsg.MakeRC(shm_, msg)) { return; }
+ DEFER1(pubmsg.Release(shm_));
+
+ Dispatch([&](const MQId &cli) { busq_.Send(cli, pubmsg, 100); });
}
};
--
Gitblit v1.8.0