| | |
| | | */ |
| | | #include "topic_node.h" |
| | | #include "bh_util.h" |
| | | #include "failed_msg.h" |
| | | #include <chrono> |
| | | #include <list> |
| | | |
| | |
| | | std::string msg_id; |
| | | }; |
| | | |
| | | class ServerFailedQ |
| | | { |
| | | struct FailedMsg { |
| | | steady_clock::time_point xpr; |
| | | std::string remote_; |
| | | BHMsgHead head_; |
| | | MsgRequestTopicReply body_; |
| | | FailedMsg(const std::string &addr, BHMsgHead &&head, MsgRequestTopicReply &&body) : |
| | | xpr(steady_clock::now() + 10s), remote_(addr), head_(std::move(head)), body_(std::move(body)) {} |
| | | bool Expired() { return steady_clock::now() > xpr; } |
| | | }; |
| | | typedef std::list<FailedMsg> Queue; |
| | | Synced<Queue> queue_; |
| | | |
| | | public: |
| | | void Push(const std::string &remote, BHMsgHead &&head, MsgRequestTopicReply &&body) |
| | | { |
| | | queue_->emplace_back(remote, std::move(head), std::move(body)); |
| | | } |
| | | void TrySend(ShmSocket &socket, const int timeout_ms = 0) |
| | | { |
| | | queue_.Apply([&](Queue &q) { |
| | | if (!q.empty()) { |
| | | auto it = q.begin(); |
| | | do { |
| | | if (it->Expired()) { |
| | | // it->msg_.Release(socket.shm()); |
| | | it = q.erase(it); |
| | | } else if (socket.Send(it->remote_.data(), it->head_, it->body_, timeout_ms)) { |
| | | it = q.erase(it); |
| | | } else { |
| | | ++it; |
| | | } |
| | | } while (it != q.end()); |
| | | } |
| | | }); |
| | | } |
| | | }; |
| | | typedef FailedMsgQ ServerFailedQ; |
| | | |
| | | } // namespace |
| | | TopicNode::TopicNode(SharedMemory &shm) : |
| | |
| | | for (int i = 0; i < head.route_size() - 1; ++i) { |
| | | reply_head.add_route()->Swap(head.mutable_route(i)); |
| | | } |
| | | if (!sock.Send(head.route().rbegin()->mq_id().data(), reply_head, reply_body, 10)) { |
| | | failed_q->Push(head.route().rbegin()->mq_id(), std::move(reply_head), std::move(reply_body)); |
| | | MsgI msg; |
| | | if (msg.Make(sock.shm(), reply_head, reply_body)) { |
| | | auto &remote = head.route().rbegin()->mq_id(); |
| | | if (!sock.Send(remote.data(), msg, 10)) { |
| | | failed_q->Push(remote, msg, 10s); |
| | | } |
| | | } |
| | | } |
| | | } |