Method ChildRun

phxqueue/consumer/consumer.cpp:480–537

void Consumer::ChildRun(const int vpid) {
    QLInfo("INFO: start vpid %u", vpid);

    impl_->vpid = vpid;

    comm::RetCode ret;

    OnChildRun(vpid);

    comm::proto::ConsumerContext &cc = impl_->cc;
    cc.set_topic_id(GetTopicID());

    comm::ConsumerBP::GetThreadInstance()->OnChildRun(cc.topic_id());

    thread t(bind(&Consumer::ConsumeThreadRun, this, vpid));

    while (true) {

        BeforeLock(cc);

        comm::ConsumerBP::GetThreadInstance()->OnLock(cc.topic_id());

        int sub_id, store_id, queue_id;
        if (comm::RetCode::RET_OK != (ret = impl_->lock.Lock(vpid, sub_id, store_id, queue_id))) {
            if (comm::as_integer(ret) < 0) QLErr("DoLock ret %d vpid %u", comm::as_integer(ret), vpid);
            sleep(2);
            continue;
        }
        if (cc.sub_id() != sub_id ||
            cc.store_id() != store_id ||
            cc.queue_id() != queue_id) {
            cc.set_sub_id(sub_id);
            cc.set_store_id(store_id);
            cc.set_queue_id(queue_id);
            cc.set_prev_cursor_id(-1);
            cc.set_next_cursor_id(-1);
        }

        comm::ConsumerBP::GetThreadInstance()->OnLockSucc(cc);

        QLInfo("QUEUEINFO: vpid %u sub_id %d store_id %d queue_id %d", vpid, cc.sub_id(), cc.store_id(), cc.queue_id());

        AfterLock(cc);

        CheckMaxLoop(vpid);
        CheckMem(vpid);

        comm::ConsumerBP::GetThreadInstance()->OnProcess(cc);

        if (comm::RetCode::RET_OK != (ret = Process(cc))) {
            comm::ConsumerBP::GetThreadInstance()->OnProcessFail(cc);
            QLErr("ERR: Process ret %d", comm::as_integer(ret));
            usleep(100000);
        } else {
            comm::ConsumerBP::GetThreadInstance()->OnProcessSucc(cc);
        }
    }
}