Method InitTopicID

phxqueue/scheduler/scheduler.cpp:122–166

}

comm::RetCode Scheduler::InitTopicID() {
    comm::RetCode ret;

    if (!impl_->opt.topic.empty()) {
        if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicIDByTopicName(impl_->opt.topic, impl_->topic_id))) {
            QLErr("GetTopicIDByTopicName ret %d topic %s", as_integer(ret), impl_->opt.topic.c_str());
        }
        return ret;
    }

    QLInfo("no topic name. find toipc_id by addr(%s:%d)", impl_->addr.ip().c_str(), impl_->addr.port());

    set<int> topic_ids;
    if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetAllTopicID(topic_ids))) {
        QLErr("GetAllTopicID ret %d", as_integer(ret));
        return ret;
    }

    for (auto &&topic_id : topic_ids) {

        QLVerb("check topic_id %d", topic_id);

        shared_ptr<const config::SchedulerConfig> scheduler_config;
        if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetSchedulerConfig(topic_id, scheduler_config))) {
            QLErr("GetSchedulerConfig ret %d", as_integer(ret));
            continue;
        }

        shared_ptr<const config::proto::Scheduler> scheduler;
        if (comm::RetCode::RET_OK != (ret = scheduler_config->GetScheduler(scheduler))) {
            QLErr("GetScheduler ret %d", as_integer(ret));
        } else {
            for (int i{0}; i < scheduler->addrs_size(); ++i) {
                auto &&addr = scheduler->addrs(i);
                if (addr.ip() == impl_->addr.ip() && addr.port() == impl_->addr.port()) {
                    impl_->topic_id = topic_id;
                    QLInfo("found toipc_id %d addr (%s:%d)", topic_id, impl_->addr.ip().c_str(), impl_->addr.port());
                    return comm::RetCode::RET_OK;
                }
            }
        }
    }

    return comm::RetCode::RET_ERR_RANGE_ADDR;
}

int Scheduler::GetTopicID() {
    return impl_->topic_id;