Method GetAllLocalQueue

phxqueue/store/keepsyncthread.cpp:622–686

}

void KeepSyncThread::GetAllLocalQueue(vector<consumer::Queue_t> &queues) {
    queues.clear();

    comm::RetCode ret;

    const int topic_id = impl_->store->GetTopicID();

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

    auto opt = impl_->store->GetStoreOption();
    comm::proto::Addr addr;
    addr.set_ip(opt->ip);
    addr.set_port(opt->port);
    addr.set_paxos_port(opt->paxos_port);

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

    int store_id;
    if (comm::RetCode::RET_OK != (ret = store_config->GetStoreIDByAddr(addr, store_id))) {
        QLErr("GetStoreIDByAddr ret %d", as_integer(ret));
        return;
    }

    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);
        return;
    }

    for (int queue_id{0}; queue_id < opt->nqueue; ++queue_id) {
        for (int sub_id{1}; sub_id <= opt->nsub; ++sub_id) {
            bool is_valid = false;
            for (auto &&pub_id : pub_ids) {
                if (topic_config->IsValidQueue(queue_id, pub_id, sub_id)) {
                    is_valid = true;
                    break;
                }
            }
            if (!is_valid) continue;

            {
                consumer::Queue_t queue;
                queue.sub_id = sub_id;
                queue.store_id = store_id;