Method Run

phxqueue/consumer/consumer.cpp:109–156

}

void Consumer::Run() {
    comm::RetCode ret;

    if (impl_->opt.log_func) {
        comm::Logger::GetInstance()->SetLogFunc(impl_->opt.log_func);
    }

    if (impl_->opt.config_factory_create_func) {
        plugin::ConfigFactory::SetConfigFactoryCreateFunc(impl_->opt.config_factory_create_func);
    }

    if (impl_->opt.break_point_factory_create_func) {
        plugin::BreakPointFactory::SetBreakPointFactoryCreateFunc(impl_->opt.break_point_factory_create_func);
    }

    if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicIDByTopicName(impl_->opt.topic, impl_->topic_id))) {
        QLErr("GetTopicIDByTopicName ret %d topic %s", comm::as_integer(ret), impl_->opt.topic.c_str());
        return;
    }

    const int shm_key{impl_->opt.shm_key_base + impl_->topic_id};
    const string lock_path{impl_->opt.lock_path_base + to_string(impl_->topic_id)};

    if (comm::RetCode::RET_OK != (ret = impl_->lock.Init(this, shm_key, lock_path, impl_->opt.nprocs))) {
        QLErr("HeartBeatLock::Init ret %d", comm::as_integer(ret));
        return;
    }

    if (comm::RetCode::RET_OK != (ret = impl_->freq.Init(impl_->topic_id, this))) {
        QLErr("FreqMan::Init ret %d", comm::as_integer(ret));
        return;
    }

    BeforeFork();


    QLInfo("INFO: run sync");
    impl_->lock.RunSync();


    QLInfo("INFO: run freqman");
    impl_->freq.Run();

    QLInfo("INFO: procs %d", impl_->opt.nprocs);
    ForkAndRun(impl_->opt.nprocs);

    QLErr("consumer stop");
}

int Consumer::GetTopicID() {
    return impl_->topic_id;