Method AfterConsume

phxqueue/consumer/consumer.cpp:1000–1071

}

void Consumer::AfterConsume(const comm::proto::ConsumerContext &cc,
                            const std::vector<std::shared_ptr<comm::proto::QItem> > &items,
                            const std::vector<comm::HandleResult> &handle_results) {

    phxqueue::comm::RetCode ret;

    if (handle_results.size() < items.size()) {
        QLErr("handle_results.size() %d < items.size() %d", (int)handle_results.size(), (int)items.size());
        return;
    }

    vector<shared_ptr<phxqueue::comm::proto::QItem> > retry_items;
    for (int i{0}; i < items.size(); ++i) {
        auto &&item = items[i];
        auto &&handle_result = handle_results[i];
        if (phxqueue::comm::HandleResult::RES_ERROR == handle_result) {
            retry_items.push_back(item);
        }
    }

    if (0 == retry_items.size()) return;

    vector<unique_ptr<phxqueue::comm::proto::AddRequest> > reqs;

    uint64_t retry_sub_ids = (1ULL << (cc.sub_id() - 1));
    {
        ret = phxqueue::producer::Producer::MakeAddRequests(cc.topic_id(), retry_items, reqs,
                                                                                        [&cc, retry_sub_ids](phxqueue::comm::proto::QItem &item)->void {
                                                                                            item.set_count(item.count() + 1);
                                                                                            item.set_sub_ids(retry_sub_ids);

                                                                                            auto now = comm::utils::Time::GetTimestampMS();
                                                                                            item.set_atime(now / 1000);
                                                                                            item.set_atime_ms(now % 1000);
                                                                                        });
        if (phxqueue::comm::RetCode::RET_OK != ret) {
            QLErr("MakeAddRequests ret %d", as_integer(ret));
            return;
        }
    }

    for (auto &&req : reqs) {
        if (!req || 0 == req->items_size()) continue;

        int retry_pub_id = req->items(0).pub_id();

        BeforeAdd(*req);

        comm::proto::AddResponse resp;
        while (true) { // retry forever
            if (impl_->opt.use_store_master_client_on_add) {
                store::StoreMasterClient<comm::proto::AddRequest, comm::proto::AddResponse> store_master_client;
                ret = store_master_client.ClientCall(*req, resp, bind(&Consumer::Add, this, placeholders::_1, placeholders::_2));
            } else {
                ret = Add(*req, resp);
            }
            if (comm::RetCode::RET_OK != ret) {
                QLErr("Retry ret %d topic_id %d retry_pub_id %d sub_id %d store_id %d queue_id %d item_size %zu retry_item_size %d",