measure_module/parkspace_allocation/database_controller.cpp

/*
 * @Description: 数据库操作类
 * @Author: yct
 * @Date: 2020-07-15 14:08:46
 * @LastEditTime: 2020-07-19 00:02:06
 * @LastEditors: yct
 */

#include "database_controller.h"

//初始化
Error_manager Database_controller::database_controller_init(std::string ip, int port, std::string username, std::string pass, std::string db_name, int conn_size)
{
    //系统结束标记与连接池大小
    mb_exit = false;
    m_conn_pool_size = conn_size;
    m_db_param.connect_string = std::string("tcp://").append(ip).append(":").append(std::to_string(port));
    m_db_param.username = username;
    m_db_param.pass = pass;
    m_db_param.db_name = db_name;
    //mysql驱动
    if(m_mysql_driver == nullptr)
    {
        m_mysql_driver = sql::mysql::get_driver_instance();
    }
    if(m_mysql_driver == nullptr)
    {
        mb_initialized = false;
        return DB_INIT_FAILED;
    }
    //创建连接池
    m_conn_mutex.lock();
    for (int i = 0; i < m_conn_pool_size; i++)
    {
        try
        {
            boost::shared_ptr<sql::Connection> t_conn_ptr(m_mysql_driver->connect(m_db_param.connect_string, m_db_param.username, m_db_param.pass));
            if (t_conn_ptr->isValid())
            {
                m_db_conn_pool.push(t_conn_ptr);
                std::cout << "conn pushed." << std::endl;
            }
            else
            {
                std::cout << "conn create failed." << std::endl;
            }
        }
        catch (sql::SQLException &e)
        {
            char buf[1024];
            memset(buf, 0, 1024);
            sprintf(buf, "# ERR: %s\n (MySQL error code: %d, SQLState: %s", e.what(), e.getErrorCode(), e.getSQLState().c_str());
            std::cout << buf << std::endl;
        }
    }
    m_conn_mutex.unlock();
    
    //守护线程，检查连接状态并保持连接数量
    mp_guard_thread = new std::thread(&Database_controller::database_status_update, this);
    mb_initialized = true;
    if(check_status() == SUCCESS)
    {
        mb_connected = true;
    }else{
        mb_connected = false;
    }

    return SUCCESS;
}

//反初始化
Error_manager Database_controller::database_controller_uninit()
{
    mb_exit = true;
    if(mp_guard_thread!=nullptr)
    {
        if(mp_guard_thread->joinable())
            mp_guard_thread->join();
        delete mp_guard_thread;
        mp_guard_thread = nullptr;
    }
}

//数据库连接状态
bool Database_controller::is_connected()
{
    return mb_connected;
}

//****** 增删改查功能 *******
//增
Error_manager Database_controller::sql_insert(std::string sql_str)
{
    // 1.检查状态，一旦无可用连接，则主动创建一个，若依然失败则直接返回错误
    Error_manager ec = check_status();
    if(ec != SUCCESS)
    {
        ec = fill_up_pool(1);
    }
    if(ec == SUCCESS)
    {
        // 从连接池获取连接
        m_conn_mutex.lock();
        boost::shared_ptr<sql::Connection> t_conn_ptr = m_db_conn_pool.front();
        m_db_conn_pool.pop();
        m_conn_mutex.unlock();
        // 使用该连接并在结束判断连接状态，将正常连接丢回队列
        if(t_conn_ptr!= nullptr && t_conn_ptr->isValid()){
            char buf[1024];
            memset(buf, 0, 1024);
            try
            {
                t_conn_ptr->setSchema(m_db_param.db_name);
                boost::scoped_ptr<sql::Statement> stmt(t_conn_ptr->createStatement());
                stmt->execute(sql_str);
                // 丢回队列
                if(t_conn_ptr!= nullptr && t_conn_ptr->isValid())
                {
                    m_conn_mutex.lock();
                    m_db_conn_pool.push(t_conn_ptr);
                    m_conn_mutex.unlock();
                }
                return SUCCESS;
            }
            catch (sql::SQLException &e)
            {
                /* Use what() (derived from std::runtime_error) to fetch the error message */
                sprintf(buf, "# ERR: %s\n (MySQL error code: %d, SQLState: %s", e.what(), e.getErrorCode(), e.getSQLState().c_str());
                usleep(1000* 3000);
                return Error_manager(DB_INSERT_FAILED, NEGLIGIBLE_ERROR, buf);
            }
            catch (std::runtime_error &e)
            {
                sprintf(buf, "# ERR: %s\n ERR: runtime_error in  %s ",e.what(),__FILE__);
                return Error_manager(DB_INSERT_FAILED, NEGLIGIBLE_ERROR, buf);
            }
            // catch (std::exception &e)
            // {
            //     sprintf(buf, "# ERR: %s\n ERR: Standard exception in  %s ",e.what(),__FILE__);
            //     return Error_manager(DB_INSERT_FAILED, NEGLIGIBLE_ERROR, buf);
            // }
        }
        else
        {
            return DB_CONNECT_FAILED;
        }
    }else{
        return ec;
    }
}

