f8/cpp/tglog.cc

#include <unistd.h>
#include <thread>
#include <mutex>
#include <condition_variable>

#include <a8/a8.h>
#include <a8/mutable_xobject.h>
#include "tglog.h"

namespace f8
{
    static const char* const TGLOG_ROOT = "/data/logs/%s/upload/";
    static const char* const POLY_TGLOG_ROOT = "/data/logs/%s/%s/upload/";
    static const char* const TGLOG_FILENAME = "log_$pid_%Y%m%d%H.log";

    struct TGLogMsgNode
    {
        int game_id = 0;
        std::string jsonstr;
        TGLogMsgNode* next = nullptr;
    };

    struct TGLogImpl
    {
        std::string filename_fmt;
        std::string project_name;
        bool is_poly_log = false;

        std::thread* save_thread = nullptr;
        volatile bool save_thread_shutdown = false;

        std::mutex msg_mutex;
        TGLogMsgNode* top_node = nullptr;
        TGLogMsgNode* bot_node = nullptr;
        TGLogMsgNode* work_node = nullptr;
        std::mutex *save_cond_mutex = nullptr;
        std::condition_variable *save_cond = nullptr;

        TGLogImpl()
        {
            save_cond_mutex = new std::mutex();
            save_cond = new std::condition_variable();
        }

        virtual ~TGLogImpl()
        {
            save_thread_shutdown = true;
            save_thread->join();
            delete save_thread;
            delete save_cond_mutex;
            save_cond_mutex = nullptr;
            delete save_cond;
            save_cond = nullptr;
        }

        std::string GetCurrentFileName()
        {
            time_t nowtime = time(nullptr);
            struct tm tm_nowtime = {0};
            localtime_r(&nowtime, &tm_nowtime);

            char szfilename[256];
            szfilename[0] = '\0';
            strftime(szfilename, a8::ArraySize(szfilename), filename_fmt.c_str(), &tm_nowtime);
            std::string filename((char*)szfilename);
            return filename;
        }
    };

    void TGLog::Init(const std::string& project_name, bool is_poly_log)
    {
        project_name_ = project_name;
        is_poly_log_ = is_poly_log;
        impl_ = new TGLogImpl();
        impl_->filename_fmt = TGLOG_FILENAME;
        a8::ReplaceString(impl_->filename_fmt, "$pid", a8::XValue(getpid()));

        impl_->project_name = project_name_;
        impl_->is_poly_log = is_poly_log_;

        impl_->save_thread_shutdown = false;
        impl_->save_thread = new std::thread(&TGLog::SaveToFileThreadProc, this);
    }

    void TGLog::UnInit()
    {
        delete impl_;
        impl_ = nullptr;
    }

    void TGLog::SetProjectInfo(const std::string& project_name, bool is_poly_log)
    {
        project_name_ = project_name;
        is_poly_log_ = is_poly_log;
    }

    void TGLog::AddTrackLog(int game_id, const std::string& accountid, unsigned long ip_saddr,
                            int logclass1, int logclass2, a8::XObject* prop)
    {
        std::string event_name = a8::Format("event_%d_%d", {logclass1, logclass2});
        AddTrackLog(game_id,
                    accountid,
                    ip_saddr,
                    event_name,
                    prop);
    }

    void TGLog::AddTrackLog(int game_id, const std::string& accountid, unsigned long ip_saddr,
                            const std::string& event_name, a8::XObject* prop)
    {
        InternalAddTrackLog(game_id,
                            accountid,
                            ip_saddr,
                            event_name,
                            prop);
    }

    void TGLog::SaveToFileThreadProc()
    {
        auto GetLogFile = [this] (std::map<int, FILE*>& opened_log_files_hash, int game_id, const std::string& filename) -> FILE*
                          {
                              auto itr = opened_log_files_hash.find(game_id);
                              if (itr == opened_log_files_hash.end()) {
                                  std::string log_dir;
                                  if (impl_->is_poly_log) {
                                      log_dir = a8::Format(POLY_TGLOG_ROOT, {impl_->project_name, impl_->work_node->game_id});
                                  } else {
                                      log_dir = a8::Format(TGLOG_ROOT, {impl_->project_name});
                                  }
                                  a8::ForceCreateDir(log_dir);
                                  FILE* logfile = fopen((log_dir + filename).c_str(), "a+");
                                  if (logfile) {
                                      opened_log_files_hash[game_id] = logfile;
                                  }
                                  return logfile;
                              } else {
                                  return itr->second;
                              }
                          };

        while (!impl_->save_thread_shutdown) {
            if (!impl_->work_node && impl_->top_node) {
                impl_->msg_mutex.lock();
                impl_->work_node = impl_->top_node;
                impl_->top_node = nullptr;
                impl_->bot_node = nullptr;
                impl_->msg_mutex.unlock();
            }

            if (impl_->work_node) {
                std::string filename = impl_->GetCurrentFileName();
                std::map<int, FILE*> opened_log_files_hash;

                while (impl_->work_node) {
                    TGLogMsgNode* nextnode = impl_->work_node->next;
                    TGLogMsgNode* work_node = impl_->work_node;

                    FILE* logfile = GetLogFile(opened_log_files_hash, work_node->game_id, filename);
                    if (logfile) {
                        work_node->jsonstr.push_back('\n');
                        fwrite(work_node->jsonstr.c_str(), 1, work_node->jsonstr.size(), logfile);
                    }
                    delete work_node;

                    impl_->work_node = nextnode;
                }

                for (auto& pair : opened_log_files_hash) {
                    if (pair.second) {
                        fclose(pair.second);
                    }
                }
            }

            {
                std::unique_lock<std::mutex> lk(*impl_->save_cond_mutex);
                impl_->save_cond->wait_for(lk, std::chrono::seconds(10));
            }
        }

    }

    void TGLog::InternalAddTrackLog(int game_id, const std::string& accountid, unsigned long ip_saddr,
                                    const std::string& event_name, a8::XObject* prop)
    {
        std::string logtime_str;
        {
            time_t nowtime = time(nullptr);
            struct tm tm_time = {0};
            localtime_r(&nowtime, &tm_time);

            char buff[256];
            strftime(buff, a8::ArraySize(buff), "%F %T", &tm_time);
            logtime_str.append((char*)buff);
        }
        a8::MutableXObject* xobj = a8::MutableXObject::NewObject();

        xobj->SetVal("#account_id", accountid);
        xobj->SetVal("#type", "track");
        xobj->SetVal("#time", logtime_str);
        xobj->SetVal("#ip", a8::GetIpAddress(ip_saddr));
        xobj->SetVal("#event_name", event_name);

        xobj->SetVal("properties", *prop);
        {
            TGLogMsgNode *p = new TGLogMsgNode();
            p->game_id = game_id;
            xobj->ToJsonStr(p->jsonstr);
            impl_->msg_mutex.lock();
            if (impl_->bot_node) {
                impl_->bot_node->next = p;
                impl_->bot_node = p;
            } else {
                impl_->top_node = p;
                impl_->bot_node = p;
            }
            impl_->msg_mutex.unlock();
            {
                std::unique_lock<std::mutex> lk(*impl_->save_cond_mutex);
                impl_->save_cond->notify_all();
            }
        }
        delete xobj;
        xobj = nullptr;
    }

}