f8/test/app.cc

#include "precompile.h"

#include <signal.h>
#include <string.h>
#include <unistd.h>
#include <mutex>
#include <condition_variable>

#include <mysql.h>

#include <a8/redis.h>
#include <a8/timer.h>
#include <a8/uuid.h>
#include <a8/mutable_xobject.h>

#include "framework/cpp/netmsghandler.h"
#include "framework/cpp/msgqueue.h"
#include "framework/cpp/dbpool.h"

#include "app.h"

struct IMMsgNode
{
    unsigned short msgid;
    a8::XParams params;
    IMMsgNode* next = nullptr;
};

const char* const PROJ_LOG_ROOT = "/data/logs/%s/logs";
const char* const PROJ_LOG_FILENAME = "log_$pid_%Y%m%d.log";

static void StopApp(int signo)
{
    printf("Stop App...\n");
    App::Instance()->UnInit();
    exit(0);
}

static void Test()
{
    a8::MutableXObject* conn_info = a8::MutableXObject::NewObject();
    conn_info->SetVal("host", a8::XValue("127.0.0.1"));
    conn_info->SetVal("port", a8::XValue(3306));
    conn_info->SetVal("user", a8::XValue("root"));
    conn_info->SetVal("passwd", a8::XValue("keji178"));
    conn_info->SetVal("database", a8::XValue("gamedb1008_6"));
    f8::DBPool::Instance()->ExecAsyncQuery(
                                           *conn_info,
                                           "SELECT 1 UNION SELECT 2 UNION SELECT 3;",
                                           {},
                                           a8::XParams(),
                                           [] (a8::XParams& param, const f8::DataSet* data_set)
                                           {
                                               for (auto& row : *data_set) {
                                                   printf("%s\n", row.at(0).c_str());
                                               }
                                           },
                                           [] (a8::XParams& param, int error_code, const std::string& error_msg)
                                           {
                                               printf("error_msg:%s\n", error_msg.c_str());
                                           }
                                           ,
                                           rand());
    delete conn_info;
}

void App::Init(int argc, char* argv[])
{
    this->argc = argc;
    this->argv = argv;

    if (!ParseOpt()) {
        terminated = true;
        a8::XPrintf("f8test启动失败,缺少-i参数\n", {});
        return;
    }
    a8::XPrintf("f8test starting instance_id:%d pid:%d\n", {instance_id, getpid()});
    signal(SIGPIPE, SIG_IGN);

    loop_mutex_ = new std::mutex();
    loop_cond_ = new std::condition_variable();
    im_msg_mutex_ = new std::mutex();

    srand(time(nullptr));
    InitLog();
    a8::Timer::Instance()->Init();
    f8::MsgQueue::Instance()->Init();
    f8::DBPool::Instance()->Init();
    f8::DBPool::Instance()->SetThreadNum(10);

    a8::UdpLog::Instance()->Info("f8test starting instance_id:%d pid:%d", {instance_id, getpid()});

    signal(SIGINT, StopApp);

    a8::Timer::Instance()->AddRepeatTimer(1000 * 1,
                                          a8::XParams(),
                                          [] (const a8::XParams& param)
                                          {
                                              #if 1
                                              Test();
                                              #endif
                                          }
                                          );
    a8::Timer::Instance()->AddDeadLineTimer(1000 * 10,
                                          a8::XParams(),
                                          [] (const a8::XParams& param)
                                          {
                                              printf("Stop App...\n");
                                              App::Instance()->UnInit();
                                              exit(0);
                                          }
                                          );
}

void App::UnInit()
{
    if (terminated) {
        return;
    }
    f8::DBPool::Instance()->UnInit();
    f8::MsgQueue::Instance()->UnInit();
    a8::Timer::Instance()->UnInit();
    UnInitLog();
    delete im_msg_mutex_;
    im_msg_mutex_ = nullptr;
    delete loop_cond_;
    loop_cond_ = nullptr;
    delete loop_mutex_;
    loop_mutex_ = nullptr;
}

void App::Terminate()
{
    terminated = true;
}

