valib/bhshmq.git

			@@ -37,9 +37,9 @@
			{
			public:
			typedef std::string ProcId;
			typedef std::string Address;
			typedef MQId Address;
			typedef bhome_msg::ProcInfo ProcInfo;
			typedef std::function<void(Address const &)> Cleaner;
			typedef std::function<void(Address const)> Cleaner;

			private:
			enum {
			@@ -84,7 +84,7 @@
			WeakNode weak_node_;
			bool operator<(const TopicDest &a) const { return mq_ < a.mq_; }
			};
			inline const std::string &SrcAddr(const BHMsgHead &head) { return head.route(0).mq_id(); }
			inline MQId SrcAddr(const BHMsgHead &head) { return head.route(0).mq_id(); }
			inline bool MatchAddr(std::set<Address> const &addrs, const Address &addr) { return addrs.find(addr) != addrs.end(); }

			NodeCenter(const std::string &id, const Cleaner &cleaner, const int64_t offline_time, const int64_t kill_time) :
			@@ -166,11 +166,23 @@
			return HandleMsg<MsgCommonReply, Func>(head, op);
			}

			MsgCommonReply Unregister(const BHMsgHead &head, MsgUnregister &msg)
			{
			return HandleMsg(
			head, [&](Node node) -> MsgCommonReply {
			NodeInfo &ni = *node;
			auto now = NowSec(); // just set to offline.
			ni.state_.timestamp_ = now - offline_time_;
			ni.state_.UpdateState(now, offline_time_, kill_time_);
			return MakeReply(eSuccess);
			});
			}

			MsgCommonReply RegisterRPC(const BHMsgHead &head, MsgRegisterRPC &msg)
			{
			return HandleMsg(
			head, [&](Node node) -> MsgCommonReply {
			auto &src = SrcAddr(head);
			auto src = SrcAddr(head);
			auto &topics = msg.topics().topic_list();
			node->services_[src].insert(topics.begin(), topics.end());
			TopicDest dest = {src, node};
			@@ -206,20 +218,17 @@
			auto query = [&](Node self) -> MsgQueryTopicReply {
			auto pos = service_map_.find(req.topic());
			if (pos != service_map_.end() && !pos->second.empty()) {
			// now just find first one.
			const TopicDest &dest = *(pos->second.begin());
			Node dest_node(dest.weak_node_.lock());
			if (!dest_node) {
			service_map_.erase(pos);
			return MakeReply<Reply>(eOffline, "topic server offline.");
			} else if (!Valid(*dest_node)) {
			return MakeReply<Reply>(eNoRespond, "topic server not responding.");
			} else {
			MsgQueryTopicReply reply = MakeReply<Reply>(eSuccess);
			reply.mutable_address()->set_mq_id(dest.mq_);
			return reply;
			auto &clients = pos->second;
			Reply reply = MakeReply<Reply>(eSuccess);
			for (auto &dest : clients) {
			Node dest_node(dest.weak_node_.lock());
			if (dest_node && Valid(*dest_node)) {
			auto node_addr = reply.add_node_address();
			node_addr->set_proc_id(dest_node->proc_.proc_id());
			node_addr->mutable_addr()->set_mq_id(dest.mq_);
			}
			}

			return reply;
			} else {
			return MakeReply<Reply>(eNotFound, "topic server not found.");
			}
			@@ -231,7 +240,7 @@
			MsgCommonReply Subscribe(const BHMsgHead &head, const MsgSubscribe &msg)
			{
			return HandleMsg(head, [&](Node node) {
			auto &src = SrcAddr(head);
			auto src = SrcAddr(head);
			auto &topics = msg.topics().topic_list();
			node->subscriptions_[src].insert(topics.begin(), topics.end());
			TopicDest dest = {src, node};
			@@ -244,7 +253,7 @@
			MsgCommonReply Unsubscribe(const BHMsgHead &head, const MsgUnsubscribe &msg)
			{
			return HandleMsg(head, [&](Node node) {
			auto &src = SrcAddr(head);
			auto src = SrcAddr(head);
			auto pos = node->subscriptions_.find(src);

			auto RemoveSubTopicDestRecord = [this](const Topic &topic, const TopicDest &dest) {
			@@ -417,8 +426,8 @@
			auto MakeReplyer = [](ShmSocket &socket, BHMsgHead &head, const std::string &proc_id) {
			return [&](auto &&rep_body) {
			auto reply_head(InitMsgHead(GetType(rep_body), proc_id, head.msg_id()));
			auto &remote = head.route(0).mq_id();
			socket.Send(remote.data(), reply_head, rep_body);
			auto remote = head.route(0).mq_id();
			socket.Send(remote, reply_head, rep_body);
			};
			};

			@@ -433,6 +442,7 @@
			switch (head.type()) {
			CASE_ON_MSG_TYPE(Register);
			CASE_ON_MSG_TYPE(Heartbeat);
			CASE_ON_MSG_TYPE(Unregister);

			CASE_ON_MSG_TYPE(RegisterRPC);
			CASE_ON_MSG_TYPE(QueryTopic);
			@@ -463,7 +473,7 @@
			if (node) {
			// should also make sure that mq is not killed before msg expires.
			// it would be ok if (kill_time - offline_time) is longer than expire time.
			socket.Send(cli.mq_.data(), msg);
			socket.Send(cli.mq_, msg);
			++it;
			} else {
			it = clients.erase(it);
			@@ -495,28 +505,24 @@
			return rec;
			}

			bool BHCenter::Install(const std::string &name, MsgHandler handler, IdleHandler idle, const std::string &mqid, const int mq_len)
			bool BHCenter::Install(const std::string &name, MsgHandler handler, IdleHandler idle, const MQId mqid, const int mq_len)
			{
			Centers()[name] = CenterInfo{name, handler, idle, mqid, mq_len};
			return true;
			}
			bool BHCenter::Install(const std::string &name, MsgHandler handler, IdleHandler idle, const MQId &mqid, const int mq_len)
			{
			return Install(name, handler, idle, std::string((const char *) &mqid, sizeof(mqid)), mq_len);
			}

			BHCenter::BHCenter(Socket::Shm &shm)
			{
			auto gc = [&](const std::string &id) {
			auto r = ShmSocket::Remove(shm, (MQId ) id.data());
			printf("remove mq : %s\n", r ? "ok" : "failed");
			auto gc = [&](const MQId id) {
			auto r = ShmSocket::Remove(shm, id);
			printf("remove mq %ld : %s\n", id, (r ? "ok" : "failed"));
			};

			AddCenter("#bhome_center", gc);

			for (auto &kv : Centers()) {
			auto &info = kv.second;
			sockets_[info.name_] = std::make_shared<ShmSocket>(shm, (MQId ) info.mqid_.data(), info.mq_len_);
			sockets_[info.name_] = std::make_shared<ShmSocket>(shm, info.mqid_, info.mq_len_);
			}
			}