//删
Error_manager Database_controller::sql_delete(std::string sql_str)
{
    // 1.检查状态，一旦无可用连接，则主动创建一个，若依然失败则直接返回错误
    Error_manager ec = check_status();
    if(ec != SUCCESS)
    {
        ec = fill_up_pool(1);
    }
    if(ec == SUCCESS)
    {
        // 从连接池获取连接
        m_conn_mutex.lock();
        boost::shared_ptr<sql::Connection> t_conn_ptr = m_db_conn_pool.front();
        m_db_conn_pool.pop();
        m_conn_mutex.unlock();
        // 使用该连接并在结束判断连接状态，将正常连接丢回队列
        if(t_conn_ptr!= nullptr && t_conn_ptr->isValid()){
            char buf[1024];
            memset(buf, 0, 1024);
            try
            {
                t_conn_ptr->setSchema(m_db_param.db_name);
                boost::scoped_ptr<sql::Statement> stmt(t_conn_ptr->createStatement());
                stmt->execute(sql_str);
                // 丢回队列
                if(t_conn_ptr!= nullptr && t_conn_ptr->isValid())
                {
                    m_conn_mutex.lock();
                    m_db_conn_pool.push(t_conn_ptr);
                    m_conn_mutex.unlock();
                }
                return SUCCESS;
            }
            catch (sql::SQLException &e)
            {
                /* Use what() (derived from std::runtime_error) to fetch the error message */
                sprintf(buf, "# ERR: %s\n (MySQL error code: %d, SQLState: %s", e.what(), e.getErrorCode(), e.getSQLState().c_str());
                return Error_manager(DB_DELETE_FAILED, NEGLIGIBLE_ERROR, buf);
            }
            catch (std::runtime_error &e)
            {
                sprintf(buf, "# ERR: %s\n ERR: runtime_error in  %s ",e.what(),__FILE__);
                return Error_manager(DB_DELETE_FAILED, NEGLIGIBLE_ERROR, buf);
            }
        }
        else
        {
            return DB_CONNECT_FAILED;
        }
    }else{
        return ec;
    }
}

//改
Error_manager Database_controller::sql_update(std::string sql_str)
{
    // 1.检查状态，一旦无可用连接，则主动创建一个，若依然失败则直接返回错误
    Error_manager ec = check_status();
    if(ec != SUCCESS)
    {
        ec = fill_up_pool(1);
    }
    if(ec == SUCCESS)
    {
        // 从连接池获取连接
        m_conn_mutex.lock();
        boost::shared_ptr<sql::Connection> t_conn_ptr = m_db_conn_pool.front();
        m_db_conn_pool.pop();
        m_conn_mutex.unlock();
        // 使用该连接并在结束判断连接状态，将正常连接丢回队列
        if(t_conn_ptr!= nullptr && t_conn_ptr->isValid()){
            char buf[1024];
            memset(buf, 0, 1024);
            try
            {
                t_conn_ptr->setSchema(m_db_param.db_name);
                boost::scoped_ptr<sql::Statement> stmt(t_conn_ptr->createStatement());
                int affected_rows = stmt->executeUpdate(sql_str);
                // 丢回队列
                if(t_conn_ptr!= nullptr && t_conn_ptr->isValid())
                {
                    m_conn_mutex.lock();
                    m_db_conn_pool.push(t_conn_ptr);
                    m_conn_mutex.unlock();
                }
                if (affected_rows > 0)
                {
                    return SUCCESS;
                }
                else
                {
                    return DB_UPDATE_FAILED;
                }
            }
            catch (sql::SQLException &e)
            {
                /* Use what() (derived from std::runtime_error) to fetch the error message */
                sprintf(buf, "# ERR: %s\n (MySQL error code: %d, SQLState: %s", e.what(), e.getErrorCode(), e.getSQLState().c_str());
                return Error_manager(DB_UPDATE_FAILED, NEGLIGIBLE_ERROR, buf);
            }
            catch (std::runtime_error &e)
            {
                sprintf(buf, "# ERR: %s\n ERR: runtime_error in  %s ",e.what(),__FILE__);
                return Error_manager(DB_UPDATE_FAILED, NEGLIGIBLE_ERROR, buf);
            }
        }
        else
        {
            return DB_CONNECT_FAILED;
        }
    }else{
        return ec;
    }
}

