Open bitliubei2 opened 2 years ago
ZMQReceiver::ZMQReceiver() {}
ZMQReceiver::~ZMQReceiver() { Stop(); }
void ZMQReceiver::Init(const std::string& ip, const std::string& port) { auto ipport = ip + ":" + port; zmqctx = zmq::context_t(1); zmq_socketsubscriber = zmq::socket_t(zmqctx, zmq::socket_type::sub); zmq_socketsubscriber.setsockopt(ZMQ_RCVBUF, 20 1024 1024);
zmq_socketsubscriber.setsockopt(ZMQ_RCVTIMEO, 1000); zmq_socketsubscriber.connect(ipport); zmq_socketsubscriber.set(zmq::sockopt::subscribe, ""); isinit = true; }
void ZMQReceiver::Start() { zmqthread.reset(new std::thread([this] { UpdateThreadFunc(); })); if (zmqthread == nullptr) { return; } isrunning = true; }
void ZMQReceiver::Stop() { if (isinit) { if (isrunning && zmqthread != nullptr && zmqthread->joinable()) { isrunning = false; zmqthread->join(); zmqthread.reset(); } zmq_socketsubscriber.close(); zmq_ctx_term(&zmqctx); } }
void ZMQReceiver::UpdateThreadFunc() { while (isrunning) { std::vector recv_msgs; zmq::recv_result_t result = zmq::recv_multipart( zmq_socketsubscriber, std::back_inserter(recv_msgs)); std::vector msg; for (auto& one : recv_msgs) { msg.push_back(one.to_string()); } if (recv_msgs.size() > 0) { Process(msg); } std::this_thread::yield(); } }
void ZMQReceiver::Process(const std::vector& msg) {
}
ZMQReceiver::ZMQReceiver() {}
ZMQReceiver::~ZMQReceiver() { Stop(); }
void ZMQReceiver::Init(const std::string& ip, const std::string& port) { auto ipport = ip + ":" + port; zmqctx = zmq::context_t(1); zmq_socketsubscriber = zmq::socket_t(zmqctx, zmq::socket_type::sub); zmq_socketsubscriber.setsockopt(ZMQ_RCVBUF, 20 1024 1024);
zmq_socketsubscriber.setsockopt(ZMQ_RCVTIMEO, 1000); zmq_socketsubscriber.connect(ipport); zmq_socketsubscriber.set(zmq::sockopt::subscribe, ""); isinit = true; }
void ZMQReceiver::Start() { zmqthread.reset(new std::thread([this] { UpdateThreadFunc(); })); if (zmqthread == nullptr) { return; } isrunning = true; }
void ZMQReceiver::Stop() { if (isinit) { if (isrunning && zmqthread != nullptr && zmqthread->joinable()) { isrunning = false; zmqthread->join(); zmqthread.reset(); } zmq_socketsubscriber.close(); zmq_ctx_term(&zmqctx); } }
void ZMQReceiver::UpdateThreadFunc() { while (isrunning) { std::vector recv_msgs;
zmq::recv_result_t result = zmq::recv_multipart(
zmq_socketsubscriber, std::back_inserter(recv_msgs));
std::vector msg;
for (auto& one : recv_msgs) {
msg.push_back(one.to_string());
}
if (recv_msgs.size() > 0) {
Process(msg);
}
std::this_thread::yield();
}
}
void ZMQReceiver::Process(const std::vector& msg) {
}