Method QueueNeedReplay

phxqueue/store/keepsyncthread.cpp:368–411

}

bool KeepSyncThread::QueueNeedReplay(const consumer::Queue_t &queue, const set<int> &pub_ids,
        const vector<unique_ptr<config::proto::ReplayInfo>> &replay_infos, int &replay_infos_idx) {
    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 topic_id %d", ret, topic_id);
        return false;
    }

    replay_infos_idx = -1;
    for (int idx{0}; idx < replay_infos.size(); ++idx) {
        auto &&replay_info = replay_infos[idx];
        if (!replay_info) continue;
        if (replay_info->sub_ids_size()) {
            bool found = false;
            for (int i{0}; i < replay_info->sub_ids_size(); ++i) {
                auto sub_id = replay_info->sub_ids(i);
                if (sub_id == queue.sub_id) {
                    found = true;
                    break;
                }
            }
            if (!found) continue;
        }
        if (replay_info->pub_ids_size()) {
            bool found = false;
            for (int i{0}; i < replay_info->pub_ids_size(); ++i) {
                auto pub_id = replay_info->pub_ids(i);
                if (pub_ids.end() != pub_ids.find(pub_id) && topic_config->IsValidQueue(queue.queue_id, pub_id)) {
                    found = true;
                    break;
                }
            }
            if (!found) continue;
        }
        replay_infos_idx = idx;
        return true;
    }
    return false;
}

void KeepSyncThread::Replay() {
    comm::RetCode ret;