Method GetAllQueues

phxqueue/consumer/hblock.cpp:446–524

}

comm::RetCode HeartBeatLock::GetAllQueues(const int sub_id, vector<Queue_t> &all_queues) {
    all_queues.clear();

    comm::RetCode ret = comm::RetCode::RET_OK;

    auto topic_id = impl_->consumer->GetTopicID();

    shared_ptr<const config::TopicConfig> topic_config;
    if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicConfigByTopicID(topic_id, topic_config))) {
        QLErr("ERR: GetTopicConfigByTopicID err. topic_id %u ret %d", topic_id, comm::as_integer(ret));
        return ret;
    }

    shared_ptr<const config::StoreConfig> store_config;
    if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetStoreConfig(topic_id, store_config))) {
        QLErr("ERR: GetStoreConfigByPubID err. topic_id %d ret %d", topic_id, comm::as_integer(ret));
        return ret;
    }

    set<int> store_ids;
    if (comm::RetCode::RET_OK != (ret = store_config->GetAllStoreID(store_ids))) {
        QLErr("ERR: GetAllStoreID ret %d", comm::as_integer(ret));
        return ret;
    }

    set<pair<int, int> > filter;
    for (auto &&store_id : store_ids) {
        set<int> pub_ids;
        if (comm::RetCode::RET_OK != (ret = config::utils::GetPubIDsByStoreID(topic_id, store_id, pub_ids))) {
            QLErr("GetPubIDsByStoreID ret %d topic_id %d store_id %d", as_integer(ret), topic_id, store_id);
            continue;
        }

        for (auto &&pub_id : pub_ids) {
            shared_ptr<const config::proto::Pub> pub;
            if (comm::RetCode::RET_OK != (ret = topic_config->GetPubByPubID(pub_id, pub))) {
                QLErr("ERR: GetPubByPubID err. ret %d", comm::as_integer(ret));
                continue;
            }

            // check pub
            {
                size_t i;
                for (i = 0; i < pub->sub_ids_size(); ++i) {
                    if (sub_id == pub->sub_ids(i)) break;
                }
                if (i == pub->sub_ids_size()) continue;
            }

            for (int i{0}; i < pub->queue_info_ids_size(); ++i) {
                auto &&queue_info_id = pub->queue_info_ids(i);

                set<int> queue_ids;
                if (comm::RetCode::RET_OK != (ret = topic_config->GetQueuesByQueueInfoID(queue_info_id, queue_ids))) {
                    QLErr("GetQueuesByQueueInfoID ret %d queue_info_id %d", comm::as_integer(ret), queue_info_id);
                    continue;
                }