lisonwu/dcapp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

调试insert

Browse Source
This commit is contained in:
why
2025-02-26 17:33:40 +08:00
parent 43ee603915
commit 583ade8cf3
10 changed files with 11 additions and 26029 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+8 -1
View File
@@ -9,6 +9,9 @@ include_directories(./
LINK_DIRECTORIES(${PROJECT_SOURCE_DIR}/lib/x86)
#LINK_DIRECTORIES(${PROJECT_SOURCE_DIR}/lib/arm)
set(CMAKE_BUILD_WITH_INSTALL_RPATH TRUE) # 启用按安装位置设置 RPATH
set(CMAKE_INSTALL_RPATH "${PROJECT_SOURCE_DIR}/lib/x86") # 定义具体的 RPATH 地址
aux_source_directory(./datacenter DC_SRC)
aux_source_directory(./dataservice DS_SRC)
@@ -20,7 +23,11 @@ add_executable(dcapp
${PROJECT_SOURCE_DIR}/datacenter/src/queue.cpp
${PROJECT_SOURCE_DIR}/datacenter/src/mqttClient.cpp
${PROJECT_SOURCE_DIR}/datacenter/src/ToolUtils.cpp
${PROJECT_SOURCE_DIR}/datacenter/bizService/F_InsertRecord.cpp
${DC_SRC} ${DS_SRC} ${DQ_SRC})
target_link_libraries(dcapp
paho-mqtt3c)
paho-mqtt3c
sqlite3)
datacenter/doc/mqttCfg.cfg
+1 -1
View File
@@ -1,7 +1,7 @@
{
"password":"shb@123",
"clientID":"3fb61fa27a2242f6ad7032a21c489930",
"serverUrl":"192.168.96.135",
"serverUrl":"192.168.96.24",
"gatewayID":"092800000001",
"username":"shb_test"
}
datacenter/doc/mqttTopCfg.cfg
+1 -1
View File
@@ -22,7 +22,7 @@
"F-DelUserDic"
],
"subTopics":[
"/test1-topic",
"/F-InsertRecord",
"/test2-topic"
]
}
datacenter/src/datacenter.cpp
+1
View File
@@ -83,6 +83,7 @@ bool DataCenter::handleMsg(msg_mqtt_t inMqttMsg) {
}
std::map<std::string, IBizService*> map = BizServiceFactovy::getBizServiceMap();
for (std::map<std::string, IBizService*>::iterator itor = map.begin(); itor != map.end(); itor++) {
printf("fid:%s\n",itor->second->GetFID().c_str());
if(itor->second->GetFID().compare(pidStr) == 0){
fRet = itor->second->HandleMsg(intMsg,outMsg, NULL);
break;
dataservice/F_InsertRecord.cpp
-269
View File
@@ -1,269 +0,0 @@
/*
* @Description: file content
* @Author: why
* @version:
* @Date: 2025-02-21 14:03:54
* @LastEditors: why
* @LastEditTime: 2025-02-26 10:13:32
*/
#include <iostream>
#include <chrono>
#include <iomanip>
#include <sstream>
#include <string>
#include <nlohmann/json.hpp>
#include "F_InsertRecord.h"
using namespace std;
using json = nlohmann::json // for convenience
typedef struct topic{
std::string src;
std::string des;
std::string sid;
std::string fid;
} topic_type;
InsertRecord::InsertRecord(void){
handpara = nullptr;
initpara = nullptr;
}
InsertRecord::~InsertRecord(void){
}
void InsertRecord::Init(initparam_t *param){
initpara = param;
}
void hexstr_to_str(void *dst, const void *src, int src_len)
{
int i;
BYTE low, high, ch;
BYTE *dst_ptr = (BYTE *) dst;
const BYTE *src_ptr = (const BYTE *) src;
for (i = 0; i < src_len / 2; i++)
{
ch = *src_ptr++;
if (ch >= '0' && ch <= '9')
high = ch - '0';
else if (ch >= 'A' && ch <= 'F')
high = ch - 'A' + 10;
else if (ch >= 'a' && ch <= 'f')
high = ch - 'a' + 10;
else
break;
ch = *src_ptr++;
if (ch >= '0' && ch <= '9')
low = ch - '0';
else if (ch >= 'A' && ch <= 'F')
low = ch - 'A' + 10;
else if (ch >= 'a' && ch <= 'f')
low = ch - 'a' + 10;
else
break;
*dst_ptr++ = high * 16 + low; //hex
}
}
void str_to_hexstr(void *dst, const void *src, int src_len)
{
int i;
BYTE low, high, ch;
char *dst_ptr = (char *) dst;
const char *src_ptr = (const char *) src;
const char *str = "0123456789abcdef";
for (i = 0; i < src_len; i++)
{
ch = *src_ptr++;
high = (ch >> 4) & 0x0f;
low = ch & 0x0f;
*dst_ptr++ = str[high];
*dst_ptr++ = str[low];
}
}
int encode_blob(void *dst, int max_len, const void *src, int src_len)
{
int dst_len = 0;
// may be changed to base64 encode
if (max_len >= src_len * 2 + 1)
{
dst_len = src_len * 2;
str_to_hexstr(dst, src, src_len);
}
((BYTE *) dst)[dst_len] = 0x00;
return dst_len;
}
int decode_blob(void *dst, int max_len, const void *src)
{
// may be changed to base64 decode
int src_len, dst_len = 0;
src_len = strlen((char *) src);
if (max_len >= src_len / 2)
{
dst_len = src_len / 2;
hexstr_to_str(dst, src, src_len);
}
return dst_len;
}
std::string InsertRecord::now_to_string() {
// 获取系统时钟时间点
auto time_now = std::chrono::system_clock::now();
// 将时间点转成time_t类型的值 (秒)
std::time_t tt = std::chrono::system_clock::to_time_t(time_now);
// 转换成tm结构体形式
struct tm* ptm = gmtime(&tt); // 或者 localtime(&tt) 如果想要本地时间
// 计算出毫秒部分
long milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(
time_now.time_since_epoch()
).count() % 1000;
char buffer[40];
strftime(buffer, sizeof(buffer), "%Y-%m-%dT%H:%M:%S", ptm);
// 添加上毫秒和时区信息
sprintf(buffer + strlen(buffer), ".%03ld+0000", milliseconds);
return std::string(buffer);
}
time_t InsertRecord::string_to_now(std::string stamp) {
struct tm t;
std::istringstream ss(stamp);
ss >> std::get_time(&t,"%Y-%m-%dT%H:%M:%S");
if(ss.fail())return -1;
return mktime(&t);
}
int InsertRecord::SQLBatch_exc(std::vector<std::string> &sql_list,std::vector<std::string> &result){
sqlite3 *db = handpara;
char *errMsg = nullptr;
const char *sql;
int rc;
int resp = 0;
// 开启事务
if (sqlite3_exec(db, "BEGIN TRANSACTION;", nullptr, nullptr, &errMsg) != SQLITE_OK) {
std::cerr << "事务开启失败: " << errMsg << std::endl;
sqlite3_free(errMsg);
return -1;
}
// 执行批量插入操作
sqlite3_stmt *stmt;
int idx = 0;
for(const auto &k:sql_list){
std::cout << "cmd:"<<k.c_str()<<std::endl;
if (sqlite3_prepare_v2(db, k.c_str(), -1, &stmt, nullptr) != SQLITE_OK) {
std::cerr << "准备语句失败: " << sqlite3_errmsg(db) << std::endl;
continue;
}
rc = sqlite3_step(stmt);
if (rc != SQLITE_DONE) {
std::cerr << "插入失败: " << sqlite3_errmsg(db) << std::endl;
result[idx] = "-1";
}else
result[idx] = "0"; //SUCCESS
sqlite3_reset(stmt); // 重置语句
}
// 提交事务
if (sqlite3_exec(db, "COMMIT;", nullptr, nullptr, &errMsg) != SQLITE_OK) {
std::cerr << "事务提交失败: " << errMsg << std::endl;
sqlite3_free(errMsg);
resp = -1;
}
// 清理
sqlite3_finalize(stmt);
return reponse;
}
int InsertRecord::HandleMsg(const msg_t msg, handparam_t *param){
topic.clear();
topic_split(msg.topic,'/');
json value,resp;
int ret = 0;
int afn_flag = 0;
std::vector<std::string> sql_list;
std::vector<std::string> result;
sql_list.clear();
result.clear();
std::string table_name;
try{
value = json::prase(msg.payload)
}
catch(const std::expection &e){
std::cout << "topic "<<msg.topic.c_str()<<" prase error:"<<e.what();
return -1;//format fail
}
printf("insert request:%s \n",msg.payload.c_str());
if(!value.is_object())return -1;
if(value.contains("token")){
resp["token"] = value["token"];
}
if(value.contains("time")){
;// value["time"]
}
if(value.contains("data") && value["data"].contains("dataarray")){
json subvalue = value["data"]["dataarray"];
if(subvalue.is_array()){
int idx = 0;
for(const auto &k:subvalue){
std::stringstream ss;
if(k["afn"].compare("0D") != 0){
afn_flag = -1;
continue;
}
//根据afn/idclass 确定数据库表位置 待添加
if 1
ss << "insert into " <<table_name.c_str() <<" (idx,metid,dataid,frozentime,createdtime,source,crcaux,databuff) values ("\
<< 0 << ","\
<<std::stoi(k["metid"])<<","\
<<k["dataid"].c_str()<<","\
<<string_to_now(k["frozentime"])<<","\
<<time(NULL)<<","\
<<std::stoi(k["crcaux"])<<","\
<<k["databuff"]<<");";
#endif
sql_list.pushback(ss.str());
ss.clear();
}
}
}
int ret = SQLBatch_exc(sql_list,result);
resp["time"]=now_to_string();
if(ret == -1 || afn_flag == -1)
resp["ack"]="false";
else if(ret == 0)
resp["ack"]="true";
resp["data"]["dataidex"]=result;
msg_t resp_msg;
std::stringstream ss;
ss << resp_msg.topic[1].c_str()<<"/"<<resp_msg.topic[0].c_str()<<"/"\
<<resp_msg.topic[2].c_str()<<"/"<<resp_msg.topic[3].c_str();
resp_msg.topic = ss.str();
resp_msg.timestamp = now_to_string();
resp_msg.payload = resp.dump();
printf("resp:%s \n",resp_msg.payload.c_str());
return WriteResponseMsg(resp_msg);
}
const std::string InsertRecord::GetFID(){
}
void InsertRecord::topic_split(const std::string& str, char delimiter) {
std::string token; // 临时存储每个子字符串
std::istringstream tokenStream(str); // 使用字符串流处理输入字符串
// 按分隔符逐个读取子字符串
while (std::getline(tokenStream, token, delimiter)) {
topic.push_back(token); // 将子字符串添加到结果中
}
}
dataservice/F_InsertRecord.h
-28
View File
@@ -1,28 +0,0 @@
/*
* @Description: file content
* @Author: why
* @version:
* @Date: 2025-02-21 14:04:30
* @LastEditors: why
* @LastEditTime: 2025-02-25 11:01:33
*/
#include <string>
#include <vector>
#include "bizservice.h"
class InsertRecord::public IBizService
{
public:
void Init(initparam_t *param);
int HandleMsg(const msg_t msg, handparam_t *param = NULL);
const std::string GetFID();
InsertRecord();
~InsertRecord();
private:
handparam_t * handpara;
initparam_t * initpara;
std::vector<std::string> topic;//0：dst 1:src 2:sid 3:fid
void topic_split(const std::string& str, char delimiter);
InsertRecord::SQLBatch_exc(std::vector<std::string> &sql_list,std::vector<std::string> &result);
std::string InsertRecord::now_to_string();
time_t InsertRecord::string_to_now(std::string stamp);
};
dataservice/F_UpdateRecord.cpp
-193
View File
@@ -1,193 +0,0 @@
/*
* @Description: file content
* @Author: why
* @version:
* @Date: 2025-02-25 11:02:37
* @LastEditors: why
* @LastEditTime: 2025-02-25 16:33:06
*/
#include <iostream>
#include <chrono>
#include <iomanip>
#include <sstream>
#include <string>
#include <nlohmann/json.hpp>
#include "F_UpdateRecord.h"
using namespace std;
using json = nlohmann::json // for convenience
typedef struct topic{
std::string src;
std::string des;
std::string sid;
std::string fid;
} topic_type;
UpdateRecord::UpdateRecord(void){
handpara = nullptr;
initpara = nullptr;
}
UpdateRecord::~UpdateRecord(void){
}
std::string search_table_db(int afn,usignend init dataid){
std::string table;
return table;
}
void UpdateRecord::Init(initparam_t *param){
initpara = param;
}
std::string UpdateRecord::now_to_string() {
// 获取系统时钟时间点
auto time_now = std::chrono::system_clock::now();
// 将时间点转成time_t类型的值 (秒)
std::time_t tt = std::chrono::system_clock::to_time_t(time_now);
// 转换成tm结构体形式
struct tm* ptm = gmtime(&tt); // 或者 localtime(&tt) 如果想要本地时间
// 计算出毫秒部分
long milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(
time_now.time_since_epoch()
).count() % 1000;
char buffer[40];
strftime(buffer, sizeof(buffer), "%Y-%m-%dT%H:%M:%S", ptm);
// 添加上毫秒和时区信息
sprintf(buffer + strlen(buffer), ".%03ld+0000", milliseconds);
return std::string(buffer);
}
time_t UpdateRecord::string_to_now(std::string stamp) {
struct tm t;
std::istringstream ss(stamp);
ss >> std::get_time(&t,"%Y-%m-%dT%H:%M:%S");
if(ss.fail())return -1;
return mktime(&t);
}
int UpdateRecord::SQLBatch_exc(std::vector<std::string> &sql_list,std::vector<std::string> &result){
sqlite3 *db = handpara;
char *errMsg = nullptr;
const char *sql;
int rc;
int resp = 0;
// 开启事务
if (sqlite3_exec(db, "BEGIN TRANSACTION;", nullptr, nullptr, &errMsg) != SQLITE_OK) {
std::cerr << "事务开启失败: " << errMsg << std::endl;
sqlite3_free(errMsg);
return -1;
}
// 执行批量插入操作
sqlite3_stmt *stmt;
int idx = 0;
for(const auto &k:sql_list){
std::cout << "cmd:"<<k.c_str()<<std::endl;
if (sqlite3_prepare_v2(db, k.c_str(), -1, &stmt, nullptr) != SQLITE_OK) {
std::cerr << "准备语句失败: " << sqlite3_errmsg(db) << std::endl;
continue;
}
rc = sqlite3_step(stmt);
if (rc != SQLITE_DONE) {
std::cerr << "插入失败: " << sqlite3_errmsg(db) << std::endl;
result[idx] = "-1";
}else
result[idx] = "0"; //SUCCESS
sqlite3_reset(stmt); // 重置语句
}
// 提交事务
if (sqlite3_exec(db, "COMMIT;", nullptr, nullptr, &errMsg) != SQLITE_OK) {
std::cerr << "事务提交失败: " << errMsg << std::endl;
sqlite3_free(errMsg);
resp = -1;
}
// 清理
sqlite3_finalize(stmt);
return reponse;
}
int UpdateRecord::HandleMsg(const msg_t msg, handparam_t *param){
topic.clear();
topic_split(msg.topic,'/');
json value,resp;
int ret = 0;
int afn_flag = 0, update_flag = 0;;
std::vector<std::string> sql_list;
std::vector<std::string> result;
sql_list.clear();
result.clear();
std::string table_name ;
try{
value = json::prase(msg.payload)
}
catch(const std::expection &e){
std::cout << "topic "<<msg.topic.c_str()<<" prase error:"<<e.what();
return -1;//format fail
}
if(!value.is_object())return -1;
if(value.contains("token")){
resp["token"] = value["token"];
}
if(value.contains("time")){
;// value["time"]
}
if(value.contains("data")){
json subvalue = value["data"];
std::stringstream ss;
table_name =search_table_db(std::stoi(subvalue["afn"]),std::stoi(subvalue["dataid"]));
int afn = std::stoi(subvalue["afn"]);
if(!(afn==0x0d | afn==0x0e |afn == 0x12 |afn == 0x13))
{
afn_flag = -1;
goto resp_process;
}
if(std::stoi(subvalue["updatetype"] !=1){
update_flag = -1;
goto resp_process;
}
ss << "replace into " <<table_name.c_str() <<" (idx,metid,dataid,frozentime,createdtime,source,crcaux,databuff) values ("\
<< 0 << ","\
<<std::stoi(subvalue["metid"])<<","\
<<k["dataid"].c_str()<<","\
<<string_to_now(k["updatefilter"])<<","\
<<time(NULL)<<","\
<<std::stoi(k["crcaux"])<<","\
<<k["databuff"]<<");";
sql_list.pushback(ss.str());
ss.clear();
}
int ret = SQLBatch_exc(sql_list,result);
resp_process:
resp["time"]=now_to_string();
if(ret == -1 || afn_flag == -1 || update_flag = -1;)
resp["ack"]="false";
else if(ret == 0)
resp["ack"]="true";
resp["data"]["dataidex"]=result;
msg_t resp_msg;
std::stringstream ss;
ss << resp_msg.topic[1].c_str()<<"/"<<resp_msg.topic[0].c_str()<<"/"\
<<resp_msg.topic[2].c_str()<<"/"<<resp_msg.topic[3].c_str();
resp_msg.topic = ss.str();
resp_msg.timestamp = now_to_string();
resp_msg.payload = resp.dump();
return WriteResponseMsg(resp_msg);
}
const std::string UpdateRecord::GetFID(){
}
void UpdateRecord::topic_split(const std::string& str, char delimiter) {
std::string token; // 临时存储每个子字符串
std::istringstream tokenStream(str); // 使用字符串流处理输入字符串
// 按分隔符逐个读取子字符串
while (std::getline(tokenStream, token, delimiter)) {
topic.push_back(token); // 将子字符串添加到结果中
}
}
dataservice/F_UpdateRecord.h
-26
View File
@@ -1,26 +0,0 @@
/*
* @Description: file content
* @Author: why
* @version:
* @Date: 2025-02-25 11:03:17
* @LastEditors: why
* @LastEditTime: 2025-02-26 08:09:35
*/
#include <string>
#include <vector>
#include "bizservice.h"
class UpdateRecord::public IBizService {
public:
void Init(initparam_t *param);
int HandleMsg(const msg_t msg, handparam_t *param = NULL);
const std::string GetFID();
UpdateRecord();
~UpdateRecord();
private:
handparam_t * handpara;
initparam_t * initpara;
std::vector<std::string> topic;//0：dst 1:src 2:sid 3:fid
void topic_split(const std::string& str, char delimiter);
InsertRecord::SQLBatch_exc(std::vector<std::string> &sql_list,std::vector<std::string> &result);
std::string InsertRecord::now_to_string();
}
dataservice/bizservice.h
-63
View File
@@ -1,63 +0,0 @@
/**
* @brief 使用方法
* 1、设计类时请继承于IBizService类
* 2、在HandleMsg中将结果信息通过WriteResponseMsg进行输出
*/
#ifndef _DCAPP_CENTER_INTERFACE_H_
#define _DCAPP_CENTER_INTERFACE_H_
#include <string>
typedef struct
{
void *pDbHandle; // 数据库句柄参数
} handparam_t;
typedef struct
{
void *resove;
} initparam_t; // 其他参数
typedef struct msg
{
std::string topic; // 消息主题
std::string payload; // 消息内容
long timestamp; // 消息时间戳
} msg_t; // 消息体结构
/**
* @brief 指令响应(将响应的信息体进行输出)
*
* @param msg //待输出信息
* @return int 是否发送成功 0:成功 -1:失败
*/
int WriteResponseMsg(const msg_t &msg);
class IBizService
{
public:
/**
* @brief 初始化
*
* @param param 输入初始化参数
*/
virtual void Init(initparam_t *param = NULL) = 0;
/**
* @brief 业务处理
*
* @param msg 输入消息体
* @param param 可用数据库连接句柄
* @return int 返回值 成功为0 失败<0 (-1待处理数据格式错误 -2操作数据库失败 -3其它错误)
*/
virtual int HandleMsg(const msg_t msg, handparam_t *param = NULL) = 0; // 业务处理
/**
* @brief 获取FID服务标识
*
* @return FID服务标识
*/
virtual const std::string GetFID() = 0; /* 如"F-GetVersion" */
};
#endif
dataservice/json.hpp
-25447
View File
File diff suppressed because it is too large Load Diff