Method Consume

phxqueue/consumer/consumer.cpp:840–871

}

comm::RetCode Consumer::Consume(const comm::proto::ConsumerContext &cc,
                                const vector<shared_ptr<comm::proto::QItem> > &items,
                                vector<comm::HandleResult> &handle_results) {
    comm::ConsumerBP::GetThreadInstance()->OnConsume(cc, items);

    handle_results.clear();
    for (int i{0}; i < items.size(); ++i) {
        handle_results.push_back(comm::HandleResult::RES_ERROR);
    }

    {
        int iret = 0;

        while (sizeof(char) != (iret = write(impl_->consume_fds[0], "a", sizeof(char)))) {
            QLErr("write ret %d err %s", iret, strerror(errno));
            poll(nullptr, 0, 100);
        }


        char ch;
        while (sizeof(char) != (iret = read(impl_->consume_fds[0], &ch, sizeof(char)))) {
            QLErr("read ret %d err %s", iret, strerror(errno));
            poll(nullptr, 0, 100);
        }
    }

    comm::ConsumerBP::GetThreadInstance()->OnConsumeSucc(cc, items, handle_results);

    impl_->freq.UpdateConsumeStat(impl_->vpid, cc, items);

    return comm::RetCode::RET_OK;
}

bool Consumer::SkipHandle(const comm::proto::ConsumerContext &cc, const comm::proto::QItem &item) {
    if (!(item.sub_ids() & (1ULL << (cc.sub_id() - 1)))) return true;