Method Get

phxqueue/store/basemgr.cpp:156–393

}

comm::RetCode BaseMgr::Get(const comm::proto::GetRequest &req, comm::proto::GetResponse &resp) {
    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", as_integer(ret), topic_id);
        return ret;
    }

    shared_ptr<const config::proto::QueueInfo> queue_info;
    if (comm::RetCode::RET_OK != (ret = topic_config->GetQueueInfoByQueue(req.queue_id(), queue_info))) {
        QLErr("GetQueueInfoByQueue ret %d", as_integer(ret));
        return ret;
    }

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

    if (req.queue_id() >= impl_->store->GetStoreOption()->nqueue) {
        QLErr("queue_id %d invalid. nqueue %d", req.queue_id(), impl_->store->GetStoreOption()->nqueue);
        return comm::RetCode::RET_ERR_RANGE_QUEUE;
    }

    QLVerb("begin. req.prev_cursor_id %llu req.next_cursor_id %llu", req.prev_cursor_id(), req.next_cursor_id());

    uint64_t cli_prev_cursor_id{req.prev_cursor_id()};
    uint64_t cli_next_cursor_id{req.next_cursor_id()};

    StoreMetaQueue &meta_queue = impl_->meta_queues[req.queue_id()];

    SyncCtrl *sync = impl_->store->GetSyncCtrl();

    if (as_integer(ret = sync->AdjustNextCursorID(req.sub_id(), req.queue_id(), cli_prev_cursor_id, cli_next_cursor_id)) < 0) {
        comm::StoreBaseMgrBP::GetThreadInstance()->OnAdjustNextCursorIDFail(req);
        QLErr("AdjustCur failed sub_id %d queue_id %d", req.sub_id(), req.queue_id());
        return comm::RetCode::RET_ERR_GET_ADJUST_CURSOR_ID_FAIL;
    } else if (as_integer(ret)) {
        comm::StoreBaseMgrBP::GetThreadInstance()->OnCursorIDNotFound(req);
        QLInfo("cursorid not found. ret %d sub_id %d queue_id %d", as_integer(ret), req.sub_id(), req.queue_id());
        cli_prev_cursor_id = cli_next_cursor_id = -1;
    }

    if (cli_prev_cursor_id != req.prev_cursor_id() || cli_next_cursor_id != req.next_cursor_id()) {
        comm::StoreBaseMgrBP::GetThreadInstance()->OnCursorIDChange(req);
        QLInfo("AdjustCur sub_id %d queue_id %d prev_cursor_id %" PRIu64
               " -> %" PRIu64 " next_cursor_id %" PRIu64 " -> %" PRIu64,
               req.sub_id(), req.queue_id(),
               static_cast<uint64_t>(req.prev_cursor_id()), cli_prev_cursor_id,
               static_cast<uint64_t>(req.next_cursor_id()), cli_next_cursor_id);
    }

    // get cp
    uint64_t cp = -1, min_instance_id = -1;
    do {
        const int paxos_group_id(req.queue_id() % impl_->store->GetStoreOption()->ngroup);
        auto stat = impl_->store->GetCheckPointStatMgr()->GetCheckPointStat(paxos_group_id);
        if (!stat) {