Method Handle

phxqueue/consumer/consumer.cpp:878–921

}

comm::RetCode Consumer::Handle(const comm::proto::ConsumerContext &cc,
                               comm::proto::QItem &item, comm::HandleResult &handle_result) {

    comm::ConsumerConsumeBP::GetThreadInstance()->OnHandle(impl_->cc, item);

    comm::RetCode ret;

    RestoreUserCookies(item.meta().user_cookies());

    std::string uncompressed_buffer;
    if (comm::RetCode::RET_OK != (ret = UncompressBuffer(item.buffer(), item.buffer_type(), uncompressed_buffer))) {
        QLErr("DecodeBuffer ret %d", comm::as_integer(ret));
        handle_result = comm::HandleResult::RES_ERROR;
        return ret;
    }

    auto &&handler = GetHandler(item.meta().handle_id());
    if (!handler) {
        QLErr("GetHandler return null handle_id %d", item.meta().handle_id());
        handle_result = comm::HandleResult::RES_ERROR;
        return comm::RetCode::RET_ERR_RANGE_HANDLE;
    }

    handle_result = handler->Handle(cc, item, uncompressed_buffer);
    QLVerb("handle_result %d handle->IsBufferUpdated() %d", static_cast<int>(handle_result), handler->IsBufferUpdated());

    if (handler->IsBufferUpdated()) {
        CompressBuffer(uncompressed_buffer, *item.mutable_buffer(), item.buffer_type());
    }

    comm::ConsumerConsumeBP::GetThreadInstance()->OnHandleEnd(impl_->cc, item, handle_result);

    auto &&client_id = item.meta().client_id();
    if (client_id.empty()) {
        QLVerb("Handle handle_id %d ori_pub_id %d pub_id %d sub_ids %" PRIu64 " hash %" PRIu64 " uin %" PRIu64 " handle_result %d",
               item.meta().handle_id(), item.meta().pub_id(), item.pub_id(), (uint64_t)item.sub_ids(), (uint64_t)item.meta().hash(), (uint64_t)item.meta().uin(), handle_result);
    } else {
        QLInfo("Handle handle_id %d ori_pub_id %d pub_id %d sub_ids %" PRIu64 " hash %" PRIu64 " uin %" PRIu64 " handle_result %d client_id %s",
               item.meta().handle_id(), item.meta().pub_id(), item.pub_id(), (uint64_t)item.sub_ids(), (uint64_t)item.meta().hash(), (uint64_t)item.meta().uin(), handle_result,
               client_id.c_str());
    }

    return comm::RetCode::RET_OK;
}

void Consumer::CheckMaxLoop(const int vpid) {
    comm::RetCode ret;