Method NeedSkipAdd

phxqueue/store/basemgr.cpp:104–154

}

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

    if (0 == req.items_size()) {
        QLInfo("add req skip. item size 0");
        return true;
    }

    const int topic_id = impl_->store->GetTopicID();

    StoreMetaQueue &meta_queue = impl_->meta_queues[req.queue_id()];
    StoreMeta back_meta;
    if (meta_queue.Back(back_meta) && back_meta.GetCursorID() >= cursor_id) {
        // cursor_id already at meta_queue's back mostly in batch propose sistuation.
        return true;
    }

    shared_ptr<const config::TopicConfig> topic_config;
    if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicConfigByTopicID(topic_id, topic_config))) {
        QLErr("GetTopicConfig ret %d", as_integer(ret));
        return true;
    }

    if (!topic_config->IsValidQueue(req.queue_id())) {
        QLErr("IsValidQueue fail. queue_id %d", req.queue_id());
        return true;
    }
/*
    shared_ptr<const config::proto::Pub> pub;
    if (comm::RetCode::RET_OK != (ret = topic_config->GetPubByPubID(req.pub_id(), pub))) {
        QLErr("GetPubByPubID ret %d pub_id %d", as_integer(ret), req.pub_id());
        return true;
    }

    SyncCtrl *sync = impl_->store->GetSyncCtrl();
    for (size_t i{0}; i < pub->sub_ids_size(); ++i) {
        auto &&sub_id(pub->sub_ids(i));

        uint64_t next_cursor_id;
        ret = sync->GetCursorID(sub_id, req.queue_id(), next_cursor_id, false);
        if (0 < as_integer(ret)) {
            QLWarn("GetCursorID ret %d sub_id %d queue_id %d", ret, sub_id, req.queue_id());
            return false;
        } else if (0 == as_integer(ret) && cursor_id <= next_cursor_id) {
            QLInfo("add req skip. cursor_id %" PRIu64 " next_cursor_id %" PRIu64, cursor_id, next_cursor_id);
            return true;
        }
    }
*/
    return false;
}

comm::RetCode BaseMgr::Get(const comm::proto::GetRequest &req, comm::proto::GetResponse &resp) {
    comm::RetCode ret;