| | |
| | | rc = nn_recv(pub->socket_, &tmp, NN_MSG, 0); |
| | | if (rc > 0){ |
| | | nn_freemsg(tmp); |
| | | printf("======>> publish topic %s data length %lu\n", msg->topic_.c_str(), msg->data_.size()); |
| | | // printf("======>> publish topic %s data length %lu\n", msg->topic_.c_str(), msg->data_.size()); |
| | | lock_guard<mutex> l{pub->mtx_msg_}; |
| | | pub->msg_.pop_front(); |
| | | continue; |
| | |
| | | const auto topic{tmp_msg.c_str()}; |
| | | string msg{}; |
| | | { |
| | | lock_guard<mutex> l{(*sub)()}; |
| | | lock_guard<mutex> l{sub->operator()()}; |
| | | for(auto && i : sub->topics_){ |
| | | if (!!!i.compare(topic)){ |
| | | msg = move(tmp_msg.substr(i.size()+1)); |
| | |
| | | } |
| | | } |
| | | } |
| | | printf("======>> subscribe recv topic %s msg length %lu\n", topic, msg.length()); |
| | | // printf("======>> subscribe recv topic %s msg length %lu\n", topic, msg.length()); |
| | | if (!msg.empty()){ |
| | | lock_guard<mutex> l(sub->mtx_msg_); |
| | | sub->msg_.emplace_back(topic, move(msg)); |
| | |
| | | lock_guard<mutex> l{sub->mtx_failed_topics_}; |
| | | if (!sub->failed_topics_.empty()){ |
| | | for(auto iter = sub->failed_topics_.begin(); iter != sub->failed_topics_.end();){ |
| | | if (nn_setsockopt(sub->socket_, NN_SUB, NN_SUB_UNSUBSCRIBE, iter->c_str(), iter->length()) >= 0){ |
| | | if (nn_setsockopt(sub->socket_, NN_SUB, NN_SUB_SUBSCRIBE, iter->c_str(), iter->length()) >= 0){ |
| | | iter = sub->failed_topics_.erase(iter); |
| | | }else{ |
| | | iter++; |
| | |
| | | get<1>(get<1>(rep->socks_)) = port; |
| | | ipc = "tcp://0.0.0.0:" + to_string(port); |
| | | if(create_server(&get<0>(get<1>(rep->socks_)), ipc, 62, rep) != 0) return -1; |
| | | printf("======>> create server for remote port %d\n", port); |
| | | // printf("======>> create server for remote port %d\n", port); |
| | | }else { |
| | | get<0>(get<1>(rep->socks_)).id = numeric_limits<int32_t>::max(); |
| | | } |