//查
Error_manager Database_controller::sql_query(std::string sql_str, boost::shared_ptr< sql::ResultSet > &query_result)
{
    // 1.检查状态，一旦无可用连接，则主动创建一个，若依然失败则直接返回错误
    Error_manager ec = check_status();
    if(ec != SUCCESS)
    {
        ec = fill_up_pool(1);
    }
    if(ec == SUCCESS)
    {
        // 从连接池获取连接
        m_conn_mutex.lock();
        boost::shared_ptr<sql::Connection> t_conn_ptr = m_db_conn_pool.front();
        // std::cout<<"conn handle: "<<t_conn_ptr<<std::endl;
        m_db_conn_pool.pop();
        // std::cout<<"conn handle after pop: "<<t_conn_ptr<<std::endl;
        m_conn_mutex.unlock();
        // 使用该连接并在结束判断连接状态，将正常连接丢回队列
        if(t_conn_ptr!= nullptr && t_conn_ptr->isValid()){
            char buf[1024];
            memset(buf, 0, 1024);
            try
            {
                t_conn_ptr->setSchema(m_db_param.db_name);
                boost::scoped_ptr< sql::Statement > stmt(t_conn_ptr->createStatement());
                query_result = boost::shared_ptr< sql::ResultSet >(stmt->executeQuery(sql_str));
                // 丢回队列
                if(t_conn_ptr!= nullptr && t_conn_ptr->isValid())
                {
                    m_conn_mutex.lock();
                    m_db_conn_pool.push(t_conn_ptr);
                    m_conn_mutex.unlock();
                }
                // 检查结果
                if(query_result!=nullptr)
                {
                    return SUCCESS;
                }else{
                    return DB_QUERY_FAILED;
                }
            }
            catch (sql::SQLException &e)
            {
                /* Use what() (derived from std::runtime_error) to fetch the error message */
                sprintf(buf, "# ERR: %s\n (MySQL error code: %d, SQLState: %s", e.what(), e.getErrorCode(), e.getSQLState().c_str());
                return Error_manager(DB_UPDATE_FAILED, NEGLIGIBLE_ERROR, buf);
            }
            catch (std::runtime_error &e)
            {
                sprintf(buf, "# ERR: %s\n ERR: runtime_error in  %s ",e.what(),__FILE__);
                return Error_manager(DB_UPDATE_FAILED, NEGLIGIBLE_ERROR, buf);
            }
        }else{
            return DB_CONNECT_FAILED;
        }
    }else{
        return ec;
    }
}

void Database_controller::database_status_update()
{
    while(!mb_exit && mb_initialized)
    {
        std::cout<<"guard thread working."<<std::endl;
        // 检查连接状态，直到获得正常连接或队列为空为止
        check_status();
        fill_up_pool(m_conn_pool_size);
        usleep(1000*DB_UPDATE_INTERVAL_MILLI);
    }
}

// 检查连接状态
Error_manager Database_controller::check_status()
{
    std::lock_guard<std::mutex> lck(m_conn_mutex);
    while (m_db_conn_pool.size() > 0)
    {
        boost::shared_ptr<sql::Connection> t_conn_ptr = m_db_conn_pool.front();
        if (t_conn_ptr != nullptr && t_conn_ptr->isValid())
        {
            mb_connected = true;
            break;
        }
        else
        {
            mb_connected = false;
            m_db_conn_pool.pop();
        }
    }
    if(mb_connected)
        return SUCCESS;
    else
        return DB_CONNECT_FAILED;
}

// 补充连接
Error_manager Database_controller::fill_up_pool(int add_num)
{
    std::lock_guard<std::mutex> lck(m_conn_mutex);
    // 填充连接池
    int retry_count = 3;
    int added_num = 0;
    while (m_db_conn_pool.size() < m_conn_pool_size && added_num<add_num)
    {
        try
        {
            // 三次创建连接失败则暂时释放锁
            if (retry_count <= 0)
            {
                mb_connected = false;
                break;
            }

            boost::shared_ptr<sql::Connection> t_conn_ptr(m_mysql_driver->connect(m_db_param.connect_string, m_db_param.username, m_db_param.pass));

            if (t_conn_ptr != nullptr && t_conn_ptr->isValid())
            {
                m_db_conn_pool.push(t_conn_ptr);
                added_num++;
            }
            else
            {
                retry_count--;
            }
            
        }
        catch (sql::SQLException &e)
        {
            char buf[1024];
            memset(buf, 0, 1024);
            sprintf(buf, "# ERR: %s\n (MySQL error code: %d, SQLState: %s", e.what(), e.getErrorCode(), e.getSQLState().c_str());
            std::cout << buf << std::endl;
            retry_count--;
        }
    }
    m_conn_mutex.unlock();
    if(mb_connected)
    {
        return SUCCESS;
    }else{
        return DB_CONNECT_FAILED;
    }
}