From 440137cac23d8e3a8a8ad1a184ef7fe18ff8d2ab Mon Sep 17 00:00:00 2001 From: HwangKC Date: Fri, 15 Nov 2024 18:22:43 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E6=88=90=E4=B8=8A=E4=BC=A0=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E6=B6=88=E6=81=AF=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- applications/ems_datahubs/WinDataHubs.vcxproj | 6 +- applications/ems_datahubs/eventhandler.cpp | 69 +++++++++++++++++-- applications/ems_datahubs/eventhandler.h | 1 + applications/ems_datahubs/frame_define.h | 11 ++- applications/ems_datahubs/iconv-utils.cpp | 4 +- applications/ems_datahubs/main.cpp | 54 +++++++++++++-- applications/ems_datahubs/opmysql.cpp | 37 +++++++++- applications/ems_datahubs/opmysql.h | 23 +++++-- 8 files changed, 185 insertions(+), 20 deletions(-) diff --git a/applications/ems_datahubs/WinDataHubs.vcxproj b/applications/ems_datahubs/WinDataHubs.vcxproj index a8a305c..c2fb411 100644 --- a/applications/ems_datahubs/WinDataHubs.vcxproj +++ b/applications/ems_datahubs/WinDataHubs.vcxproj @@ -120,11 +120,15 @@ true _DEBUG;_CONSOLE;_CRT_SECURE_NO_WARNINGS;%(PreprocessorDefinitions) true + D:\Workspace\EmsProjects\emsApplication\sdk\mysql\include;%(AdditionalIncludeDirectories) Console true - hv.lib;zlibd.lib;mysqlcppconn.lib;libiconvD.lib;%(AdditionalDependencies) + hv.lib;zlibd.lib;libmysql.lib;mysqlclient.lib;libiconvD.lib;%(AdditionalDependencies) + + + D:\Workspace\EmsProjects\emsApplication\sdk\mysql\lib\debug;%(AdditionalLibraryDirectories) diff --git a/applications/ems_datahubs/eventhandler.cpp b/applications/ems_datahubs/eventhandler.cpp index 92882c3..8cb6ee7 100644 --- a/applications/ems_datahubs/eventhandler.cpp +++ b/applications/ems_datahubs/eventhandler.cpp @@ -1,5 +1,7 @@ #include "eventhandler.h" +#include + #include #include #include @@ -37,7 +39,8 @@ void EventHandler::onRecvHandler(hio_t* io, void* buf, int readbytes) SOCKADDR_STR(hio_peeraddr(io), peeraddrstr)); MessageFrame respFrame; - if( readbytes > 0xFFFF - 1 ) + + if( readbytes > FRAME_MAX_LENGTH - 1 ) //限制1Mb 字节 { hloge("too large data buffer to process: %d", readbytes); respFrame.setErrorFrame(ERR_INVALID_BUF_LEN); @@ -45,12 +48,12 @@ void EventHandler::onRecvHandler(hio_t* io, void* buf, int readbytes) return; } - hlogi("<=== decode OK [\n%s\n]", printHex(buf, readbytes).c_str()); + hlogi("<=== decode OK [\n%s\n]", printHex(buf, MIN(readbytes,512)).c_str()); MessageFrame* pReadFrame = (MessageFrame*)buf; hlogi("on_recv fd=%d frame_len=%d [0x%x] ", hio_fd(io), pReadFrame->frame_len, pReadFrame->frame_len); - if( pReadFrame->frame_len > 0xFFFF - 1 ) + if( pReadFrame->frame_len > FRAME_MAX_LENGTH - 1 ) { hloge("too big string buffer to process: %d, it should be less than %d", pReadFrame->frame_len, 0xFFFF); respFrame.setErrorFrame(ERR_INVALID_LEN); @@ -58,13 +61,36 @@ void EventHandler::onRecvHandler(hio_t* io, void* buf, int readbytes) return; } - if( Frame_DeviceData_Request == pReadFrame->frame_type ) + switch (pReadFrame->frame_type) + { + case Frame_DeviceData_Request: { handleGatherData(io, buf, readbytes); + break; } - else + case Frame_Echo_Request: + { + hlogd("<== recieve test echo request and response signal back!"); + hio_write(io, (void*)&respFrame, sizeof respFrame); + break; + } + case Frame_Configure_DB_Request: + { + hlogd("<== recieve configuration file upload request!"); + int ret = handleConfigureData(io, buf, readbytes); + + if(ret!=0) + respFrame.setErrorFrame((ErrorCode)ret); + + hio_write(io, (void*)&respFrame, sizeof respFrame); + + break; + } + default: { assert(false); + break; + } } } @@ -246,4 +272,37 @@ void EventHandler::handleGatherData(hio_t* io, void* buf, int readbytes) hio_write(io, (void*)&respFrame, sizeof respFrame); return; } +} + +int EventHandler::handleConfigureData(hio_t* io, void* buf, int readbytes) +{ + assert(buf); + + __USING_NAMESPACE_HJ__; + + MessageFrame respFrame; + MessageFrame* pReadFrame = (MessageFrame*)buf; + + //保存文件 + std::string filename("d://test_1.zip"); + std::ofstream file(filename, std::ios::binary); + + if (!file) + { + hloge("failed to create file: %s",filename.c_str()); + return ERR_PERMISSION_DENIED; + } + // 将 pdata 中的内容写入文件 + file.write((const char*)(pReadFrame->frame_content), pReadFrame->frame_len); + + if (!file) + { + hloge("failed to write to file: %s", filename.c_str()); + return ERR_RETRANS_CONTENT; + } + else + { + hlogd("successfuly to write to file: %s", filename.c_str()); + return ErrorCode::ERR_OK; + } } \ No newline at end of file diff --git a/applications/ems_datahubs/eventhandler.h b/applications/ems_datahubs/eventhandler.h index 0f8be18..4ef694a 100644 --- a/applications/ems_datahubs/eventhandler.h +++ b/applications/ems_datahubs/eventhandler.h @@ -14,5 +14,6 @@ public: protected: //处理采集程序上传的数据，以JSON数据上报 static void handleGatherData(hio_t* io, void* buf, int readbytes); + static int handleConfigureData(hio_t* io, void* buf, int readbytes); }; diff --git a/applications/ems_datahubs/frame_define.h b/applications/ems_datahubs/frame_define.h index c0cbe62..2635915 100644 --- a/applications/ems_datahubs/frame_define.h +++ b/applications/ems_datahubs/frame_define.h @@ -7,7 +7,7 @@ __NAMESPACE_BEGIN__(HJ) #define FRAME_HEADER_LENGTH (5) #define FRAME_TAILE_LENGTH (4) - +#define FRAME_MAX_LENGTH (0x100000) //限制1Mb 字节 typedef enum tagErrorCode : unsigned char { ERR_OK = 0X00, @@ -16,6 +16,9 @@ typedef enum tagErrorCode : unsigned char ERR_INVALID_BUF_LEN = 0X03, ERR_INVALID_JSON_FMT = 0X04, ERR_INVALID_FSUCODE = 0X05, + ERR_INVALID_CFG_CONTENT = 0X06, // 文件内容错误 + ERR_RETRANS_CONTENT = 0X07, // 重传文件内容 + ERR_PERMISSION_DENIED = 0X08, //写文件权限不够 ERR_UNKOWN = 0XFF }ErrorCode; @@ -23,7 +26,9 @@ typedef enum tagFrameType : unsigned char { Frame_Response = 0x00, //返回帧 Frame_Request = 0x01, //请求帧 - Frame_DeviceData_Request = 0x02, //来自采集程序的数据请求包，将数据保存到数据库中 + Frame_Echo_Request = 0x02, //测试请求帧 + Frame_DeviceData_Request = 0x03, //来自采集程序的数据请求包，将数据保存到数据库中 + Frame_Configure_DB_Request = 0x04, //上传配置文件 }FrameType; typedef struct tagFrameTail @@ -31,6 +36,8 @@ typedef struct tagFrameTail unsigned char frame_delimiter[4] = { 0xEE,0xFF,0xEE,0xFF }; }FrameTail; +#define FRAME_HEADER_LENGTH (5) // frame_type(1) + frame_len(4) + typedef struct tagFrame { FrameType frame_type; //帧类型 diff --git a/applications/ems_datahubs/iconv-utils.cpp b/applications/ems_datahubs/iconv-utils.cpp index fc07941..34b9b23 100644 --- a/applications/ems_datahubs/iconv-utils.cpp +++ b/applications/ems_datahubs/iconv-utils.cpp @@ -155,7 +155,7 @@ int g2u(char* inbuf, size_t inlen, char* outbuf, size_t outlen) bool GBKToUTF8(const std::string& strGBK,std::string& str_result) { - int length = strGBK.size() * 2 + 1; + size_t length = strGBK.size() * 2 + 1; char* temp = (char*)malloc(sizeof(char) * length); @@ -175,7 +175,7 @@ bool GBKToUTF8(const std::string& strGBK,std::string& str_result) bool UTFtoGBK(const char* utf8, std::string& str_result) { - int length = strlen(utf8); + size_t length = strlen(utf8); char* temp = (char*)malloc(sizeof(char) * length); diff --git a/applications/ems_datahubs/main.cpp b/applications/ems_datahubs/main.cpp index fcbf455..ccc66cd 100644 --- a/applications/ems_datahubs/main.cpp +++ b/applications/ems_datahubs/main.cpp @@ -118,6 +118,7 @@ void print_help() int parse_confile(const char* confile) { +#ifdef __linux__ int ret = g_conf_ctx.parser->LoadFromFile(confile); if (ret != 0) { @@ -158,25 +159,26 @@ int parse_confile(const char* confile) logger_enable_fsync(hlog, hv_getboolean(str.c_str())); } #endif - +#endif // first log here hlogi("=========--- Welcome to the Earth ---========="); hlogi("%s version: %s", g_main_ctx.program_name, K22_VERSION); hlog_fsync(); -#if 0 +#if 1 g_conf_ctx.worker_processes = 1; g_conf_ctx.worker_threads = 1; g_conf_ctx.host = "0.0.0.0"; g_conf_ctx.port = 44242; g_conf_ctx.dbname = "hjems"; g_conf_ctx.dbuser = "root"; - g_conf_ctx.dbserver = "tcp://127.0.0.1:3306"; + g_conf_ctx.dbserver = "127.0.0.1"; #endif // worker_processes int worker_processes = 0; -#if 1 + +#if __linux__ #ifdef DEBUG // Disable multi-processes mode for debugging worker_processes = 0; @@ -338,7 +340,48 @@ int already_running(void) #endif /// end of checking only one instance /// //////////////////////////////////////////////// +#ifdef _WIN32 +int main(int argc, char** argv) +{ + // g_main_ctx + main_ctx_init(argc, argv); + conf_ctx_init(&g_conf_ctx); + // first log here + hlog_set_level_by_str("DEBUG"); + hlogi("=========--- Welcome to the Earth ---========="); + hlogi("%s version: %s", g_main_ctx.program_name, K22_VERSION); + hlog_fsync(); + +#if 1 + g_conf_ctx.worker_processes = 1; + g_conf_ctx.worker_threads = 1; + g_conf_ctx.host = "0.0.0.0"; + g_conf_ctx.port = 44242; + g_conf_ctx.dbname = "hjems"; + g_conf_ctx.dbuser = "root"; + g_conf_ctx.dbserver = "127.0.0.1"; +#endif + +#if TEST_UNPACK + memset(&unpack_setting, 0, sizeof(unpack_setting_t)); + unpack_setting.package_max_length = DEFAULT_PACKAGE_MAX_LENGTH; + unpack_setting.mode = UNPACK_BY_DELIMITER; + unpack_setting.delimiter[0] = 0xEE; + unpack_setting.delimiter[1] = 0xFF; + unpack_setting.delimiter[2] = 0xEE; + unpack_setting.delimiter[3] = 0xFF; + unpack_setting.delimiter_bytes = 4; +#endif + + master_workers_run(worker_fn, (void*)(intptr_t)&g_conf_ctx, g_conf_ctx.worker_processes, g_conf_ctx.worker_threads); + + hlogi("=========--- I'll be back! ---========="); + + return 0; +} + +#else int main(int argc, char** argv) { #if 0 @@ -453,6 +496,7 @@ int main(int argc, char** argv) return 0; } +#endif void worker_fn(void* userdata) { @@ -460,7 +504,7 @@ void worker_fn(void* userdata) long port = ptrCtx->port; //initialize database connection - bool dbok = OpDatabase::getInstance()->OpenDatabase(ptrCtx->dbserver,ptrCtx->dbuser,ptrCtx->dbname); + bool dbok = OpDatabase::getInstance()->OpenDatabase(ptrCtx->dbserver,3306,ptrCtx->dbuser,"Hj57471000",ptrCtx->dbname); if (!dbok) { hloge("failed to open database, exit now..."); diff --git a/applications/ems_datahubs/opmysql.cpp b/applications/ems_datahubs/opmysql.cpp index a0c249f..be7170f 100644 --- a/applications/ems_datahubs/opmysql.cpp +++ b/applications/ems_datahubs/opmysql.cpp @@ -1,10 +1,14 @@ #include "opmysql.h" + +#ifndef _USING_MYSQL_51_LIB_ #include #include #include #include #include #include +#endif + #include #include #include "openjson.h" @@ -26,8 +30,9 @@ OpDatabase* OpDatabase::getInstance() return &m_instance; } -bool OpDatabase::OpenDatabase(const std::string& server, const std::string& dbuser, const std::string& database) +bool OpDatabase::OpenDatabase(const std::string& server /*= "127.0.0.1"*/, int dbport /*= 3306*/, const std::string& dbuser/* = "root"*/, const std::string& dbpasswd /*= "Hj57471000"*/, const std::string& database /*= "hjems"*/) { +#ifndef _USING_MYSQL_51_LIB_ // 打开数据库 try { @@ -64,18 +69,46 @@ bool OpDatabase::OpenDatabase(const std::string& server, const std::string& dbus hloge("Failed to connect to database: %s",ss.str().c_str()); return false; } +#else + bool ok = false; + m_pDbConnection = mysql_init(NULL); + //连接到MySQL服务器 + if (!mysql_real_connect(m_pDbConnection, server.c_str(), dbuser.c_str(), dbpasswd.c_str(), database.c_str(), dbport, NULL, 0)) + { + hloge("MySQL connection error: %s", mysql_error(m_pDbConnection)); + } + else + { + ok = true; + mysql_set_character_set(m_pDbConnection, "utf8"); + } + if (!ok) + { + hloge("Failed to connect to database."); + } + else + { + hlogd("new database connection created successfully!"); + } + return ok; +#endif } // 关闭数据库连接 void OpDatabase::CloseDatabase() { +#ifdef _USING_MYSQL_51_LIB_ + mysql_close(m_pDbConnection); +#else m_pDbConnection->close(); //delete m_pDbConnection; +#endif } void OpDatabase::InsertMessage(const std::string& ts, const std::string& msg_type, const std::string& fsu, const std::string& content, int topic, int dev_id) { +#ifndef _USING_MYSQL_51_LIB_ const char* insertSql = "INSERT INTO tbl_data (data_timestamp, data_type, FsuCode, data_content, topic, device_id) VALUES (?,?,?,?,?,?);"; // 创建预编译的prepared statement std::unique_ptr insertStmt(m_pDbConnection->prepareStatement(insertSql)); @@ -100,4 +133,6 @@ void OpDatabase::InsertMessage(const std::string& ts, const std::string& msg_typ { hloge( "No rows affected during insert operation." ); } +#else +#endif } diff --git a/applications/ems_datahubs/opmysql.h b/applications/ems_datahubs/opmysql.h index c258500..adbf765 100644 --- a/applications/ems_datahubs/opmysql.h +++ b/applications/ems_datahubs/opmysql.h @@ -2,9 +2,18 @@ #define __MY_OPERATE_MYSQL__ #include -#include +#ifdef _WIN32 +# define _USING_MYSQL_51_LIB_ +# include +# include //MySQL C API include file +#else +# ifdef __linux__ +# include +# endif +#endif + class OpDatabase { protected: @@ -14,12 +23,18 @@ public: static OpDatabase* getInstance(); void CloseDatabase(); - bool OpenDatabase(const std::string& server = "tcp://127.0.0.1:3306", const std::string& dbuser = "root", const std::string& database="hjems"); + bool OpenDatabase(const std::string& server = "127.0.0.1", int dbport = 3306, const std::string& dbuser = "root", const std::string& dbpasswd = "Hj57471000", const std::string& database = "hjems"); + void InsertMessage(const std::string& ts, const std::string& msg_type, const std::string& fsu, const std::string& content, int topic, int dev_id); protected: - //std::unique_ptr m_pDbConnection; - sql::Connection* m_pDbConnection; +#ifdef _WIN32 + MYSQL* m_pDbConnection; +#else +# ifdef __linux__ + sql::Connection* m_pDbConnection; +# endif +#endif private: static OpDatabase m_instance; };