int App::Run()
{
    if (terminated) {
        return 0;
    }
    int ret = 0;
    a8::UdpLog::Instance()->Info("f8test running", {});
    while (!terminated) {
        #if 0
        a8::tick_t begin_tick = a8::XGetTickCount();
        #endif
        QuickExecute();
        SlowerExecute();
        #if 0
        a8::tick_t end_tick = a8::XGetTickCount();
        #endif
        Schedule();
    }
    return ret;
}

void App::AddIMMsg(unsigned short imcmd, a8::XParams params)
{
    IMMsgNode *p = new IMMsgNode;
    p->msgid = imcmd;
    p->params = params;
    p->next = nullptr;
    im_msg_mutex_->lock();
    if (im_bot_node_) {
        im_bot_node_->next = p;
        im_bot_node_ = p;
    } else {
        im_top_node_ = p;
        im_bot_node_ = p;
    }
    im_msg_mutex_->unlock();
    NotifyLoopCond();
}

void App::QuickExecute()
{
    ProcessIMMsg();
    a8::Timer::Instance()->Update();
}

void App::SlowerExecute()
{
}

void App::NotifyLoopCond()
{
    std::unique_lock<std::mutex> lk(*loop_mutex_);
    loop_cond_->notify_all();
}

void App::Schedule()
{
    std::unique_lock<std::mutex> lk(*loop_mutex_);
    if (!HasTask()) {
        int sleep_time = a8::Timer::Instance()->GetIdleableMillSeconds();
        loop_cond_->wait_for(lk, std::chrono::milliseconds(sleep_time));
    }
}

bool App::HasTask()
{
    {
        if (!im_work_node_) {
            im_msg_mutex_->lock();
            if (!im_work_node_ && im_top_node_) {
                im_work_node_ = im_top_node_;
                im_top_node_ = nullptr;
                im_bot_node_ = nullptr;
            }
            im_msg_mutex_->unlock();
        }
        if (im_work_node_) {
            return true;
        }
    }
    return false;
}

void App::ProcessIMMsg()
{
    if (!im_work_node_ && im_top_node_) {
        im_msg_mutex_->lock();
        im_work_node_ = im_top_node_;
        im_top_node_ = nullptr;
        im_bot_node_ = nullptr;
        im_msg_mutex_->unlock();
    }
    while (im_work_node_) {
        IMMsgNode *pdelnode = im_work_node_;
        switch (im_work_node_->msgid) {
        case f8::IM_SysMsgQueue:
            {
                const a8::XParams* param = (const a8::XParams*)pdelnode->params.param1.GetUserData();
                f8::MsgQueue::Instance()->ProcessMsg(pdelnode->params.sender.GetInt(),
                                                     *param
                                                     );
                delete param;
            }
            break;
        }
        im_work_node_ = im_work_node_->next;
        delete pdelnode;
    }
}

void App::InitLog()
{
    std::string filename_fmt = PROJ_LOG_FILENAME;
    if (getenv("is_dev_env")) {
        a8::ReplaceString(filename_fmt, "$pid", a8::XValue(0));
    } else {
        a8::ReplaceString(filename_fmt, "$pid", a8::XValue(getpid()));
    }

    a8::MkDir(a8::Format(PROJ_ROOT, {PROJ_NAME}));
    a8::MkDir(a8::Format(PROJ_LOG_ROOT, {PROJ_NAME}));
    a8::UdpLog::Instance()->SetLogFileName(a8::Format(PROJ_LOG_ROOT, {PROJ_NAME}) + "/" + filename_fmt);
    a8::UdpLog::Instance()->Init();
}

void App::UnInitLog()
{
    a8::UdpLog::Instance()->UnInit();
}

bool App::ParseOpt()
{
    int ch = 0;
    while ((ch = getopt(argc, argv, "i:")) != -1) {
        switch (ch) {
        case 'i':
            {
                instance_id = a8::XValue(optarg);
            }
            break;
        }
    }
    return instance_id > 0;
}

a8::XParams* App::AddContext(long long context_id)
{
    context_hash_[context_id] = a8::XParams();
    return GetContext(context_id);
}

void App::DelContext(long long context_id)
{
    context_hash_.erase(context_id);
}

a8::XParams* App::GetContext(long long context_id)
{
    auto itr = context_hash_.find(context_id);
    return itr != context_hash_.end() ? &(itr->second) : nullptr;
}