Method GetItemsByCursorID

phxqueue/store/basemgr.cpp:397–443

comm::RetCode BaseMgr::GetItemsByCursorID(const int queue_id, const uint64_t cursor_id,
                                          vector<comm::proto::QItem> &items) {
    items.clear();

    const int paxos_group_id(queue_id % impl_->store->GetStoreOption()->ngroup);


    std::vector<std::pair<std::string, int>> values;
    int paxos_ret = impl_->store->GetNode()->GetInstanceValue(paxos_group_id, cursor_id, values);
    if (phxpaxos::Paxos_GetInstanceValue_Value_Not_Chosen_Yet == paxos_ret) {
        return comm::RetCode::RET_ERR_PAXOS_NOT_CHOSEN;
    } else if (phxpaxos::Paxos_GetInstanceValue_Value_NotExist == paxos_ret) {
        QLWarn("GetInstanceValue not exist paxos_group_id %d cursor_id %" PRIu64,
               paxos_group_id, cursor_id);
        return comm::RetCode::RET_OK;
    } else if (0 != paxos_ret) {
        QLErr("GetInstanceValue paxos_ret %d paxos_group_id %d cursor_id %" PRIu64,
              paxos_ret, paxos_group_id, cursor_id);
        return comm::RetCode::RET_ERR_PAXOS_GET_INSTANCE_VALUE;
    }

    for (auto &&value : values) {
        QLVerb("value.length() %zu", value.first.length());
        if (value.second != StoreSM::ID) continue;

        proto::StorePaxosArgs args;
        if (!args.ParseFromString(value.first)) {
            QLErr("ParseFromString fail");
            return comm::RetCode::RET_ERR_PAXOS_VALUE_PARSE;
        }

        if (!args.add_req().items_size()) continue;
        if (queue_id != args.add_req().queue_id()) continue;

        for (size_t i{0}; i < args.add_req().items_size(); ++i) {
            items.push_back(args.add_req().items(i));
            items.back().set_cursor_id(cursor_id);
            QLVerb("add item. topic_id %d uin %" PRIu64,
                   args.add_req().items(i).meta().topic_id(),
                   args.add_req().items(i).meta().uin());
            QLVerb("back item. topic_id %d uin %" PRIu64,
                   items.back().meta().topic_id(), items.back().meta().uin());
        }
    }

    return comm::RetCode::RET_OK;
}

StoreMetaQueue *BaseMgr::GetMetaQueue(const int queue_id) {
    if (queue_id >= impl_->store->GetStoreOption()->nqueue) {