Method SelectAndAdd

phxqueue/producer/producer.cpp:301–376

comm::RetCode Producer::SelectAndAdd(comm::proto::AddRequest &req, comm::proto::AddResponse &resp, StoreSelector *ss, QueueSelector *qs) {
    QLVerb("SelectAndAdd");

    comm::ProducerBP::GetThreadInstance()->OnSelectAndAdd(req);

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

    auto pub_id = req.items(0).pub_id();
    auto uin = req.items(0).meta().uin();
    auto count = req.items(0).count();

    comm::RetCode ret;

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

    std::unique_ptr<QueueSelector> default_qs;
    if (!qs) {
        default_qs = NewQueueSelector(req.topic_id(), pub_id, uin, count, topic_config->GetProto().topic().producer_retry_switch_queue());
        qs = default_qs.get();
        comm::ProducerBP::GetThreadInstance()->OnUseDefaultQueueSelector(req);
    } else {
        comm::ProducerBP::GetThreadInstance()->OnUseCustomQueueSelector(req);
    }

    std::unique_ptr<StoreSelector> default_ss;
    if (!ss) {
        default_ss = NewStoreSelector(req.topic_id(), pub_id, uin, topic_config->GetProto().topic().producer_retry_switch_store());
        ss = default_ss.get();
        comm::ProducerBP::GetThreadInstance()->OnUseDefaultStoreSelector(req);
    } else {
        comm::ProducerBP::GetThreadInstance()->OnUseCustomStoreSelector(req);
    }

    int nretry_raw_add{topic_config->GetProto().topic().producer_nretry_raw_add()};
    do {
        int queue_id;
        if (comm::RetCode::RET_OK != (ret = qs->GetQueueID(queue_id))) {
            comm::ProducerBP::GetThreadInstance()->OnGetQueueIDFail(req);
            QLErr("GetQueue ret %d", as_integer(ret));
            return ret;
        }
        req.set_queue_id(queue_id);

        int store_id;
        if (comm::RetCode::RET_OK != (ret = ss->GetStoreID(store_id))) {
            comm::ProducerBP::GetThreadInstance()->OnGetStoreIDFail(req);
            QLErr("GetStore ret %d", as_integer(ret));
            return ret;
        }
        req.set_store_id(store_id);

        if (nullptr != impl_->batch_helper) {
            if (comm::RetCode::RET_OK != (ret = impl_->batch_helper->BatchRawAdd(req))) {