Method Get

phxqueue/store/store.cpp:537–578

}

comm::RetCode Store::Get(const comm::proto::GetRequest &req, comm::proto::GetResponse &resp) {
    QLVerb("Get");

    comm::StoreBP::GetThreadInstance()->OnGet(req);

    comm::RetCode ret;

    if (comm::RetCode::RET_OK != (ret = CheckGetRequest(req))) {
        comm::StoreBP::GetThreadInstance()->OnGetRequestInvalid(req);
        QLErr("CheckGetRequest ret %d", as_integer(ret));
        return ret;
    }

    int paxos_group_id{req.queue_id() % impl_->opt.ngroup};
    if (comm::RetCode::RET_OK !=
        (ret = CheckMaster(paxos_group_id, *resp.mutable_redirect_addr()))) {
        comm::StoreBP::GetThreadInstance()->OnGetCheckMasterUnpass(req);
        if (as_integer(ret) < 0) {
            QLErr("CheckMaster ret %d paxos_group_id %d", as_integer(ret), paxos_group_id);
        }
        return ret;
    }
    comm::StoreBP::GetThreadInstance()->OnGetCheckMasterPass(req);

    comm::StoreBP::GetThreadInstance()->OnBaseMgrGet(req);
    if (comm::RetCode::RET_OK != (ret = impl_->basemgr->Get(req, resp))) {
        comm::StoreBP::GetThreadInstance()->OnBaseMgrGetFail(req);
        QLErr("basemgr Get ret %d", as_integer(ret));
        return ret;
    }
    comm::StoreBP::GetThreadInstance()->OnBaseMgrGetSucc(req, resp);

    QLVerb("Get Succ. "
           "req: topic_id %d store_id %d queue_id %d limit %d atime %" PRIu64
           " sub_id %d prev_cursor_id %" PRIu64 " next_cursor_id %" PRIu64 ". "
           "resp: items.size() %d prev_cursor_id %" PRIu64 " next_cursor_id %" PRIu64,
           req.topic_id(), req.store_id(), req.queue_id(), req.limit(),
           req.atime(), req.sub_id(), req.prev_cursor_id(), req.next_cursor_id(),
           resp.items_size(), resp.prev_cursor_id(), resp.next_cursor_id());

    return comm::RetCode::RET_OK;
}

bool Store::SkipGet(const comm::proto::QItem &item, const comm::proto::GetRequest &req) {
    comm::RetCode ret;