| | |
| | | |
| | | #include <vector> |
| | | #include "common.h" |
| | | using namespace std; |
| | | |
| | | #include <nng/protocol/reqrep0/rep.h> |
| | | #include <nng/supplemental/util/platform.h> |
| | |
| | | #include "nng/compat/nanomsg/reqrep.h" |
| | | #include "nng/compat/nanomsg/pubsub.h" |
| | | #include "nng/compat/nanomsg/survey.h" |
| | | |
| | | using namespace std; |
| | | |
| | | namespace nng_wrap { |
| | | |
| | |
| | | sub->socket_ = sock; |
| | | sub->t_ = get_thread([](const auto sub){ |
| | | while (!sub->t_quit_.load()) { |
| | | // for(auto&& i : sub->topics_) { |
| | | // printf("======>> sub topic %s\n", i.c_str()); |
| | | // } |
| | | char* m{}; |
| | | // int m_len = nn_recv(sub->socket_, &m, NN_MSG, NN_DONTWAIT); |
| | | int m_len = nn_recv(sub->socket_, &m, NN_MSG, 0); |
| | |
| | | } |
| | | } |
| | | } |
| | | // printf("======>> subscribe recv topic %s msg length %lu\n", topic, msg.length()); |
| | | // printf("======>> subscribe recv topic %s msg length %lu\n", topic, msgl); |
| | | if (found_topic){ |
| | | lock_guard<mutex> l(sub->mtx_msg_); |
| | | sub->msg_.emplace_back(string(topic), string(msg, msgl)); |
| | |
| | | } |
| | | nn_freemsg(m); |
| | | }else { |
| | | if (!sub->failed_topics_.empty()) |
| | | { |
| | | lock_guard<mutex> l{sub->mtx_failed_topics_}; |
| | | if (!sub->failed_topics_.empty()){ |
| | |
| | | } |
| | | } |
| | | } |
| | | this_thread::sleep_for(chrono::milliseconds{6}); |
| | | // this_thread::sleep_for(chrono::milliseconds{6}); |
| | | // printf("======>> subscribe nn_recv failed %s\n", nn_strerror(nn_errno())); |
| | | } |
| | | } |
| | |
| | | |
| | | if (!rep->t_unblock_){ |
| | | rep->t_unblock_.reset(new thread(get_thread([](const auto rep){ |
| | | constexpr int idle = 10; |
| | | constexpr int idle = 216; |
| | | const auto data = rr_unblocking_msg_.data(); |
| | | const auto data_size = rr_unblocking_msg_.size(); |
| | | constexpr int life_span = timeout_req_rep*10; |
| | |
| | | return tmp; |
| | | }; |
| | | while (!rep->t_quit_.load()) { |
| | | this_thread::sleep_for(chrono::milliseconds{10}); |
| | | this_thread::sleep_for(chrono::milliseconds{idle}); |
| | | vector<struct work*> tmp = f(); |
| | | for(auto && w : tmp){ |
| | | aio_unblock(w, data, data_size); |