| 18 | |
| 19 | |
| 20 | comm::RetCode GetSubIDsByConsumerAddr(const int topic_id, const comm::proto::Addr &addr, std::set<int> &sub_ids) { |
| 21 | sub_ids.clear(); |
| 22 | |
| 23 | comm::RetCode ret; |
| 24 | |
| 25 | shared_ptr<const config::ConsumerConfig> consumer_config; |
| 26 | if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetConsumerConfig(topic_id, consumer_config))) { |
| 27 | NLErr("GetConsumerConfig ret %d", comm::as_integer(ret)); |
| 28 | return ret; |
| 29 | } |
| 30 | |
| 31 | shared_ptr<const config::proto::Consumer> consumer; |
| 32 | if (comm::RetCode::RET_OK != (ret = consumer_config->GetConsumerByAddr(addr, consumer))) { |
| 33 | NLErr("GetConsumerByAddr ret %d addr(%s:%d)", comm::as_integer(ret), addr.ip().c_str(), addr.port()); |
| 34 | return ret; |
| 35 | } |
| 36 | |
| 37 | shared_ptr<const config::TopicConfig> topic_config; |
| 38 | if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicConfigByTopicID(topic_id, topic_config))) { |
| 39 | NLErr("GetTopicConfigByTopicID ret %d", comm::as_integer(ret)); |
| 40 | return ret; |
| 41 | } |
| 42 | |
| 43 | set<int> ignore_sub_ids; |
| 44 | for (int i{0}; i < topic_config->GetProto().topic().consumer_ignore_sub_ids_size(); ++i) { |
| 45 | ignore_sub_ids.insert(topic_config->GetProto().topic().consumer_ignore_sub_ids(i)); |
| 46 | } |
| 47 | |
| 48 | if (consumer->sub_ids_size()) { |
| 49 | for (int i{0}; i < consumer->sub_ids_size(); ++i) { |
| 50 | auto sub_id = consumer->sub_ids(i); |
| 51 | if (topic_config->IsValidSubID(sub_id) && ignore_sub_ids.end() == ignore_sub_ids.find(sub_id)) { |
| 52 | sub_ids.insert(sub_id); |
| 53 | } |
| 54 | } |
| 55 | return comm::RetCode::RET_OK; |
| 56 | } |
| 57 | |
| 58 | if (comm::RetCode::RET_OK != (ret = topic_config->GetAllSubID(sub_ids))) { |
| 59 | NLErr("GetAllSubID ret %d", comm::as_integer(ret)); |
| 60 | return ret; |
| 61 | } |
| 62 | |
| 63 | for (auto &&ignore_sub_id : ignore_sub_ids) { |
| 64 | sub_ids.erase(ignore_sub_id); |
| 65 | } |
| 66 | |
| 67 | return comm::RetCode::RET_OK; |
| 68 | } |
| 69 | |
| 70 | |
| 71 | } // namespace utils |
no test coverage detected