| | |
| | | |
| | | BusManager::BusManager(SharedMemory &shm): |
| | | shm_(shm), |
| | | busq_(kBHBusQueueId, shm, 1000), |
| | | busq_(kBHBusQueueId, shm, 16), |
| | | run_(false) |
| | | { |
| | | } |
| | |
| | | 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); |
| | |
| | | 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()) { |
| | |
| | | |
| | | 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); }); |
| | | } |
| | | }; |
| | | |