更改为支持mysql

applications/ems_datahubs/Makefile

@@ -13,7 +13,7 @@ TARGET  := datahubs
 #compile and lib parameter
 #编译参数
 CC      := g++
-LIBS    := -L/usr/local/lib -lhv -lsqlite3 -lopdb -lpthread -liconv -ldl -lz
+LIBS    := -L/usr/local/lib -lhv -lmysqlcppconn -lpthread -liconv -ldl -lz
 LDFLAGS :=
 DEFINES :=
 INCLUDE := -I. -I/usr/local/include

applications/ems_datahubs/Makefile.arm

@@ -14,7 +14,7 @@ TARGET  := datahubs.arm
 #编译参数
 CC      := arm-g++
 CXX     := arm-g++
-LIBS    := -L/usr/local/arm/lib -lhv -lsqlite3 -lopdb -lpthread -liconv -ldl -lz
+LIBS    := -L/usr/local/arm/lib -lhv -lmysqlcppconn -lpthread -liconv -ldl -lz
 LDFLAGS :=
 DEFINES :=
 INCLUDE := -I. -I/usr/local/arm/include

applications/ems_datahubs/datahubs.conf

@@ -13,4 +13,6 @@ worker_threads = auto
 host = 0.0.0.0
 port = 44242
 
-dbfile = /home/hkc/workspace/projects/datahubs/ems.db
+database = hjems
+dbuser = root
+dbserver = tcp://127.0.0.1:3306

applications/ems_datahubs/main.cpp

@@ -16,8 +16,8 @@
 #include <zlib/zlib.h>
 
 #include "mqtt_msg.h"
-#include <openjson.h>
+#include "openjson.h"
-#include <opdatabase.h>
+#include "opmysql.h"
 
 #include "iconv-utils.h"
 
@@ -181,7 +181,7 @@ int DecompressString(const char* in_str, size_t in_len, std::string& out_str)
 }
 
 
-static OpDatabase gOpDatabase;
+//static OpDatabase gOpDatabase;
 
 #if TEST_UNPACK
 static unpack_setting_t unpack_setting;
@@ -208,7 +208,9 @@ typedef struct conf_ctx_s
     int worker_threads;
     std::string host;
     int port;
-    std::string db_file;
+    std::string dbserver;
+    std::string dbuser;
+    std::string dbname;
 } conf_ctx_t;
 conf_ctx_t g_conf_ctx;
 
@@ -219,6 +221,9 @@ inline void conf_ctx_init(conf_ctx_t* ctx)
     ctx->worker_processes = 0;
     ctx->worker_threads = 0;
     ctx->port = 0;
+    ctx->dbname = "hjems";
+    ctx->dbuser = "root";
+    ctx->dbserver = "tcp://127.0.0.1:3306";
 }
 
 static void print_version();
@@ -380,17 +385,41 @@ int parse_confile(const char* confile)
 
     g_conf_ctx.port = port;
 
-    str = g_conf_ctx.parser->GetValue("dbfile");
+    str = g_conf_ctx.parser->GetValue("database");
     if (str.size() != 0)
     {
-        g_conf_ctx.db_file = str;
+        g_conf_ctx.dbname = str;
     }
     else
     {
-        g_conf_ctx.db_file = "./ems.db";
+        g_conf_ctx.dbname = "hjems";
     }
 
-    hlogi("database = ('%s')", g_conf_ctx.db_file.c_str());
+    hlogi("database = ('%s')", g_conf_ctx.dbname.c_str());
+
+	str = g_conf_ctx.parser->GetValue("dbuser");
+	if (str.size() != 0)
+	{
+		g_conf_ctx.dbuser = str;
+	}
+	else
+	{
+		g_conf_ctx.dbuser = "root";
+	}
+
+	hlogi("dbuser = ('%s')", g_conf_ctx.dbuser.c_str());
+
+	str = g_conf_ctx.parser->GetValue("dbserver");
+	if (str.size() != 0)
+	{
+		g_conf_ctx.dbserver = str;
+	}
+	else
+	{
+		g_conf_ctx.dbserver = "tcp://127.0.0.1:3306";
+	}
+
+	hlogi("dbserver = ('%s')", g_conf_ctx.dbserver.c_str());
 
     hlogi("parse_confile('%s') OK", confile);
     return 0;
@@ -809,7 +838,7 @@ static void on_recv(hio_t* io, void* buf, int readbytes)
         return;
     }
 
