Method Replay

phxqueue/store/keepsyncthread.cpp:413–575

}

void KeepSyncThread::Replay() {
    comm::RetCode ret;
    const int topic_id{impl_->store->GetTopicID()};
    auto opt(impl_->store->GetStoreOption());

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

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

    vector<unique_ptr<config::proto::ReplayInfo>> replay_infos;
    if (comm::RetCode::RET_OK != (ret = topic_config->GetAllReplayInfo(replay_infos))) {
        QLErr("GetAllReplayInfo ret %d", as_integer(ret));
        return;
    }

    if (!replay_infos.size()) return;

    QLInfo("Do Replay!!!");

    vector<consumer::Queue_t> queues;
    GetAllLocalQueue(queues);

    for (auto &&queue : queues) {
        int replay_infos_idx{-1};
        bool replay = QueueNeedReplay(queue, pub_ids, replay_infos, replay_infos_idx);

        if (replay && -1 != replay_infos_idx) {  // need replay