Method GetAddrScale

phxqueue/consumer/hblock.cpp:321–444

}

comm::RetCode HeartBeatLock::GetAddrScale(SubID2AddrScales &sub_id2addr_scales) {
    QLVerb("start");

    sub_id2addr_scales.clear();

    auto &&opt = impl_->consumer->GetConsumerOption();

    AddrScales config_addr_scales, dynamic_addr_scales;

    comm::RetCode ret;

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

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

    std::set<int> sub_ids;
    {
        comm::proto::Addr addr;
        addr.set_ip(opt->ip);
        addr.set_port(opt->port);

        if (comm::RetCode::RET_OK != (ret = config::utils::GetSubIDsByConsumerAddr(topic_id, addr, sub_ids))) {
            QLErr("GetSubIDs ret %d", comm::as_integer(ret));
            return ret;
        }
    }

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

    std::set<int> dynamic_sub_ids;
    for (auto &&sub_id : sub_ids) {
        shared_ptr<const config::proto::Sub> sub;
        if (comm::RetCode::RET_OK != (ret = topic_config->GetSubBySubID(sub_id, sub))) {
            QLErr("GetSubBySubID ret %d", comm::as_integer(ret));
            continue;
        }
        if (sub->use_dynamic_scale()) {
            dynamic_sub_ids.insert(sub_id);
        }
    }
    QLVerb("dynamic sub_ids size %d", dynamic_sub_ids.size());


    if (dynamic_sub_ids.size()) {
        comm::proto::GetAddrScaleRequest req;
        comm::proto::GetAddrScaleResponse resp;
        req.set_topic_id(topic_id);
        auto &&addr = req.mutable_addr();
        addr->set_ip(opt->ip);
        addr->set_port(opt->port);