-    hlogd("<<<<Compress result: string size from [%d] to [%d]", len, out_compress.size());
+    hlogd("<<<<Compress result: string size from original [%d] to [%d]", len, out_compress.size());
 
 #ifdef _DEBUG
     hlogd("<<<<Compress string here>>>> \n[\n%s\n]\n", printHex(out_compress.c_str(), out_compress.size()).c_str());
@@ -834,7 +863,7 @@ static void on_recv(hio_t* io, void* buf, int readbytes)
         auto& pNode = IdCodeContent[0]; //这是只解析第一个节点
         std::string oid = pNode["OID"].s();
 
-        gOpDatabase.InsertMessage(timestamp, msg_type, fsucode, out_compress, (int)pData->mqtt_topic, (int)pData->device_id);
+        OpDatabase::getInstance()->InsertMessage(timestamp, msg_type, fsucode, out_compress, (int)pData->mqtt_topic, (int)pData->device_id);
     }
 
     if (msg_type == "web-read")
@@ -850,7 +879,7 @@ static void on_recv(hio_t* io, void* buf, int readbytes)
         auto& pNode = IdCodeContent[0]; //这是只解析第一个节点
         std::string oid = pNode.s();
 
-        gOpDatabase.InsertMessage(timestamp, msg_type, fsucode, out_compress, (int)pData->mqtt_topic, (int)pData->device_id);
+        OpDatabase::getInstance()->InsertMessage(timestamp, msg_type, fsucode, out_compress, (int)pData->mqtt_topic, (int)pData->device_id);
     }
     //delete[] pTmp;
 }
