派生自 development/c++
syncDBTool/ErlangDbTool.cpp
@@ -11,6 +11,8 @@ #include <QtCore/QString> #include <QtCore/QDebug> #include <QtCore/QVariantList> #include <basic/util/app/AppPreference.hpp> using ErlangTool::map_DevDataCache; using ErlangTool::map_TabDataCache; @@ -69,6 +71,7 @@ return true; } // 检测返回的数据的数据类型 static int ErlangTool::checkETERMType(ETERM *elemen) { if (elemen == NULL) { return -1; @@ -185,9 +188,9 @@ g_syncDbFile.setDatabaseName(str_SyncDbFile.c_str()); g_syncDbFile.open(); // c erlang init erl_init(NULL, 0); struct in_addr addr; //#todo addr.s_addr = inet_addr("127.0.0.1"); m_pName.append(std::to_string((int) getpid())); @@ -196,6 +199,8 @@ t_cNodeName.append("@127.0.0.1"); m_cNodeName.swap(t_cNodeName); //m_pName //int erl_connect_init(number, cookie, creation)int erl_connect_xinit(host, alive, node, addr, cookie, creation) // https://cloud.tencent.com/developer/section/1123179 if (erl_connect_xinit(const_cast<char *>(m_pName.c_str()), const_cast<char *>(m_pName.c_str()), const_cast<char *>(m_cNodeName.c_str()), &addr, const_cast<char *>(m_cookie.c_str()), 0) == -1) { @@ -212,15 +217,18 @@ ERR("m_nodeName is null "); return 3; } // 判断节点是否启动 m_ret = pingNode(m_nodeName, m_cookie); if (!m_ret) { if (dir_file_exists(m_path, true)) { // 启动erlang节点的bash命令 std::string cmd = std::string("cd " + m_path + " && erl -name " + m_nodeName + " -setcookie " + m_cookie + " -mnesia dir '\"" + m_path + "\"' -detached -noshell");// >>log std::cout << cmd << std::endl; system(cmd.c_str()); std::cout << m_ret << std::endl; // 判断节点是否启动成功 m_ret = waitNode(); resetConn(); return m_ret; @@ -262,7 +270,7 @@ } ErlMessage emsg; /* Incoming message */ //hanshu canshu // 拼接函数参数 ETERM *arrlist[5]; arrlist[0] = erl_mk_atom(m_nodeName.c_str()); arrlist[1] = erl_mk_atom(FatherNodeName.c_str()); @@ -280,6 +288,8 @@ //erl_close_connection(m_fd); // m_mutex.unlock(); if (ret == ERL_MSG) { // 消息解析 // 消息体为链表结构 std::map<std::string, std::string> t_results; ETERM *key, *value; ETERM *tuplep[6]; @@ -311,7 +321,7 @@ INFO("item is " << item.first << " " << item.second); } } }//ret == ERL_MSG end } //ret == ERL_MSG end else { ERR(" ret is " << ret); @@ -327,18 +337,14 @@ // m_mutex.lock(); m_ret = resetConn(); if (!m_ret) { //#todo error message std::cout << __FILE__ << __FUNCTION__ << __LINE__ << " error " << m_ret << " " << m_fd << std::endl; ERR(" error " << m_ret << " " << m_fd); return false; } int ret = -1; ErlMessage emsg; /* Incoming message */ //hanshu canshu // ETERM *arrlist[1]; // arrlist[0] = erl_mk_atom(m_nodeName.c_str()); ETERM *list = erl_mk_empty_list();// (arrlist, 0); // std::cout << __FILE__ << __FUNCTION__ <<__LINE__ << " " << m_fd << std::endl; ret = erl_rpc_to(m_fd, "syncDB", "removeNode", list); if (ret == ERL_TICK) { ret = erl_rpc_from(m_fd, TIMEOUT, &emsg); @@ -362,7 +368,7 @@ break; } default: printf("error add case todo \n\r"); ERR("error add case todo \n\r"); } auto it = t_results.find("atomic"); if (t_results.size() > 0 && t_results.end() != it && it->second == "ok") { @@ -469,9 +475,7 @@ map_DevDataCache devDataCache; int ret = -1; ErlMessage emsg; /* Incoming message */ //hanshu canshu // ETERM *arrlist[0]; // arrlist[0] = erl_mk_atom(m_nodeName.c_str()); ETERM *list = erl_mk_empty_list();//erl_mk_list(arrlist, 0); ret = erl_rpc_to(m_fd, "syncDB", "findAllNode", list); @@ -532,7 +536,7 @@ erlangFreeEterm(3, key, value, tuplep[j]); } // printf("\none list end\n\n\n\n"); // #todo this is have a bug // #todo this is have a bug 内存泄露?或者地址空间越界? // device_info.create_by = ""; devDataCache.insert(std::make_pair(device_info.uuid, device_info)); erlangFreeEterm(1, tail_tuple); @@ -639,6 +643,14 @@ return false; } bool ErlangTool::ErlangDbTool::createDatabase(std::string t_tableName, FieldValues &fieldValues) { // #todo 记得测试中文 std::string strSql = getCreateFaceTableSql(t_tableName); strSql.append(getInsertSql("sys_o_tables", fieldValues)); getExecSqlResult(strSql); } bool ErlangTool::ErlangDbTool::updateDatabase(std::string UUID, std::string TableType, std::string TableName, bool SyncType, std::string BwType, std::string StartTime, std::string EndTime) { @@ -722,6 +734,86 @@ return false; } bool ErlangTool::ErlangDbTool::updateDatabase(std::string t_tableName, FieldValues &fieldValues) { // #todo 记得测试中文 std::string strSql = getUpdateFaceTableSql(t_tableName, fieldValues); return getExecSqlResult(strSql); } bool ErlangTool::ErlangDbTool::getExecSqlResult(const std::string &strSql) { ErlangTool::LockG lock(m_mutex); // m_mutex.lock(); m_ret = resetConn(); if (!m_ret) { //#todo error message std::cout << __FILE__ << __FUNCTION__ << __LINE__ << " error " << this->m_ret << " " << this->m_fd << std::endl; return false; } ErlMessage emsg; /* Incoming message */ int ret = -1; ETERM *arrlist[7]; arrlist[0] = erl_mk_string(strSql.c_str()); ETERM *list = erl_mk_list(arrlist, 1); ret = erl_rpc_to(m_fd, "syncDB", "exeSomeSql", list); if (ret == ERL_TICK) { ret = erl_rpc_from(m_fd, TIMEOUT, &emsg); //erl_close_connection(m_fd); // m_mutex.unlock(); if (ret == ERL_MSG) { std::map<std::__cxx11::string, std::__cxx11::string> t_results; ETERM *tail_list; ETERM *list_ret[1000], *arr_ret[2]; ETERM *key, *value; ETERM *tuplep[6]; arr_ret[0] = erl_element(2, emsg.msg); int erlLength = erl_length(arr_ret[0]); printf("arr_ret[0]:%d\n", erlLength); list_ret[0] = erl_hd(arr_ret[0]); tail_list = erl_tl(arr_ret[0]); for (int i = 0; i < erlLength; i++) { if (i > 0) { list_ret[i] = erl_hd(tail_list); tail_list = erl_tl(tail_list); } // printf("list_ret[%d]:%d\n", i, erl_length(list_ret[i])); tuplep[0] = erl_hd(list_ret[i]); key = erl_element(1, list_ret[i]); value = erl_element(2, list_ret[i]); // printf("key:%s\n", ERL_ATOM_PTR(key)); switch (ErlangTool::checkETERMType(value)) { case ERL_ATOM: { // int atomSize = ERL_ATOM_SIZE(value); char *atomValue = ERL_ATOM_PTR(value); // printf("value is a atom: atomSize:%d, atomValue:%s \n\r", atomSize, atomValue); t_results.insert(std::make_pair(ERL_ATOM_PTR(key), atomValue)); break; } default: printf("error add case todo \n\r"); } ErlangTool::erlangFreeEterm(4, key, value, tuplep[0], list_ret[i]); } ErlangTool::erlangFreeEterm(2, tail_list, arr_ret[0]); erl_eterm_release(); auto it = t_results.find("atomic"); if (t_results.size() > 0 && t_results.end() != it && it->second == "ok") { return true; } } } erl_free_array(arrlist, 1); // erlangFreeEterm(4, emsg.to, emsg.msg, emsg.from, list); erl_eterm_release(); return false; } bool ErlangTool::ErlangDbTool::deleteDatabase(std::string TableType, std::string TableName, bool SyncType) { LockG lock(m_mutex); // m_mutex.lock(); @@ -791,6 +883,19 @@ erl_free_array(arrlist, 2); erlangFreeEterm(4, emsg.to, emsg.msg, emsg.from, list); return false; } bool ErlangTool::ErlangDbTool::deleteDatabase(std::string t_tableName) { FieldValues fieldValues; fieldValues.insert(std::make_pair("uuid", t_tableName)); fieldValues.insert(std::make_pair("del_flag", "1")); std::string strSql = getUpdateFaceTableSql("sys_o_tables", fieldValues); strSql.append("DROP TABLE '" + t_tableName + "';DROP TABLE '" + t_tableName + "_fea';"); // #todo 记得测试中文 return getExecSqlResult(strSql); } // 和erlang交互 @@ -894,7 +999,7 @@ map_TabDataCache tabDataCache; QString sql = QString::fromStdString( "SELECT uuid,tableName,tableType,bwType,startTime,endTime FROM sys_o_tables where del_flag=0;"); "SELECT uuid,tableName,tableType,bwType,startTime,endTime,uploadFlag,cmpThreshold,enabled,create_by FROM sys_o_tables where del_flag=0;"); QSqlQuery query(g_syncDbFile); query.prepare(sql); if (!query.exec()) { @@ -908,6 +1013,10 @@ t_table_info.bwType = query.value(3).toString().toStdString(); t_table_info.startTime = query.value(4).toString().toStdString(); t_table_info.endTime = query.value(5).toString().toStdString(); t_table_info.uploadFlag = query.value(6).toString().toStdString(); t_table_info.cmpThreshold = query.value(7).toString().toStdString(); t_table_info.enabled = query.value(8).toString().toStdString(); t_table_info.create_by = query.value(9).toString().toStdString(); tabDataCache.insert(std::make_pair(t_table_info.uuid, t_table_info)); } @@ -1080,6 +1189,35 @@ return false; } bool ErlangTool::ErlangDbTool::addPerson(std::string t_tableName, AddFaceData &faceData, FieldValues &fieldValues) { // #todo 记得测试中文 std::string strSql = getInsertSql(t_tableName, fieldValues); fieldValues.clear(); fieldValues.insert(make_pair("uuid", faceData.uuid)); fieldValues.insert(make_pair("faceUrl", faceData.faceUrl)); fieldValues.insert(make_pair("feature", faceData.feature_base64)); std::string strSqlFea = getInsertSql(t_tableName + "_fea", fieldValues); strSql.append(strSqlFea); // #todo 记得测试中文 只有人员信息表更新，特征更新暂时没做 return getExecSqlResult(strSql); } bool ErlangTool::ErlangDbTool::updatePerson(std::string t_tableName, AddFaceData &faceData, FieldValues &fieldValues) { // #todo 记得测试中文 std::string strSql = getUpdateFaceTableSql(t_tableName, fieldValues); fieldValues.clear(); fieldValues.insert(make_pair("uuid", faceData.uuid)); fieldValues.insert(make_pair("faceUrl", faceData.faceUrl)); fieldValues.insert(make_pair("feature", faceData.feature_base64)); std::string strSqlFea = getInsertSql(t_tableName + "_fea", fieldValues); strSql.append(strSqlFea); return getExecSqlResult(strSql); } ErlangTool::vec_AddDataCache ErlangTool::ErlangDbTool::addPersons(std::string TableUuid, std::vector<SyncDB::AddPersonInfo> &tmpPer) { @@ -1480,8 +1618,10 @@ } if (str_tableUuid.size() > 0) { QString sql = QString::fromStdString( "Select a.uuid as id ,a.faceUrl as img,a.feature,b.idCard as idcard from '" + str_tableUuid + "Select a.uuid as id ,a.faceUrl as img,a.feature,b.idCard as idcard,b.enabled,b.monitorLevel from '" + str_tableUuid + "_fea' as a ,'" + str_tableUuid + "' as b where a.uuid = b.uuid and ( a.del_flag=0 AND b.del_flag=0);"); //Select a.uuid as id ,a.faceUrl as img,a.feature,b.idCard as idcard,b.enabled,b.monitorLevel from '0cd82a8b-5285-5737-ab19-8c07247c797e_fea' as a ,'0cd82a8b-5285-5737-ab19-8c07247c797e' as b where a.uuid = b.uuid and ( a.del_flag=0 AND b.del_flag=0); QSqlQuery query(g_syncDbFile); query.prepare(sql); if (!query.exec()) { @@ -1493,6 +1633,8 @@ t_feature_info.img = query.value(1).toString().toStdString(); t_feature_info.feature = query.value(2).toString().toStdString(); t_feature_info.idcard = query.value(3).toString().toStdString(); t_feature_info.enable = query.value(4).toString().toStdString(); t_feature_info.monLevel = query.value(5).toString().toStdString(); dataCache.insert(std::make_pair(t_feature_info.id, t_feature_info)); } @@ -1517,9 +1659,12 @@ } } if (str_tableUuid.size() > 0) { // QString sql = QString::fromStdString( // "Select b.uuid as id,b.faceUrl as img,a.idCard as idcard,a.enable,a.monitorLevel from '" + str_tableUuid + // "' as a, '" + str_tableUuid + "_fea' As b where a.uuid = b.uuid and ( a.del_flag=0 AND b.del_flag=0);"); QString sql = QString::fromStdString( "Select b.uuid as id,b.faceUrl as img,a.idCard as idcard from '" + str_tableUuid + "' as a, '" + str_tableUuid + "_fea' As b where a.uuid = b.uuid and ( a.del_flag=0 AND b.del_flag=0);"); "Select b.uuid as id,b.faceUrl as img,a.idCard as idcard from '" + str_tableUuid + "' as a, '" + str_tableUuid + "_fea' As b where a.uuid = b.uuid and ( a.del_flag=0 AND b.del_flag=0);"); QSqlQuery query(g_syncDbFile); query.prepare(sql); if (!query.exec()) { @@ -1530,6 +1675,8 @@ t_feature_info.id = query.value(0).toString().toStdString(); t_feature_info.img = query.value(1).toString().toStdString(); t_feature_info.idcard = query.value(2).toString().toStdString(); // t_feature_info.enable = query.value(3).toString().toStdString(); // t_feature_info.monLevel = query.value(4).toString().toStdString(); dataCache.insert(std::make_pair(t_feature_info.id, t_feature_info)); } @@ -1542,13 +1689,14 @@ bool ErlangTool::ErlangDbTool::resetConn() { //#todo if (m_pName.size() <= 0) { // 根据当前时间设置种子,秒级 srand(time(0)); // 生成随机数 m_loop = rand() % 1000; int ret = -1; erl_init(NULL, 0); // m_loop %= 10; // m_loop++; // 根据随机数生成c节点 ret = erl_connect_init(m_loop, const_cast<char *>(m_cookie.c_str()), 0); DBG("node name is " << m_loop); if (-1 == ret) { @@ -1556,26 +1704,22 @@ return false; } // std::cout << __FILE__ << __FUNCTION__ <<__LINE__ << " " << m_fd << std::endl; // 关闭上次链接的文件描述符 #TODO 第一次创建节点有概率出现短时间内无法查询的bug erl_close_connection(m_fd); // std::cout << __FILE__ << __FUNCTION__ <<__LINE__ << " " << m_fd << std::endl; m_fd = erl_connect(const_cast<char *>(m_nodeName.c_str())); // std::cout << __FILE__ << __FUNCTION__ <<__LINE__ << " " << m_fd << std::endl; if (0 > m_fd) { // erl_err_quit("erl_connect"); return false; } return true; } else { // 设置了 mPName erl_close_connection(m_fd); m_fd = erl_connect(const_cast<char *>(m_nodeName.c_str())); if (0 > m_fd) { // erl_err_quit("erl_connect"); // std::cout << __FILE__ << __FUNCTION__ <<__LINE__ << " error " << fd << std::endl; ERR("m_fd is " << m_fd << " m_nodeName is" << m_nodeName); return false; } // std::cout << __FILE__ << __FUNCTION__ <<__LINE__ << " " << fd << std::endl; INFO("m_fd is " << m_fd << " m_nodeName is" << m_nodeName); return true; } @@ -1594,15 +1738,17 @@ loop++; } } std::cout << "start waitNode" << std::endl; INFO("start waitNode"); return m_ret; } void ErlangTool::ErlangDbTool::test(pthread_cond_t *t_pthcon, pthread_mutex_t *t_pthmut) { bool ret = setCNodeName(); { // 监听socket的线程 std::thread th1([&] { int res; // 外部传入的信号和锁 pthread_cond_t *t_pthCon = t_pthcon; pthread_mutex_t *t_pthMut = t_pthmut; @@ -1618,7 +1764,7 @@ ETERM *fromp, *tuplep, *fnp, *argp, *resp; int port; /* Listen port number */ port = PORT; port = appPref.getIntData("ErlNodePort"); /* Make a listen socket */ if ((listen = my_listen(port)) <= 0) { ERR("my_listen is error"); @@ -1642,7 +1788,8 @@ } else if (got == ERL_ERROR) { if ((fd = erl_accept(listen, &conn)) == ERL_ERROR) erl_err_quit("erl_accept"); INFO("Connected to " << conn.nodename); else INFO("Connected to " << conn.nodename); } else { if (emsg.type == ERL_REG_SEND) { @@ -1659,6 +1806,7 @@ pthread_cond_signal(t_pthCon); pthread_mutex_unlock(t_pthMut); //call back func //#TODO 消息解析 } else { ERR("message not is update"); }
