派生自 development/c++
syncDBTool/ErlangDbTool.cpp
@@ -71,6 +71,7 @@ return true; } // 检测返回的数据的数据类型 static int ErlangTool::checkETERMType(ETERM *elemen) { if (elemen == NULL) { return -1; @@ -187,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())); @@ -198,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) { @@ -214,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; @@ -264,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()); @@ -282,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]; @@ -313,7 +321,7 @@ INFO("item is " << item.first << " " << item.second); } } }//ret == ERL_MSG end } //ret == ERL_MSG end else { ERR(" ret is " << ret); @@ -329,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); @@ -364,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") { @@ -471,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); @@ -534,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); @@ -1619,6 +1621,7 @@ "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()) { @@ -1686,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) { @@ -1700,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; } @@ -1738,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; @@ -1786,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) { @@ -1803,6 +1806,7 @@ pthread_cond_signal(t_pthCon); pthread_mutex_unlock(t_pthMut); //call back func //#TODO 消息解析 } else { ERR("message not is update"); }