@@ -879,11 +908,10 @@ static void on_accept(hio_t* io)
 void worker_fn(void* userdata)
 {
     conf_ctx_t* ptrCtx = (conf_ctx_t*)(intptr_t)(userdata);
-    //long port = (long)(intptr_t)(userdata);
     long port = ptrCtx->port;
 
     //initialize database connection
-    bool dbok = gOpDatabase.OpenDatabase(ptrCtx->db_file);
+    bool dbok = OpDatabase::getInstance()->OpenDatabase(ptrCtx->dbserver,ptrCtx->dbuser,ptrCtx->dbname);
     if (!dbok)
     {
         hloge("failed to open database, exit now...");
@@ -905,5 +933,9 @@ void worker_fn(void* userdata)
     hlogi("port=%ld pid=%ld tid=%ld listenfd=%d", port, hv_getpid(), hv_gettid(), hio_fd(listenio));
 
     hloop_run(loop);
+
+    hlogi("database connection close!");
+    OpDatabase::getInstance()->CloseDatabase();
+
     hloop_free(&loop);
 }

applications/ems_datahubs/openjson.h

@@ -0,0 +1,276 @@
+/***************************************************************************
+ * Copyright (C) 2023-, openlinyou, <linyouhappy@outlook.com>
+ *
+ * You may opt to use, copy, modify, merge, publish, distribute and/or sell
+ * copies of the Software, and permit persons to whom the Software is
+ * furnished to do so, under the terms of the COPYING file.
+ ***************************************************************************/
+
+#ifndef HEADER_OPEN_JSON_H
+#define HEADER_OPEN_JSON_H
+
+#include <string>
+#include <stddef.h>
+#include <stdint.h>
+#include <vector>
+
+
+#ifdef _MSC_VER
+#pragma warning( disable: 4251 )
+#   if _MSC_VER >= 1600 || defined(__MINGW32__)
+#   else
+#       if (_MSC_VER < 1300)
+typedef signed char       int8_t;
+typedef signed short      int16_t;
+typedef signed int        int32_t;
+typedef unsigned char     uint8_t;
+typedef unsigned short    uint16_t;
+typedef unsigned int      uint32_t;
+#       else
+typedef signed __int8     int8_t;
+typedef signed __int16    int16_t;
+typedef signed __int32    int32_t;
+typedef unsigned __int8   uint8_t;
+typedef unsigned __int16  uint16_t;
+typedef unsigned __int32  uint32_t;
+#       endif
+typedef signed __int64       int64_t;
+typedef unsigned __int64     uint64_t;
+#   endif
+#endif // _MSC_VER
+
+//#pragma warning(disable:4100)
+
+class OpenJson
+{
+    enum JsonType
+    {
+        EMPTY = 0,
+        STRING,
+        NUMBER,
+        OBJECT,
+        ARRAY,
+        UNKNOWN
+    };
+    class Box
+    {
+        std::vector<OpenJson*> childs_;
+    public:
+        Box();
+        ~Box();
+        inline void clear()
+        {
+            childs_.clear();
+        }
+        inline bool empty()
+        {
+            return childs_.empty();
+        }
+        inline size_t size()
+        {
+            return childs_.size();
+        }
+        inline void add(OpenJson* node)
+        {
+            childs_.push_back(node);
+        }
+        inline OpenJson* operator[](size_t idx)
+        {
+            return childs_[idx];
+        }
+        bool remove(OpenJson* node);
+        friend class OpenJson;
+    };
+    class Context
+    {
+        char* data_;
+        size_t size_;
+        size_t offset_;
+
+        OpenJson* root_;
+        std::string rbuffer_;
+        std::string wbuffer_;
+
+        std::string stringNull_;
+    public:
+        Context();
+        ~Context();
+        void startRead();
+        void startWrite();
+        friend class OpenJson;
+    };
+    class Segment
+    {
+    public:
+        enum SegmentType
+        {
+            NIL = 0,
+            BOOL,
+            INT32,
+            INT64,
+            DOUBLE,
+            STRING
+        };
+        SegmentType type_;
+        std::string content_;
+        union
+        {
+            bool bool_;
+            int32_t int32_;
+            int64_t int64_;
+            double double_;
+        } value_;
+
+        Segment(SegmentType type = NIL);
+        ~Segment();
+
+        void clear();
+        void toString();
+        void setType(SegmentType type);
+    };
+
+    JsonType type_;
+    Context* context_;
+    Context* wcontext_;
+    size_t idx_;
+    size_t len_;
+    Box* box_;
+    OpenJson* key_;
+    Segment* segment_;
+
+    void trimSpace();
+    bool makeRContext();
+    OpenJson* createNode(unsigned char code);
+    JsonType codeToType(unsigned char code);
+    unsigned char getCharCode();
+    unsigned char getChar();
+    unsigned char checkCode(unsigned char charCode);
+    size_t searchCode(unsigned char charCode);
+    void throwError(const char* errMsg);
+
+    const char* data();
+    int32_t stringToInt32();
+    int64_t stringToInt64();
+    double stringToDouble();
+
+    OpenJson& array(size_t idx);
+    OpenJson& object(const char* key);
+    void addNode(OpenJson* node);
+    void removeNode(size_t idx);
+    void removeNode(const char* key);
+    OpenJson(OpenJson& /*json*/) {}
+    OpenJson(const OpenJson& /*json*/) {}
+    void operator=(OpenJson& /*json*/) {}
+    void operator=(const OpenJson& /*json*/) {}
+
+    const std::string& emptyString();
+    static OpenJson NodeNull;
+    static std::string StringNull;
+public:
+    OpenJson(JsonType type = EMPTY);
+    ~OpenJson();
+
+    inline size_t size()
+    {
+        return box_ ? box_->size() : 0;
+    }
+    inline bool empty()
+    {
+        return box_ ? box_->empty() : true;
+    }
+    inline OpenJson& operator[] (int idx)
+    {
+        return array(idx);
+    }
+    inline OpenJson& operator[] (size_t idx)
+    {
+        return array(idx);
+    }
+    inline OpenJson& operator[] (const char* key)
+    {
+        return object(key);
+    }
+    inline OpenJson& operator[] (const std::string& key)
+    {
+        return object(key.c_str());
+    }
+
+    inline void remove(int idx)
+    {
+        removeNode(idx);
+    }
+    inline void remove(size_t idx)
+    {
+        removeNode(idx);
+    }
+    inline void remove(const char* key)
+    {
+        removeNode(key);
+    }
+    inline void remove(const std::string& key)
+    {
+        removeNode(key.c_str());
+    }
+    void clear();
+
+    inline bool isNull()
+    {
+        return type_ == EMPTY;
+    }
+    inline bool isNumber()
+    {
+        return type_ == NUMBER;
+    }
+    inline bool isString()
+    {
+        return type_ == STRING;
+    }
+    inline bool isObject()
+    {
+        return type_ == OBJECT;
+    }
+    inline bool isArray()
+    {
+        return type_ == ARRAY;
+    }
+
+    bool b(bool def = false);
+    int32_t i32(int32_t def = 0);
+    int64_t i64(int64_t def = 0);
+    double d(double def = 0);
+    const std::string& s();
+    const std::string& key();
+
+    void operator=(bool val);
+    void operator=(int32_t val);
+    void operator=(uint32_t val);
+    void operator=(int64_t val);
+    void operator=(uint64_t val);
+    void operator=(double val);
+    void operator=(const char* val);
+    void operator=(const std::string& val);
+
+    bool decode(const std::string& buffer);
+    bool decodeFile(const std::string& filePath);
+    const std::string& encode();
+    void encodeFile(const std::string& filePath);
+
+    static void EnableLog(bool enable);
+private:
+    static bool EnableLog_;
+    static void Log(const char* format, ...);
+    void read(Context* context, bool isRoot = false);
+    void readNumber();
+    void readString();
+    void readObject();
+    void readArray();
+
+    void write(Context* context, bool isRoot = false);
+    void writeNumber();
+    void writeString();
+    void writeObject();
+    void writeArray();
+};
+
+
+#endif /* HEADER_OPEN_JSON_H */

applications/ems_datahubs/opmysql.cpp

@@ -0,0 +1,102 @@
+#include "opmysql.h"
+#include <mysql_driver.h>
+#include <mysql_connection.h>
+#include <cppconn/prepared_statement.h>
+#include <cppconn/resultset.h>
+#include <cppconn/statement.h>
+#include <cppconn/exception.h>
+#include <hv/hlog.h>
+#include <sstream>  
+#include "openjson.h"
+
+OpDatabase OpDatabase::m_instance;
+
+OpDatabase::OpDatabase()
+	:m_pDbConnection(nullptr)
+{
+}
+
+OpDatabase::~OpDatabase()
+{
+	CloseDatabase();
+}
+
+OpDatabase* OpDatabase::getInstance()
+{
+	return &m_instance;
+}
+
+bool OpDatabase::OpenDatabase(const std::string& server, const std::string& dbuser, const std::string& database)
+{
+	// 打开数据库
+	try
+	{
+		// 数据库连接配置
+		//std::string server = "tcp://127.0.0.1:3306";
+		//std::string dbuser = "root";
+		std::string password = "Hj57471000";
+		//std::string database = "hjems";
+
+		// 创建连接
+		sql::mysql::MySQL_Driver* driver;
+		driver = sql::mysql::get_mysql_driver_instance();
+		//m_pDbConnection.reset(driver->connect(server, dbuser, password));
+		m_pDbConnection = driver->connect(server, dbuser, password);
+
+		if (!m_pDbConnection)
+		{
+			hloge("Failed to connect to database.");
+			return false;
+		}
+
+		// 设置为使用指定数据库
+		m_pDbConnection->setSchema(database);
+
+		return true;
+	}
+	catch (sql::SQLException& e)
+	{
+		std::ostringstream ss;
+		ss << "SQLException: " << e.what();
+		ss<< " (MySQL error code: " << e.getErrorCode();
+		ss << ", SQLState: " << e.getSQLState() << " )";
+		hloge("Failed to connect to database: %s",ss.str().c_str());
+		return false;
+	}
+}
+
+// 关闭数据库连接
+void OpDatabase::CloseDatabase()
+{
+	//m_pDbConnection->close();
+	//delete m_pDbConnection;
+}
+
+
+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)
+{
+	const char* insertSql = "INSERT INTO tbl_data (data_timestamp, data_type, FsuCode, data_content, topic, device_id) VALUES (?,?,?,?,?,?);";
+	// 创建预编译的prepared statement
+	std::unique_ptr<sql::PreparedStatement> insertStmt(m_pDbConnection->prepareStatement(insertSql));
+
+	// 绑定插入数据的参数
+	insertStmt->setString(1, ts);
+	insertStmt->setString(2, msg_type);
+	insertStmt->setString(3, fsu);
+	insertStmt->setString(4, content);
+	insertStmt->setInt(5, topic);
+	insertStmt->setInt(6, dev_id);
+
+	// 执行插入操作
+	int rowsAffected = insertStmt->executeUpdate();
+
+	// 检查插入操作是否成功
+	if (rowsAffected > 0)
+	{
+		hlogi( "New message inserted successfully." );
+	}
+	else
+	{
+		hloge( "No rows affected during insert operation." );
+	}
+}

applications/ems_datahubs/opmysql.h

@@ -0,0 +1,27 @@
+#ifndef __MY_OPERATE_MYSQL__
+#define __MY_OPERATE_MYSQL__
+
+#include <string>
+#include <mysql_connection.h>
+
+
+class OpDatabase
+{
+protected:
+	OpDatabase();
+public:
+	~OpDatabase();
+	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");
+	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<sql::Connection> m_pDbConnection;
+	sql::Connection* m_pDbConnection;
+private:
+	static OpDatabase m_instance;
+};
+
+#endif