Method Add

phxqueue/store/basemgr.cpp:66–102

comm::RetCode BaseMgr::Add(const uint64_t cursor_id, const comm::proto::AddRequest &req) {
    QLVerb("Add. req: topic_id %d store_id %d queue_id %d", req.topic_id(), req.store_id(), req.queue_id());
    comm::RetCode ret;

    //__OssAttrQueue(req.queue());

    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()->OnAdd(req, queue_info->queue_info_id());

    if (0 == req.items_size()) return comm::RetCode::RET_OK;

    if (NeedSkipAdd(cursor_id, req)) {
        comm::StoreBaseMgrBP::GetThreadInstance()->OnAddSkip(req);
        //if (ossid) OssAttrInc(ossid, phxqueue::ossattr::store::ADD_SKIP, req.items_size());
        return comm::RetCode::RET_ADD_SKIP;
    }

    auto &&meta_queue = impl_->meta_queues[req.queue_id()];
    meta_queue.PushBack(std::move(StoreMeta(cursor_id)));

    comm::StoreBaseMgrBP::GetThreadInstance()->OnAddSucc(req, cursor_id);

    return comm::RetCode::RET_OK;
}

bool BaseMgr::NeedSkipAdd(const uint64_t cursor_id, const comm::proto::AddRequest &req) {
    comm::RetCode ret;