Method MakeAddRequests

phxqueue/producer/producer.cpp:182–298

comm::RetCode Producer::MakeAddRequests(const int topic_id,
                                        const std::vector<std::shared_ptr<comm::proto::QItem> > &items,
                                        vector<unique_ptr<comm::proto::AddRequest> > &reqs,
                                        ItemUpdateFunc item_update_func) {

    reqs.clear();

    comm::ProducerBP::GetThreadInstance()->OnMakeAddRequests(topic_id, items);

    comm::RetCode ret;

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

    size_t byte_size_limit = topic_config->GetProto().topic().items_byte_size_limit();
    int batch_limit = topic_config->GetProto().topic().batch_limit();


    map<int, vector<unique_ptr<comm::proto::QItem> > > queue_info_id2new_items;
    for (auto &&item : items) {
        if (nullptr == item) continue;

        auto new_item = unique_ptr<comm::proto::QItem>(new comm::proto::QItem());
        new_item->CopyFrom(*item);
        if (item_update_func) item_update_func(*new_item);

        int queue_info_id = 0;
        if (comm::RetCode::RET_OK != (ret = topic_config->GetQueueInfoIDByCount(new_item->pub_id(), new_item->count(), queue_info_id))) {
            if (comm::RetCode::RET_ERR_RANGE_CNT == ret) {
                comm::ProducerBP::GetThreadInstance()->OnCountLimit(topic_id, new_item->pub_id(), *item);
                NLInfo("skip. GetQueueInfoIDByCount ret %d count %d "
                       "handle_id %d ori_pub_id %d pub_id %d sub_ids %" PRIu64 " hash %" PRIu64 " uin %" PRIu64,
                       as_integer(ret), new_item->count(),
                       new_item->meta().handle_id(), new_item->meta().pub_id(), new_item->pub_id(), (uint64_t)new_item->sub_ids(),
                       (uint64_t)new_item->meta().hash(), (uint64_t)new_item->meta().uin());
            } else {
                NLErr("GetQueueInfoIDByCount ret %d count %d "
                      "handle_id %d ori_pub_id %d pub_id %d sub_ids %" PRIu64 " hash %" PRIu64 " uin %" PRIu64,
                      as_integer(ret), new_item->count(),
                      new_item->meta().handle_id(), new_item->meta().pub_id(), new_item->pub_id(), (uint64_t)new_item->sub_ids(),
                      (uint64_t)new_item->meta().hash(), (uint64_t)new_item->meta().uin());
            }
            continue;
        }

        queue_info_id2new_items[queue_info_id].push_back(move(new_item));
    }


    for (auto &&kv : queue_info_id2new_items) {
        auto &&new_items = kv.second;

        unique_ptr<comm::proto::AddRequest> req = nullptr;
        int batch = 0;